79 lines
2.9 KiB
Python
79 lines
2.9 KiB
Python
from typing import Any, cast
|
|
|
|
from redis.asyncio import Redis
|
|
|
|
from gapido_auth.domain.errors import OtpAttemptsExceeded, OtpExpired
|
|
from gapido_auth.domain.ports import OtpStore
|
|
|
|
|
|
class RedisOtpStore(OtpStore):
|
|
"""Redis adapter for OTP hashes, verification attempts, and request throttling."""
|
|
|
|
def __init__(self, redis: Redis) -> None:
|
|
"""Bind the store to an async Redis client."""
|
|
self._redis = redis
|
|
|
|
async def allow_request(self, key: str, limit: int, window_seconds: int) -> bool:
|
|
"""Increment a rate-limit counter and report whether it remains within limit."""
|
|
|
|
count = await self._redis.incr(key)
|
|
if count == 1:
|
|
await self._redis.expire(key, window_seconds)
|
|
return int(count) <= limit
|
|
|
|
async def store_otp(
|
|
self,
|
|
mobile: str,
|
|
purpose: str,
|
|
otp_hash: str,
|
|
ttl_seconds: int,
|
|
max_attempts: int,
|
|
) -> None:
|
|
"""Store a hashed OTP and reset its attempt counter with the same TTL window."""
|
|
|
|
key = self._otp_key(mobile, purpose)
|
|
attempts_key = self._attempts_key(mobile, purpose)
|
|
async with self._redis.pipeline(transaction=True) as pipe:
|
|
pipe.hset(key, mapping={"hash": otp_hash, "max_attempts": str(max_attempts)})
|
|
pipe.expire(key, ttl_seconds)
|
|
pipe.delete(attempts_key)
|
|
pipe.expire(attempts_key, ttl_seconds)
|
|
await pipe.execute()
|
|
|
|
async def verify_otp(self, mobile: str, purpose: str, candidate_hash: str) -> bool:
|
|
"""Compare a submitted OTP hash and delete state on success or exhausted attempts."""
|
|
|
|
key = self._otp_key(mobile, purpose)
|
|
attempts_key = self._attempts_key(mobile, purpose)
|
|
|
|
stored_hash = await cast(Any, self._redis.hget(key, "hash"))
|
|
if stored_hash is None:
|
|
raise OtpExpired("OTP expired or was not requested")
|
|
|
|
max_attempts_raw = await cast(Any, self._redis.hget(key, "max_attempts"))
|
|
max_attempts = int(max_attempts_raw or 5)
|
|
attempts = int(await self._redis.incr(attempts_key))
|
|
ttl = await self._redis.ttl(key)
|
|
if ttl > 0:
|
|
await self._redis.expire(attempts_key, ttl)
|
|
if attempts > max_attempts:
|
|
await self._redis.delete(key)
|
|
raise OtpAttemptsExceeded("OTP attempts exceeded")
|
|
|
|
expected = stored_hash.decode() if isinstance(stored_hash, bytes) else str(stored_hash)
|
|
if expected != candidate_hash:
|
|
return False
|
|
|
|
await self._redis.delete(key, attempts_key)
|
|
return True
|
|
|
|
@staticmethod
|
|
def _otp_key(mobile: str, purpose: str) -> str:
|
|
"""Return the Redis hash key for an OTP challenge."""
|
|
return f"otp:{mobile}:{purpose}"
|
|
|
|
@staticmethod
|
|
def _attempts_key(mobile: str, purpose: str) -> str:
|
|
"""Return the Redis counter key for OTP verification attempts."""
|
|
return f"otp-attempts:{mobile}:{purpose}"
|