added reset of TTL for login tokens when user is active
This commit is contained in:
@@ -16,6 +16,7 @@ class InMemoryTokenRepository(TokenRepositoryInterface):
|
|||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
# token -> (user_id, expiry_time)
|
# token -> (user_id, expiry_time)
|
||||||
self._tokens: dict[str, tuple[int, datetime]] = {}
|
self._tokens: dict[str, tuple[int, datetime]] = {}
|
||||||
|
self._default_ttl = 3600 # Store default TTL for sliding expiration
|
||||||
|
|
||||||
def store(self, token: str, user_id: int, ttl: int = 3600) -> None:
|
def store(self, token: str, user_id: int, ttl: int = 3600) -> None:
|
||||||
"""Store a token with TTL (time to live in seconds)."""
|
"""Store a token with TTL (time to live in seconds)."""
|
||||||
@@ -23,7 +24,10 @@ class InMemoryTokenRepository(TokenRepositoryInterface):
|
|||||||
self._tokens[token] = (user_id, expiry)
|
self._tokens[token] = (user_id, expiry)
|
||||||
|
|
||||||
def verify(self, token: str) -> int | None:
|
def verify(self, token: str) -> int | None:
|
||||||
"""Verify token and return user_id if valid, None otherwise."""
|
"""Verify token and return user_id if valid, None otherwise.
|
||||||
|
|
||||||
|
Implements sliding expiration: extends token lifetime on each verification.
|
||||||
|
"""
|
||||||
if token not in self._tokens:
|
if token not in self._tokens:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@@ -34,6 +38,10 @@ class InMemoryTokenRepository(TokenRepositoryInterface):
|
|||||||
del self._tokens[token]
|
del self._tokens[token]
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
# Sliding expiration: extend the token lifetime
|
||||||
|
new_expiry = datetime.now(UTC) + timedelta(seconds=self._default_ttl)
|
||||||
|
self._tokens[token] = (user_id, new_expiry)
|
||||||
|
|
||||||
return user_id
|
return user_id
|
||||||
|
|
||||||
def invalidate(self, token: str) -> None:
|
def invalidate(self, token: str) -> None:
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ class RedisTokenRepository(TokenRepositoryInterface):
|
|||||||
redis_client: Redis client instance from redis.from_url()
|
redis_client: Redis client instance from redis.from_url()
|
||||||
"""
|
"""
|
||||||
self.redis = redis_client
|
self.redis = redis_client
|
||||||
|
self.default_ttl = 3600 # Store default TTL for sliding expiration
|
||||||
|
|
||||||
def store(self, token: str, user_id: int, ttl: int = 3600) -> None:
|
def store(self, token: str, user_id: int, ttl: int = 3600) -> None:
|
||||||
"""Store a token with TTL (time to live in seconds)."""
|
"""Store a token with TTL (time to live in seconds)."""
|
||||||
@@ -28,10 +29,15 @@ class RedisTokenRepository(TokenRepositoryInterface):
|
|||||||
self.redis.setex(key, ttl, str(user_id))
|
self.redis.setex(key, ttl, str(user_id))
|
||||||
|
|
||||||
def verify(self, token: str) -> int | None:
|
def verify(self, token: str) -> int | None:
|
||||||
"""Verify token and return user_id if valid, None otherwise."""
|
"""Verify token and return user_id if valid, None otherwise.
|
||||||
|
|
||||||
|
Implements sliding expiration: extends token lifetime on verification.
|
||||||
|
"""
|
||||||
key = f"token:{token}"
|
key = f"token:{token}"
|
||||||
user_id_str = self.redis.get(key)
|
user_id_str = self.redis.get(key)
|
||||||
if user_id_str:
|
if user_id_str:
|
||||||
|
# Sliding expiration: extend the token lifetime
|
||||||
|
self.redis.expire(key, self.default_ttl)
|
||||||
return int(user_id_str)
|
return int(user_id_str)
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user