feat(grpc): add auth service transport

src/gapido_auth/transport/grpc/auth_servicer.py

@@ -0,0 +1,122 @@
+# mypy: disable-error-code="attr-defined"
+import logging
+from collections.abc import Sequence
+
+import grpc
+
+from gapido_auth.application.auth_service import AuthService
+from gapido_auth.domain.entities import Role, TokenPair, User
+from gapido_auth.domain.errors import (
+    AppError,
+    AuthenticationError,
+    InactiveUser,
+    InvalidOtp,
+    OtpAttemptsExceeded,
+    OtpExpired,
+    PermissionDenied,
+    RateLimitExceeded,
+    ValidationError,
+)
+from gapido_auth.generated import auth_pb2, auth_pb2_grpc
+
+logger = logging.getLogger(__name__)
+
+
+class AuthGrpcServicer(auth_pb2_grpc.AuthServiceServicer):
+    def __init__(self, auth_service: AuthService) -> None:
+        self._auth_service = auth_service
+
+    async def RequestOtp(self, request, context):  # type: ignore[no-untyped-def]
+        try:
+            await self._auth_service.request_otp(
+                mobile=request.mobile,
+                purpose=request.purpose or "login",
+                client_key=context.peer(),
+            )
+            return auth_pb2.RequestOtpResponse(accepted=True)
+        except AppError as exc:
+            await _abort_for_app_error(context, exc)
+
+    async def VerifyOtp(self, request, context):  # type: ignore[no-untyped-def]
+        try:
+            token_pair = await self._auth_service.verify_otp(
+                mobile=request.mobile,
+                code=request.code,
+                purpose=request.purpose or "login",
+            )
+            return _token_response(token_pair)
+        except AppError as exc:
+            await _abort_for_app_error(context, exc)
+
+    async def RefreshToken(self, request, context):  # type: ignore[no-untyped-def]
+        try:
+            token_pair = await self._auth_service.refresh_token(request.refresh_token)
+            return _token_response(token_pair)
+        except AppError as exc:
+            await _abort_for_app_error(context, exc)
+
+    async def RevokeRefreshToken(self, request, context):  # type: ignore[no-untyped-def]
+        try:
+            token = _extract_bearer_token(context.invocation_metadata())
+            await self._auth_service.get_authenticated_user(token)
+            await self._auth_service.revoke_refresh_token(request.refresh_token)
+            return auth_pb2.RevokeRefreshTokenResponse(revoked=True)
+        except AppError as exc:
+            await _abort_for_app_error(context, exc)
+
+    async def PublicPing(self, request, context):  # type: ignore[no-untyped-def]
+        return auth_pb2.PingResponse(message="public ok")
+
+    async def UserOnly(self, request, context):  # type: ignore[no-untyped-def]
+        try:
+            user = await self._auth_service.get_authenticated_user(
+                _extract_bearer_token(context.invocation_metadata())
+            )
+            return _protected_response(user, "authenticated user ok")
+        except AppError as exc:
+            await _abort_for_app_error(context, exc)
+
+    async def AdminOnly(self, request, context):  # type: ignore[no-untyped-def]
+        try:
+            user = await self._auth_service.require_role(
+                _extract_bearer_token(context.invocation_metadata()), Role.ADMIN
+            )
+            return _protected_response(user, "admin ok")
+        except AppError as exc:
+            await _abort_for_app_error(context, exc)
+
+
+def _token_response(token_pair: TokenPair):  # type: ignore[no-untyped-def]
+    return auth_pb2.TokenResponse(
+        access_token=token_pair.access_token,
+        refresh_token=token_pair.refresh_token,
+        token_type=token_pair.token_type,
+        expires_in=token_pair.expires_in,
+        role=token_pair.role.value,
+    )
+
+
+def _protected_response(user: User, message: str):  # type: ignore[no-untyped-def]
+    return auth_pb2.ProtectedResponse(user_id=user.id, role=user.role.value, message=message)
+
+
+def _extract_bearer_token(metadata: Sequence[tuple[str, str]]) -> str:
+    for key, value in metadata:
+        if key.lower() == "authorization" and value.startswith("Bearer "):
+            return value.removeprefix("Bearer ").strip()
+    raise AuthenticationError("missing bearer token")
+
+
+async def _abort_for_app_error(context: grpc.aio.ServicerContext, exc: AppError) -> None:
+    if isinstance(exc, AuthenticationError | InvalidOtp | OtpExpired | OtpAttemptsExceeded):
+        await context.abort(grpc.StatusCode.UNAUTHENTICATED, str(exc))
+    if isinstance(exc, InactiveUser):
+        await context.abort(grpc.StatusCode.PERMISSION_DENIED, str(exc))
+    if isinstance(exc, PermissionDenied):
+        await context.abort(grpc.StatusCode.PERMISSION_DENIED, str(exc))
+    if isinstance(exc, RateLimitExceeded):
+        await context.abort(grpc.StatusCode.RESOURCE_EXHAUSTED, str(exc))
+    if isinstance(exc, ValidationError):
+        await context.abort(grpc.StatusCode.INVALID_ARGUMENT, str(exc))
+    logger.exception("Unhandled application error")
+    await context.abort(grpc.StatusCode.INTERNAL, "internal error")