89 lines
3.0 KiB
Python
89 lines
3.0 KiB
Python
# mypy: disable-error-code="attr-defined,no-untyped-call"
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import grpc
|
|
|
|
from gapido_auth.generated import auth_pb2, auth_pb2_grpc
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DemoTokenResponse:
|
|
access_token: str
|
|
refresh_token: str
|
|
token_type: str
|
|
expires_in: int
|
|
role: str
|
|
|
|
|
|
class AuthGrpcClient:
|
|
def __init__(self, target: str) -> None:
|
|
self._channel = grpc.aio.insecure_channel(target)
|
|
self._stub = auth_pb2_grpc.AuthServiceStub(self._channel)
|
|
|
|
async def close(self) -> None:
|
|
await self._channel.close()
|
|
|
|
async def request_otp(self, mobile: str, purpose: str) -> dict[str, bool]:
|
|
response = await self._stub.RequestOtp(
|
|
auth_pb2.RequestOtpRequest(mobile=mobile, purpose=purpose)
|
|
)
|
|
return {"accepted": bool(response.accepted)}
|
|
|
|
async def verify_otp(self, mobile: str, code: str, purpose: str) -> DemoTokenResponse:
|
|
response = await self._stub.VerifyOtp(
|
|
auth_pb2.VerifyOtpRequest(mobile=mobile, code=code, purpose=purpose)
|
|
)
|
|
return _token_response(response)
|
|
|
|
async def refresh_token(self, refresh_token: str) -> DemoTokenResponse:
|
|
response = await self._stub.RefreshToken(
|
|
auth_pb2.RefreshTokenRequest(refresh_token=refresh_token)
|
|
)
|
|
return _token_response(response)
|
|
|
|
async def revoke_refresh_token(self, access_token: str, refresh_token: str) -> dict[str, bool]:
|
|
response = await self._stub.RevokeRefreshToken(
|
|
auth_pb2.RevokeRefreshTokenRequest(refresh_token=refresh_token),
|
|
metadata=_auth_metadata(access_token),
|
|
)
|
|
return {"revoked": bool(response.revoked)}
|
|
|
|
async def public_ping(self) -> dict[str, str]:
|
|
response = await self._stub.PublicPing(auth_pb2.PingRequest())
|
|
return {"message": str(response.message)}
|
|
|
|
async def user_only(self, access_token: str) -> dict[str, str]:
|
|
response = await self._stub.UserOnly(
|
|
auth_pb2.ProtectedRequest(), metadata=_auth_metadata(access_token)
|
|
)
|
|
return _protected_response(response)
|
|
|
|
async def admin_only(self, access_token: str) -> dict[str, str]:
|
|
response = await self._stub.AdminOnly(
|
|
auth_pb2.ProtectedRequest(), metadata=_auth_metadata(access_token)
|
|
)
|
|
return _protected_response(response)
|
|
|
|
|
|
def _auth_metadata(access_token: str) -> tuple[tuple[str, str], ...]:
|
|
return (("authorization", f"Bearer {access_token}"),)
|
|
|
|
|
|
def _token_response(response: Any) -> DemoTokenResponse:
|
|
return DemoTokenResponse(
|
|
access_token=str(response.access_token),
|
|
refresh_token=str(response.refresh_token),
|
|
token_type=str(response.token_type),
|
|
expires_in=int(response.expires_in),
|
|
role=str(response.role),
|
|
)
|
|
|
|
|
|
def _protected_response(response: Any) -> dict[str, str]:
|
|
return {
|
|
"user_id": str(response.user_id),
|
|
"role": str(response.role),
|
|
"message": str(response.message),
|
|
}
|