From 61f7b5c3225ee8d1f05ff57e68ff210dbab81578 Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Tue, 14 Jul 2026 09:32:58 +0330 Subject: [PATCH] feat(infra): add persistence and SMS adapters --- src/gapido_auth/infrastructure/__init__.py | 2 + src/gapido_auth/infrastructure/container.py | 60 ++++++++ .../infrastructure/kavenegar_client.py | 45 ++++++ .../infrastructure/mongo_repositories.py | 135 ++++++++++++++++++ src/gapido_auth/infrastructure/rabbitmq.py | 53 +++++++ .../infrastructure/redis_otp_store.py | 67 +++++++++ src/gapido_auth/infrastructure/worker.py | 78 ++++++++++ 7 files changed, 440 insertions(+) create mode 100644 src/gapido_auth/infrastructure/__init__.py create mode 100644 src/gapido_auth/infrastructure/container.py create mode 100644 src/gapido_auth/infrastructure/kavenegar_client.py create mode 100644 src/gapido_auth/infrastructure/mongo_repositories.py create mode 100644 src/gapido_auth/infrastructure/rabbitmq.py create mode 100644 src/gapido_auth/infrastructure/redis_otp_store.py create mode 100644 src/gapido_auth/infrastructure/worker.py diff --git a/src/gapido_auth/infrastructure/__init__.py b/src/gapido_auth/infrastructure/__init__.py new file mode 100644 index 0000000..aad11f3 --- /dev/null +++ b/src/gapido_auth/infrastructure/__init__.py @@ -0,0 +1,2 @@ +"""Infrastructure adapters.""" + diff --git a/src/gapido_auth/infrastructure/container.py b/src/gapido_auth/infrastructure/container.py new file mode 100644 index 0000000..ceb4f8d --- /dev/null +++ b/src/gapido_auth/infrastructure/container.py @@ -0,0 +1,60 @@ +from dataclasses import dataclass +from typing import Any + +from aio_pika.abc import AbstractChannel +from motor.motor_asyncio import AsyncIOMotorClient +from redis.asyncio import Redis + +from gapido_auth.application.auth_service import AuthConfig, AuthService +from gapido_auth.application.security import JwtTokenCodec +from gapido_auth.config import Settings +from gapido_auth.infrastructure.mongo_repositories import ( + MongoRefreshSessionRepository, + MongoUserRepository, +) +from gapido_auth.infrastructure.rabbitmq import RabbitMqSmsPublisher +from gapido_auth.infrastructure.redis_otp_store import RedisOtpStore + + +@dataclass(slots=True) +class AppContainer: + auth_service: AuthService + mongo_client: AsyncIOMotorClient[Any] + redis: Redis + + +async def build_auth_service(settings: Settings, rabbitmq_channel: AbstractChannel) -> AppContainer: + mongo_client: AsyncIOMotorClient[Any] = AsyncIOMotorClient(settings.mongo_uri) + db = mongo_client[settings.mongo_db_name] + users = MongoUserRepository(db) + refresh_sessions = MongoRefreshSessionRepository(db) + await users.create_indexes() + await refresh_sessions.create_indexes() + await users.ensure_admin_user(settings.admin_mobile) + + redis = Redis.from_url(settings.redis_url, decode_responses=False) + otp_store = RedisOtpStore(redis) + sms_publisher = RabbitMqSmsPublisher(rabbitmq_channel) + token_codec = JwtTokenCodec( + secret_key=settings.jwt_secret_key, + issuer=settings.jwt_issuer, + access_ttl_seconds=settings.access_token_ttl_seconds, + ) + auth_service = AuthService( + users=users, + refresh_sessions=refresh_sessions, + otp_store=otp_store, + sms_publisher=sms_publisher, + token_codec=token_codec, + config=AuthConfig( + otp_secret=settings.jwt_secret_key, + otp_ttl_seconds=settings.otp_ttl_seconds, + otp_max_attempts=settings.otp_max_attempts, + otp_request_limit=settings.otp_request_limit, + otp_request_window_seconds=settings.otp_request_window_seconds, + refresh_token_ttl_seconds=settings.refresh_token_ttl_seconds, + sms_template=settings.kavenegar_login_template, + ), + ) + return AppContainer(auth_service=auth_service, mongo_client=mongo_client, redis=redis) + diff --git a/src/gapido_auth/infrastructure/kavenegar_client.py b/src/gapido_auth/infrastructure/kavenegar_client.py new file mode 100644 index 0000000..066f9b7 --- /dev/null +++ b/src/gapido_auth/infrastructure/kavenegar_client.py @@ -0,0 +1,45 @@ +import logging +from typing import Any + +import httpx + +from gapido_auth.domain.errors import ExternalServiceError +from gapido_auth.domain.ports import SmsClient + +logger = logging.getLogger(__name__) + + +class KavenegarSmsClient(SmsClient): + def __init__( + self, + api_key: str, + timeout_seconds: float = 10.0, + transport: httpx.AsyncBaseTransport | None = None, + ) -> None: + self._api_key = api_key + self._timeout_seconds = timeout_seconds + self._transport = transport + + async def send_otp(self, mobile: str, code: str, template: str) -> None: + url = f"https://api.kavenegar.com/v1/{self._api_key}/verify/lookup.json" + payload = {"receptor": mobile, "template": template, "token": code, "type": "sms"} + try: + async with httpx.AsyncClient( + timeout=self._timeout_seconds, transport=self._transport + ) as client: + response = await client.post(url, data=payload) + except httpx.HTTPError as exc: + logger.warning("Kavenegar network error for mobile=%s", mobile, exc_info=True) + raise ExternalServiceError("Kavenegar network error") from exc + + if response.status_code != 200: + logger.warning("Kavenegar HTTP error status=%s mobile=%s", response.status_code, mobile) + raise ExternalServiceError("Kavenegar HTTP error") + + data: dict[str, Any] = response.json() + status = data.get("return", {}).get("status") + if status != 200: + logger.warning("Kavenegar API error status=%s mobile=%s", status, mobile) + raise ExternalServiceError("Kavenegar API error") + + logger.info("OTP SMS sent successfully to mobile=%s", mobile) diff --git a/src/gapido_auth/infrastructure/mongo_repositories.py b/src/gapido_auth/infrastructure/mongo_repositories.py new file mode 100644 index 0000000..3e7b47a --- /dev/null +++ b/src/gapido_auth/infrastructure/mongo_repositories.py @@ -0,0 +1,135 @@ +from datetime import UTC, datetime +from typing import Any + +from bson import ObjectId +from motor.motor_asyncio import AsyncIOMotorDatabase +from pymongo import ASCENDING + +from gapido_auth.domain.entities import RefreshSession, Role, User +from gapido_auth.domain.ports import RefreshSessionRepository, UserRepository + + +def _now() -> datetime: + return datetime.now(UTC) + + +def _user_from_doc(doc: dict[str, Any]) -> User: + return User( + id=str(doc["_id"]), + mobile=str(doc["mobile"]), + role=Role(str(doc["role"])), + is_active=bool(doc["is_active"]), + created_at=doc["created_at"], + updated_at=doc["updated_at"], + ) + + +def _session_from_doc(doc: dict[str, Any]) -> RefreshSession: + return RefreshSession( + id=str(doc["_id"]), + user_id=str(doc["user_id"]), + token_hash=str(doc["token_hash"]), + expires_at=doc["expires_at"], + revoked_at=doc.get("revoked_at"), + replaced_by_hash=doc.get("replaced_by_hash"), + created_at=doc["created_at"], + ) + + +class MongoUserRepository(UserRepository): + def __init__(self, db: AsyncIOMotorDatabase[Any]) -> None: + self._collection = db.users + + async def create_indexes(self) -> None: + await self._collection.create_index([("mobile", ASCENDING)], unique=True) + await self._collection.create_index([("role", ASCENDING)]) + + async def get_by_id(self, user_id: str) -> User | None: + if not ObjectId.is_valid(user_id): + return None + doc = await self._collection.find_one({"_id": ObjectId(user_id)}) + return _user_from_doc(doc) if doc else None + + async def get_by_mobile(self, mobile: str) -> User | None: + doc = await self._collection.find_one({"mobile": mobile}) + return _user_from_doc(doc) if doc else None + + async def get_or_create_user(self, mobile: str, role: Role = Role.USER) -> User: + now = _now() + await self._collection.update_one( + {"mobile": mobile}, + { + "$setOnInsert": { + "mobile": mobile, + "role": role.value, + "is_active": True, + "created_at": now, + }, + "$set": {"updated_at": now}, + }, + upsert=True, + ) + user = await self.get_by_mobile(mobile) + if user is None: + raise RuntimeError("failed to create user") + return user + + async def ensure_admin_user(self, mobile: str) -> User: + now = _now() + await self._collection.update_one( + {"mobile": mobile}, + { + "$set": {"role": Role.ADMIN.value, "is_active": True, "updated_at": now}, + "$setOnInsert": {"mobile": mobile, "created_at": now}, + }, + upsert=True, + ) + user = await self.get_by_mobile(mobile) + if user is None: + raise RuntimeError("failed to seed admin user") + return user + + +class MongoRefreshSessionRepository(RefreshSessionRepository): + def __init__(self, db: AsyncIOMotorDatabase[Any]) -> None: + self._collection = db.refresh_sessions + + async def create_indexes(self) -> None: + await self._collection.create_index([("token_hash", ASCENDING)], unique=True) + await self._collection.create_index([("user_id", ASCENDING), ("expires_at", ASCENDING)]) + await self._collection.create_index([("revoked_at", ASCENDING)]) + + async def create(self, user_id: str, token_hash: str, expires_at: datetime) -> RefreshSession: + now = _now() + result = await self._collection.insert_one( + { + "user_id": user_id, + "token_hash": token_hash, + "expires_at": expires_at, + "revoked_at": None, + "replaced_by_hash": None, + "created_at": now, + } + ) + doc = await self._collection.find_one({"_id": result.inserted_id}) + if doc is None: + raise RuntimeError("failed to create refresh session") + return _session_from_doc(doc) + + async def get_active_by_hash(self, token_hash: str, now: datetime) -> RefreshSession | None: + doc = await self._collection.find_one( + {"token_hash": token_hash, "revoked_at": None, "expires_at": {"$gt": now}} + ) + return _session_from_doc(doc) if doc else None + + async def revoke( + self, token_hash: str, now: datetime, replaced_by_hash: str | None = None + ) -> None: + update: dict[str, Any] = {"revoked_at": now} + if replaced_by_hash is not None: + update["replaced_by_hash"] = replaced_by_hash + await self._collection.update_one( + {"token_hash": token_hash, "revoked_at": None}, + {"$set": update}, + ) + diff --git a/src/gapido_auth/infrastructure/rabbitmq.py b/src/gapido_auth/infrastructure/rabbitmq.py new file mode 100644 index 0000000..eb36076 --- /dev/null +++ b/src/gapido_auth/infrastructure/rabbitmq.py @@ -0,0 +1,53 @@ +import json +from dataclasses import asdict +from typing import Any + +import aio_pika +from aio_pika.abc import AbstractChannel, AbstractQueue, AbstractRobustConnection, DeliveryMode + +from gapido_auth.domain.entities import SmsJob +from gapido_auth.domain.ports import SmsPublisher + +SMS_EXCHANGE = "gapido.sms" +SMS_QUEUE = "gapido.sms.otp" +SMS_DLQ = "gapido.sms.otp.dlq" +SMS_ROUTING_KEY = "otp" +SMS_DLX = "gapido.sms.dlx" + + +async def connect_robust(url: str) -> AbstractRobustConnection: + return await aio_pika.connect_robust(url) + + +async def declare_sms_topology(channel: AbstractChannel) -> tuple[Any, AbstractQueue, Any]: + exchange = await channel.declare_exchange( + SMS_EXCHANGE, aio_pika.ExchangeType.DIRECT, durable=True + ) + dlx = await channel.declare_exchange(SMS_DLX, aio_pika.ExchangeType.DIRECT, durable=True) + queue = await channel.declare_queue( + SMS_QUEUE, + durable=True, + arguments={"x-dead-letter-exchange": SMS_DLX, "x-dead-letter-routing-key": SMS_ROUTING_KEY}, + ) + dlq = await channel.declare_queue(SMS_DLQ, durable=True) + await queue.bind(exchange, routing_key=SMS_ROUTING_KEY) + await dlq.bind(dlx, routing_key=SMS_ROUTING_KEY) + return exchange, queue, dlx + + +class RabbitMqSmsPublisher(SmsPublisher): + def __init__(self, channel: AbstractChannel) -> None: + self._channel = channel + self._exchange: Any | None = None + + async def publish(self, job: SmsJob) -> None: + if self._exchange is None: + self._exchange, _, _ = await declare_sms_topology(self._channel) + + body = json.dumps(asdict(job)).encode() + message = aio_pika.Message( + body=body, + content_type="application/json", + delivery_mode=DeliveryMode.PERSISTENT, + ) + await self._exchange.publish(message, routing_key=SMS_ROUTING_KEY) diff --git a/src/gapido_auth/infrastructure/redis_otp_store.py b/src/gapido_auth/infrastructure/redis_otp_store.py new file mode 100644 index 0000000..67378c7 --- /dev/null +++ b/src/gapido_auth/infrastructure/redis_otp_store.py @@ -0,0 +1,67 @@ +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): + def __init__(self, redis: Redis) -> None: + self._redis = redis + + async def allow_request(self, key: str, limit: int, window_seconds: int) -> bool: + 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: + 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: + 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 f"otp:{mobile}:{purpose}" + + @staticmethod + def _attempts_key(mobile: str, purpose: str) -> str: + return f"otp-attempts:{mobile}:{purpose}" diff --git a/src/gapido_auth/infrastructure/worker.py b/src/gapido_auth/infrastructure/worker.py new file mode 100644 index 0000000..20dfe26 --- /dev/null +++ b/src/gapido_auth/infrastructure/worker.py @@ -0,0 +1,78 @@ +import asyncio +import json +import logging +from typing import Any, cast + +import aio_pika +from aio_pika.abc import AbstractIncomingMessage + +from gapido_auth.config import get_settings +from gapido_auth.domain.entities import SmsJob +from gapido_auth.infrastructure.kavenegar_client import KavenegarSmsClient +from gapido_auth.infrastructure.rabbitmq import ( + SMS_ROUTING_KEY, + connect_robust, + declare_sms_topology, +) + +logger = logging.getLogger(__name__) +MAX_RETRIES = 3 + + +async def handle_message( + message: AbstractIncomingMessage, + client: KavenegarSmsClient, + exchange: Any, + dlx: Any, +) -> None: + async with message.process(ignore_processed=True, requeue=False): + payload = json.loads(message.body.decode()) + job = SmsJob(**payload) + retry_count = int(cast(int | str, (message.headers or {}).get("x-retry-count", 0))) + + try: + await client.send_otp(job.mobile, job.code, job.template) + except Exception: + if retry_count >= MAX_RETRIES: + logger.exception("SMS job failed permanently for mobile=%s", job.mobile) + await dlx.publish( + aio_pika.Message( + body=message.body, + content_type="application/json", + headers={"x-retry-count": retry_count}, + delivery_mode=aio_pika.DeliveryMode.PERSISTENT, + ), + routing_key=SMS_ROUTING_KEY, + ) + return + + logger.warning( + "SMS job failed for mobile=%s retry=%s", job.mobile, retry_count + 1, exc_info=True + ) + await exchange.publish( + aio_pika.Message( + body=message.body, + content_type="application/json", + headers={"x-retry-count": retry_count + 1}, + delivery_mode=aio_pika.DeliveryMode.PERSISTENT, + ), + routing_key=SMS_ROUTING_KEY, + ) + + +async def main() -> None: + logging.basicConfig(level=logging.INFO) + settings = get_settings() + connection = await connect_robust(settings.rabbitmq_url) + async with connection: + channel = await connection.channel() + await channel.set_qos(prefetch_count=20) + exchange, queue, dlx = await declare_sms_topology(channel) + client = KavenegarSmsClient(settings.kavenegar_api_key) + await queue.consume(lambda message: handle_message(message, client, exchange, dlx)) + logger.info("SMS worker started") + await asyncio.Future() + + +if __name__ == "__main__": + asyncio.run(main())