Compare commits

..

6 Commits

47 changed files with 1321 additions and 36 deletions

View File

@@ -1,4 +1,5 @@
APP_ENV=local APP_ENV=local
MONGO_IMAGE=mongo:4.4.29-focal
MONGO_URI=mongodb://localhost:27017 MONGO_URI=mongodb://localhost:27017
MONGO_DB_NAME=gapido_auth MONGO_DB_NAME=gapido_auth
REDIS_URL=redis://localhost:6379/0 REDIS_URL=redis://localhost:6379/0

View File

@@ -1,4 +1,5 @@
APP_ENV=production APP_ENV=production
MONGO_IMAGE=mongo:4.4.29-focal
MONGO_DB_NAME=gapido_auth MONGO_DB_NAME=gapido_auth
JWT_SECRET_KEY=replace-with-a-long-random-production-secret JWT_SECRET_KEY=replace-with-a-long-random-production-secret
JWT_ISSUER=gapido-auth JWT_ISSUER=gapido-auth
@@ -21,4 +22,3 @@ DEMO_ENABLE_DEBUG_OTP=false
DEMO_DEBUG_SMS_TTL_SECONDS=300 DEMO_DEBUG_SMS_TTL_SECONDS=300
GRPC_HOST=0.0.0.0 GRPC_HOST=0.0.0.0
GRPC_PORT=50051 GRPC_PORT=50051

View File

@@ -37,6 +37,8 @@ Before running production, set real values in `.env.production`:
Only ports `80` and `443` are published in the production Compose overlay. MongoDB, Redis, RabbitMQ, gRPC, and the FastAPI demo service stay private on the Docker network. Only ports `80` and `443` are published in the production Compose overlay. MongoDB, Redis, RabbitMQ, gRPC, and the FastAPI demo service stay private on the Docker network.
The Compose default uses `MONGO_IMAGE=mongo:4.4.29-focal` because MongoDB 5.0+ requires AVX CPU support. On newer hosts you can override it with a newer MongoDB image, for example `MONGO_IMAGE=mongo:7`.
## SMS Provider ## SMS Provider
The SMS integration uses the Strategy pattern behind the `SmsClient` port. Select the provider with: The SMS integration uses the Strategy pattern behind the `SmsClient` port. Select the provider with:
@@ -73,6 +75,15 @@ pytest
The production Docker image installs runtime dependencies only. Development tools are installed locally through `.[dev]`. The production Docker image installs runtime dependencies only. Development tools are installed locally through `.[dev]`.
## Documentation
Project documentation is available in [`docs/`](docs/README.md):
- [Architecture](docs/architecture.md)
- [Implementation decisions](docs/implementation-decisions.md)
- [Architecture decision records](docs/adr/README.md)
- [Presentation slides](docs/slides/index.html)
## gRPC Methods ## gRPC Methods
- `RequestOtp`: public; creates a short-lived OTP and publishes an SMS job. - `RequestOtp`: public; creates a short-lived OTP and publishes an SMS job.

View File

@@ -1,23 +1,35 @@
services: services:
auth-service: auth-service:
restart: unless-stopped
environment: environment:
SMS_PROVIDER: ${SMS_PROVIDER:?set SMS_PROVIDER in production env} SMS_PROVIDER: ${SMS_PROVIDER:?set SMS_PROVIDER in production env}
JWT_SECRET_KEY: ${JWT_SECRET_KEY:?set JWT_SECRET_KEY in production env} JWT_SECRET_KEY: ${JWT_SECRET_KEY:?set JWT_SECRET_KEY in production env}
ADMIN_MOBILE: ${ADMIN_MOBILE:?set ADMIN_MOBILE in production env} ADMIN_MOBILE: ${ADMIN_MOBILE:?set ADMIN_MOBILE in production env}
sms-worker: sms-worker:
restart: unless-stopped
environment: environment:
SMS_PROVIDER: ${SMS_PROVIDER:?set SMS_PROVIDER in production env} SMS_PROVIDER: ${SMS_PROVIDER:?set SMS_PROVIDER in production env}
KAVENEGAR_API_KEY: ${KAVENEGAR_API_KEY:-} KAVENEGAR_API_KEY: ${KAVENEGAR_API_KEY:-}
SMS_IR_API_KEY: ${SMS_IR_API_KEY:-} SMS_IR_API_KEY: ${SMS_IR_API_KEY:-}
demo-app: demo-app:
restart: unless-stopped
environment: environment:
DEMO_ENABLE_DEBUG_OTP: "false" DEMO_ENABLE_DEBUG_OTP: "false"
depends_on: depends_on:
auth-service: auth-service:
condition: service_healthy condition: service_healthy
mongo:
restart: unless-stopped
redis:
restart: unless-stopped
rabbitmq:
restart: unless-stopped
caddy: caddy:
image: caddy:2.8-alpine image: caddy:2.8-alpine
restart: unless-stopped restart: unless-stopped
@@ -41,4 +53,3 @@ services:
volumes: volumes:
caddy-data: caddy-data:
caddy-config: caddy-config:

View File

@@ -35,9 +35,9 @@ services:
start_period: 20s start_period: 20s
depends_on: depends_on:
mongo: mongo:
condition: service_started condition: service_healthy
redis: redis:
condition: service_started condition: service_healthy
rabbitmq: rabbitmq:
condition: service_healthy condition: service_healthy
@@ -54,6 +54,8 @@ services:
depends_on: depends_on:
rabbitmq: rabbitmq:
condition: service_healthy condition: service_healthy
redis:
condition: service_healthy
demo-app: demo-app:
build: . build: .
@@ -72,15 +74,26 @@ services:
auth-service: auth-service:
condition: service_healthy condition: service_healthy
redis: redis:
condition: service_started condition: service_healthy
mongo: mongo:
image: mongo:7 image: ${MONGO_IMAGE:-mongo:4.4.29-focal}
volumes: volumes:
- mongo-data:/data/db - mongo-data:/data/db
healthcheck:
test: ["CMD-SHELL", "mongosh --quiet --eval \"db.adminCommand('ping').ok\" || mongo --quiet --eval \"db.adminCommand('ping').ok\""]
interval: 5s
timeout: 5s
retries: 20
start_period: 10s
redis: redis:
image: redis:7-alpine image: redis:7-alpine
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 5s
timeout: 5s
retries: 20
rabbitmq: rabbitmq:
image: rabbitmq:3.13-management-alpine image: rabbitmq:3.13-management-alpine

21
docs/README.md Normal file
View File

@@ -0,0 +1,21 @@
# Documentation
This directory contains the design notes, diagrams, and presentation material for the Gapido backend code challenge.
## Contents
- [Architecture](architecture.md): service boundaries, flows, deployment modes, and runtime topology.
- [Implementation Decisions](implementation-decisions.md): key decisions and tradeoffs made during development.
- [ADR Index](adr/README.md): focused architectural decision records.
- [Slides](slides/index.html): static HTML presentation deck.
- [Diagrams](assets/diagrams/): generated PNG diagrams used by docs and slides.
## Regenerate Diagrams
Graphviz must be installed on the machine because the Python `diagrams` package renders through Graphviz.
```bash
pip install -e ".[dev]"
python docs/diagrams/generate.py
```

View File

@@ -0,0 +1,18 @@
# ADR 001: Use Python gRPC for Auth
## Status
Accepted.
## Context
The assessment requires Python with gRPC and focuses on microservice communication.
## Decision
Implement the authentication boundary as an async `grpc.aio` service with protobuf contracts.
## Consequences
Service-to-service calls are strongly typed and efficient. Browser interaction needs a separate demo client service because native browser gRPC is not practical without grpc-web or a proxy.

View File

@@ -0,0 +1,18 @@
# ADR 002: Store OTP State In Redis
## Status
Accepted.
## Context
OTP state is temporary, security-sensitive, and must expire automatically.
## Decision
Store only HMAC hashes of OTP codes in Redis, with TTLs, attempt counters, and request rate limits.
## Consequences
OTP verification is fast and self-expiring. Plaintext OTPs are not persisted. The local debug provider writes a separate development-only key for demos.

View File

@@ -0,0 +1,18 @@
# ADR 003: Dispatch SMS Through RabbitMQ
## Status
Accepted.
## Context
SMS providers are external services and can fail or respond slowly.
## Decision
Publish OTP SMS jobs to RabbitMQ and process them in a dedicated worker with manual acknowledgements, retries, and a dead-letter queue.
## Consequences
The auth service remains responsive. SMS delivery is isolated and easier to retry, monitor, and replace.

View File

@@ -0,0 +1,18 @@
# ADR 004: Select SMS Providers With Strategy
## Status
Accepted.
## Context
The project supports Kavenegar, SMS.ir, and local debug delivery.
## Decision
Expose a stable `SmsClient` port and choose the concrete provider through a small factory based on `SMS_PROVIDER`.
## Consequences
Adding a new SMS provider does not change the auth use case or worker flow. Tests can mock providers without network access.

View File

@@ -0,0 +1,18 @@
# ADR 005: Add FastAPI Demo Client Service
## Status
Accepted.
## Context
The gRPC service needs a simple browser demo for interviews and reviewers.
## Decision
Add `gapido_demo`, a small FastAPI backend-for-frontend that serves static UI and calls the auth service over gRPC.
## Consequences
The demo shows how a separate microservice consumes auth. It remains separate from the auth service package and can be exposed safely through Caddy.

View File

@@ -0,0 +1,18 @@
# ADR 006: Use Caddy For Production Ingress
## Status
Accepted.
## Context
Production should expose only the demo UI under `gapido.amiirkhl.ir` with HTTPS.
## Decision
Use Caddy as the reverse proxy and TLS terminator. Route public traffic to `demo-app:8080`.
## Consequences
TLS automation is simple. Internal services remain private on Docker networking.

View File

@@ -0,0 +1,18 @@
# ADR 007: Split Local And Production Compose Overlays
## Status
Accepted.
## Context
Local development needs direct access to service ports, while production must expose only Caddy.
## Decision
Keep `docker-compose.yml` private by default, add `docker-compose.local.yml` for local ports, and add `docker-compose.prod.yml` for Caddy and production environment rules.
## Consequences
The same services run locally and on a server, but exposure is controlled by the chosen Compose overlay.

10
docs/adr/README.md Normal file
View File

@@ -0,0 +1,10 @@
# Architecture Decision Records
- [ADR 001: Use Python gRPC for auth](001-use-python-grpc.md)
- [ADR 002: Store OTP state in Redis](002-redis-otp-state.md)
- [ADR 003: Dispatch SMS through RabbitMQ](003-rabbitmq-sms-worker.md)
- [ADR 004: Select SMS providers with Strategy](004-sms-provider-strategy.md)
- [ADR 005: Add FastAPI demo client service](005-fastapi-demo-client.md)
- [ADR 006: Use Caddy for production ingress](006-caddy-production-ingress.md)
- [ADR 007: Split local and production Compose overlays](007-compose-overlays.md)

40
docs/architecture.md Normal file
View File

@@ -0,0 +1,40 @@
# Architecture
The project demonstrates a production-shaped authentication microservice built around gRPC, OTP login, asynchronous SMS dispatch, and a small demo client service.
## System View
![Service architecture](assets/diagrams/service_architecture.png)
The browser talks only to `demo-app`. The demo service acts as a backend-for-frontend and calls the gRPC `auth-service`. Auth data is split by responsibility: MongoDB persists users and refresh sessions, Redis stores short-lived OTP state and rate-limit counters, and RabbitMQ decouples OTP generation from SMS delivery.
## Clean Architecture
![Clean architecture layers](assets/diagrams/clean_architecture.png)
The domain layer owns roles, users, refresh sessions, and ports. The application layer coordinates OTP, tokens, and access control. Infrastructure adapters implement MongoDB, Redis, RabbitMQ, and SMS providers. Transports translate gRPC and demo HTTP requests into application use cases.
## OTP Login Flow
![OTP login sequence](assets/diagrams/otp_login_flow.png)
OTP codes are generated with `secrets`, stored only as HMAC hashes in Redis, and sent through RabbitMQ. Verification compares hashes and deletes successful OTP state before issuing an access token and opaque refresh token.
## Refresh Rotation
![Refresh token rotation](assets/diagrams/refresh_rotation_flow.png)
Refresh tokens are opaque random values. Only SHA-256 hashes are stored in MongoDB. Every refresh creates a new session and revokes the old token hash.
## SMS Provider Strategy
![SMS provider strategy](assets/diagrams/sms_provider_strategy.png)
The worker depends on the `SmsClient` port and receives the concrete provider from a small factory. Kavenegar, SMS.ir, and debug-local delivery share the same interface.
## Production Deployment
![Production deployment](assets/diagrams/production_deployment.png)
Production exposes only Caddy on ports `80` and `443`. Caddy routes `gapido.amiirkhl.ir` to `demo-app`. gRPC, MongoDB, Redis, RabbitMQ, and the SMS worker remain private on the Docker network.

Binary file not shown.

After

Width:  |  Height:  |  Size: 155 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 176 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 158 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 169 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 156 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 130 KiB

416
docs/diagrams/generate.py Normal file
View File

@@ -0,0 +1,416 @@
import struct
import zlib
from pathlib import Path
from diagrams import Cluster, Diagram, Edge
from diagrams.generic.blank import Blank
from diagrams.onprem.client import Client, User
from diagrams.onprem.compute import Server
from diagrams.onprem.container import Docker
from diagrams.onprem.database import Mongodb
from diagrams.onprem.inmemory import Redis
from diagrams.onprem.network import Caddy, Internet
from diagrams.onprem.queue import Rabbitmq
from diagrams.onprem.security import Vault
ROOT = Path(__file__).resolve().parents[2]
OUT = ROOT / "docs" / "assets" / "diagrams"
TARGET_RATIO = 16 / 9
def graph_attr() -> dict[str, str]:
"""Return shared Graphviz settings for clean 16:9 diagram output."""
return {
"bgcolor": "white",
"pad": "0.42",
"ranksep": "1.05",
"nodesep": "0.72",
"fontname": "Arial",
"fontsize": "18",
"dpi": "220",
"size": "16,9!",
"ratio": "fill",
"splines": "spline",
"outputorder": "edgesfirst",
}
def node_attr() -> dict[str, str]:
"""Return readable node text styling while preserving provider icons."""
return {
"fontname": "Arial",
"fontsize": "14",
"fontcolor": "#172033",
"labelloc": "b",
"margin": "0.12",
}
def edge_attr() -> dict[str, str]:
"""Return restrained edge styling for presentation diagrams."""
return {
"fontname": "Arial",
"fontsize": "11",
"color": "#64748B",
"fontcolor": "#334155",
"arrowsize": "0.75",
"penwidth": "1.6",
}
def cluster_attr() -> dict[str, str]:
"""Return subtle cluster styling that keeps related services visually grouped."""
return {
"bgcolor": "#F8FAFC",
"fontname": "Arial",
"fontsize": "16",
"fontcolor": "#334155",
"pencolor": "#CBD5E1",
}
def diagram(name: str, filename: str, *, direction: str = "LR") -> Diagram:
"""Create a 16:9 icon-based diagram using the diagrams package."""
return Diagram(
name,
filename=str(OUT / filename),
show=False,
direction=direction,
outformat="png",
curvestyle="curved",
graph_attr=graph_attr(),
node_attr=node_attr(),
edge_attr=edge_attr(),
)
def normalize_png(filename: str) -> None:
"""Center a rendered PNG on an exact 16:9 white canvas."""
path = OUT / f"{filename}.png"
png = _read_png(path)
target_width, target_height = _target_size(png.width, png.height)
if (png.width, png.height) == (target_width, target_height):
return
channels = png.channels
white_pixel = b"\xff" * channels
rows = [bytearray(white_pixel * target_width) for _ in range(target_height)]
offset_x = (target_width - png.width) // 2
offset_y = (target_height - png.height) // 2
for source_y, source_row in enumerate(png.rows):
target_row = rows[offset_y + source_y]
start = offset_x * channels
target_row[start : start + len(source_row)] = source_row
_write_png(path, PngImage(target_width, target_height, png.color_type, rows))
class PngImage:
"""Small in-memory PNG representation for deterministic 16:9 padding."""
def __init__(
self,
width: int,
height: int,
color_type: int,
rows: list[bytearray],
) -> None:
self.width = width
self.height = height
self.color_type = color_type
self.rows = rows
@property
def channels(self) -> int:
"""Return channel count for supported truecolor PNG types."""
return 4 if self.color_type == 6 else 3
def _target_size(width: int, height: int) -> tuple[int, int]:
"""Calculate the smallest exact 16:9 canvas that contains the diagram."""
if width / height < TARGET_RATIO:
return round(height * TARGET_RATIO), height
return width, round(width / TARGET_RATIO)
def _read_png(path: Path) -> PngImage:
"""Read a non-interlaced 8-bit RGB/RGBA PNG produced by Graphviz."""
data = path.read_bytes()
if not data.startswith(b"\x89PNG\r\n\x1a\n"):
raise ValueError(f"{path} is not a PNG file")
position = 8
width = height = color_type = 0
compressed = bytearray()
while position < len(data):
length = struct.unpack(">I", data[position : position + 4])[0]
chunk_type = data[position + 4 : position + 8]
chunk_data = data[position + 8 : position + 8 + length]
position += 12 + length
if chunk_type == b"IHDR":
width, height, bit_depth, color_type, _, _, interlace = struct.unpack(
">IIBBBBB", chunk_data
)
if bit_depth != 8 or color_type not in {2, 6} or interlace != 0:
raise ValueError(f"Unsupported PNG encoding in {path}")
elif chunk_type == b"IDAT":
compressed.extend(chunk_data)
elif chunk_type == b"IEND":
break
channels = 4 if color_type == 6 else 3
stride = width * channels
raw = zlib.decompress(bytes(compressed))
rows: list[bytearray] = []
previous = bytearray(stride)
cursor = 0
for _ in range(height):
filter_type = raw[cursor]
cursor += 1
row = bytearray(raw[cursor : cursor + stride])
cursor += stride
_unfilter(row, previous, filter_type, channels)
rows.append(row)
previous = row
return PngImage(width, height, color_type, rows)
def _unfilter(row: bytearray, previous: bytearray, filter_type: int, channels: int) -> None:
"""Reverse PNG scanline filters used by Graphviz output."""
for index in range(len(row)):
left = row[index - channels] if index >= channels else 0
up = previous[index]
upper_left = previous[index - channels] if index >= channels else 0
match filter_type:
case 0:
value = row[index]
case 1:
value = row[index] + left
case 2:
value = row[index] + up
case 3:
value = row[index] + ((left + up) // 2)
case 4:
value = row[index] + _paeth(left, up, upper_left)
case _:
raise ValueError(f"Unsupported PNG filter type: {filter_type}")
row[index] = value & 0xFF
def _paeth(left: int, up: int, upper_left: int) -> int:
"""Return the PNG Paeth predictor for a single channel."""
prediction = left + up - upper_left
left_distance = abs(prediction - left)
up_distance = abs(prediction - up)
upper_left_distance = abs(prediction - upper_left)
if left_distance <= up_distance and left_distance <= upper_left_distance:
return left
if up_distance <= upper_left_distance:
return up
return upper_left
def _write_png(path: Path, png: PngImage) -> None:
"""Write a simple unfiltered PNG with the same color type as the source."""
raw = bytearray()
for row in png.rows:
raw.append(0)
raw.extend(row)
ihdr = struct.pack(">IIBBBBB", png.width, png.height, 8, png.color_type, 0, 0, 0)
chunks = [
_chunk(b"IHDR", ihdr),
_chunk(b"IDAT", zlib.compress(bytes(raw), level=6)),
_chunk(b"IEND", b""),
]
path.write_bytes(b"\x89PNG\r\n\x1a\n" + b"".join(chunks))
def _chunk(chunk_type: bytes, data: bytes) -> bytes:
"""Create one PNG chunk with CRC."""
return (
struct.pack(">I", len(data))
+ chunk_type
+ data
+ struct.pack(">I", zlib.crc32(chunk_type + data) & 0xFFFFFFFF)
)
def link(label: str) -> Edge:
"""Create a consistent labeled connector."""
return Edge(label=label)
def render_service_architecture() -> None:
with diagram("Service Architecture", "service_architecture"):
with Cluster("Client boundary", graph_attr=cluster_attr()):
browser = Client("Browser")
demo = Docker("demo-app\nFastAPI BFF")
with Cluster("Application services", graph_attr=cluster_attr()):
auth = Server("auth-service\ngRPC API")
worker = Docker("sms-worker")
with Cluster("State and messaging", graph_attr=cluster_attr()):
mongo = Mongodb("MongoDB\nusers + sessions")
redis = Redis("Redis\nOTP TTL + limits")
rabbit = Rabbitmq("RabbitMQ\nSMS jobs")
providers = Internet("Kavenegar / SMS.ir")
browser >> link("HTTPS / local HTTP") >> demo >> link("gRPC") >> auth
auth >> link("documents") >> mongo
auth >> link("ephemeral state") >> redis
auth >> link("durable job") >> rabbit >> link("consume") >> worker
worker >> link("provider API") >> providers
normalize_png("service_architecture")
def render_clean_architecture() -> None:
with diagram("Clean Architecture Layers", "clean_architecture", direction="TB"):
with Cluster("Transport", graph_attr=cluster_attr()):
demo = Client("Demo BFF")
grpc = Server("gRPC servicer")
with Cluster("Application", graph_attr=cluster_attr()):
auth = Server("AuthService\nuse cases")
security = Vault("JWT + OTP\nsecurity helpers")
with Cluster("Domain", graph_attr=cluster_attr()):
entities = Blank("Entities\nUser, Session, Role")
ports = Blank("Ports\nrepositories + SMS")
with Cluster("Infrastructure", graph_attr=cluster_attr()):
mongo = Mongodb("Mongo repositories")
redis = Redis("Redis OTP store")
rabbit = Rabbitmq("RabbitMQ publisher")
sms = Internet("SMS strategies")
demo >> link("calls") >> grpc >> link("executes") >> auth
auth >> link("uses") >> entities
auth >> link("depends on") >> ports
auth >> link("delegates crypto") >> security
ports << link("implements") << [mongo, redis, rabbit, sms]
normalize_png("clean_architecture")
def render_otp_flow() -> None:
with diagram("OTP Login Flow", "otp_login_flow"):
user = User("User")
with Cluster("Request OTP", graph_attr=cluster_attr()):
demo_request = Client("demo-app")
auth_request = Server("RequestOtp")
redis_store = Redis("Redis\nstore HMAC + TTL")
rabbit = Rabbitmq("RabbitMQ\npublish SMS job")
with Cluster("Delivery", graph_attr=cluster_attr()):
worker = Docker("sms-worker")
sms = Internet("SMS provider")
with Cluster("Verify OTP", graph_attr=cluster_attr()):
demo_verify = Client("demo-app")
auth_verify = Server("VerifyOtp")
redis_verify = Redis("Redis\ncompare hash")
mongo = Mongodb("MongoDB\nuser + session")
user >> link("1. mobile") >> demo_request >> link("2. gRPC") >> auth_request
auth_request >> link("3. hash only") >> redis_store
auth_request >> link("4. queued") >> rabbit >> link("5. consume") >> worker
worker >> link("6. send OTP") >> sms
user >> link("7. code") >> demo_verify >> link("8. gRPC") >> auth_verify
auth_verify >> link("9. verify") >> redis_verify
auth_verify >> link("10. issue session") >> mongo
auth_verify >> link("11. token pair") >> demo_verify
normalize_png("otp_login_flow")
def render_refresh_flow() -> None:
with diagram("Refresh Token Rotation", "refresh_rotation_flow"):
client = Client("Client")
auth = Server("auth-service\nRefreshToken")
mongo = Mongodb("MongoDB\nrefresh_sessions")
with Cluster("Rotation result", graph_attr=cluster_attr()):
old = Vault("Old hash\nrevoked")
new = Vault("New hash\nactive")
client >> link("1. opaque refresh token") >> auth
auth >> link("2. hash lookup") >> mongo >> link("3. active session") >> auth
auth >> link("4. revoke") >> old
auth >> link("5. create") >> new >> link("6. persist") >> mongo
auth >> link("7. new token pair") >> client
normalize_png("refresh_rotation_flow")
def render_sms_strategy() -> None:
with diagram("SMS Provider Strategy", "sms_provider_strategy", direction="TB"):
worker = Docker("sms-worker")
factory = Server("create_sms_client()\nprovider factory")
port = Blank("SmsClient port\nsend_otp(...)")
with Cluster("Concrete strategies", graph_attr=cluster_attr()):
kavenegar = Internet("Kavenegar")
sms_ir = Internet("SMS.ir")
debug = Redis("Debug provider\nlocal Redis")
worker >> link("startup wiring") >> factory >> link("returns interface") >> port
port >> link("SMS_PROVIDER=kavenegar") >> kavenegar
port >> link("SMS_PROVIDER=sms_ir") >> sms_ir
port >> link("SMS_PROVIDER=debug") >> debug
normalize_png("sms_provider_strategy")
def render_production_deployment() -> None:
with diagram("Production Deployment", "production_deployment"):
internet = Internet("Internet\ngapido.amiirkhl.ir")
caddy = Caddy("Caddy\n80 / 443 only")
with Cluster("Private Docker network", graph_attr=cluster_attr()):
demo = Docker("demo-app\nFastAPI UI")
auth = Server("auth-service\nprivate gRPC")
mongo = Mongodb("MongoDB\nprivate")
redis = Redis("Redis\nprivate")
rabbit = Rabbitmq("RabbitMQ\nprivate")
worker = Docker("sms-worker")
sms = Internet("SMS provider")
internet >> link("HTTPS") >> caddy >> link("reverse proxy") >> demo
demo >> link("gRPC") >> auth
auth >> link("sessions") >> mongo
auth >> link("OTP state") >> redis
auth >> link("SMS job") >> rabbit >> link("consume") >> worker
worker >> link("provider API") >> sms
normalize_png("production_deployment")
def main() -> None:
"""Regenerate all 16:9 icon-based diagrams for docs and slides."""
OUT.mkdir(parents=True, exist_ok=True)
render_service_architecture()
render_clean_architecture()
render_otp_flow()
render_refresh_flow()
render_sms_strategy()
render_production_deployment()
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,34 @@
# Implementation Decisions
## Python gRPC Service
The challenge explicitly required Python with gRPC. The auth boundary is therefore a `grpc.aio` service instead of REST. A separate FastAPI demo service was added only as a browser-friendly client.
## OTP Storage
OTP codes are never stored in plaintext. Redis stores HMAC hashes with a short TTL, attempt counters, and request rate-limit counters. This keeps OTP state fast, temporary, and easy to expire.
## Token Model
Access tokens are signed JWTs with short TTLs. Refresh tokens are opaque random values, stored only as hashes in MongoDB, and rotated on every use.
## RabbitMQ SMS Dispatch
SMS delivery is asynchronous. The auth service publishes a durable message and returns quickly. The worker uses manual acknowledgement, bounded retries, and a dead-letter queue for failed deliveries.
## SMS Provider Strategy
Kavenegar, SMS.ir, and debug delivery implement the same `SmsClient` port. The provider is selected through configuration, which keeps the worker closed for modification when adding providers.
## Demo Service
Browsers do not speak native gRPC directly. A small FastAPI backend-for-frontend demonstrates how another microservice consumes the auth service through gRPC while serving a minimal UI.
## Local vs Production Compose
The base Compose file is production-safe and keeps service ports private. `docker-compose.local.yml` publishes developer ports. `docker-compose.prod.yml` adds Caddy as the only public entrypoint.
## Caddy Reverse Proxy
Caddy was chosen for automatic HTTPS and a compact configuration. Production serves only `https://gapido.amiirkhl.ir`; internal service ports are not published.

87
docs/slides/index.html Normal file
View File

@@ -0,0 +1,87 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<title>Gapido Auth Challenge</title>
<link rel="stylesheet" href="styles.css" />
</head>
<body>
<header class="topbar">
<span>Gapido Backend Challenge</span>
<div class="controls">
<button id="theme-toggle" type="button">Toggle theme</button>
<button id="prev" type="button">Prev</button>
<button id="next" type="button">Next</button>
</div>
</header>
<main class="deck">
<section class="slide active">
<p class="eyebrow">The problem</p>
<h1>Secure OTP auth for microservices</h1>
<p class="lede">Python / gRPC / MongoDB / Redis / RabbitMQ / Docker</p>
</section>
<section class="slide diagram">
<p class="eyebrow">Architecture</p>
<h2>Project architecture</h2>
<img src="../assets/diagrams/service_architecture.png" alt="Service architecture" />
</section>
<section class="slide diagram">
<p class="eyebrow">Codebase</p>
<h2>Clean architecture structure</h2>
<img src="../assets/diagrams/clean_architecture.png" alt="Clean architecture layers" />
</section>
<section class="slide diagram">
<p class="eyebrow">Auth flow</p>
<h2>OTP login flow</h2>
<img src="../assets/diagrams/otp_login_flow.png" alt="OTP login flow" />
</section>
<section class="slide diagram">
<p class="eyebrow">Extensibility</p>
<h2>SMS provider strategy</h2>
<img src="../assets/diagrams/sms_provider_strategy.png" alt="SMS provider strategy" />
</section>
<section class="slide diagram">
<p class="eyebrow">Token safety</p>
<h2>Refresh token rotation</h2>
<img src="../assets/diagrams/refresh_rotation_flow.png" alt="Refresh token rotation" />
</section>
<section class="slide diagram">
<p class="eyebrow">Deployment</p>
<h2>Production deployment</h2>
<img src="../assets/diagrams/production_deployment.png" alt="Production deployment" />
</section>
<section class="slide split">
<div>
<p class="eyebrow">Tests</p>
<h2>Coverage report</h2>
</div>
<ul class="tiles">
<li>16 automated tests passing</li>
<li>OTP and token use cases</li>
<li>gRPC auth behavior</li>
<li>Demo BFF routes</li>
<li>SMS provider adapters</li>
<li>ruff, mypy, compose config</li>
</ul>
</section>
<section class="slide">
<p class="eyebrow">Outcome</p>
<h2>Interview-ready microservice demo</h2>
<p class="lede">A focused auth service with clean boundaries, secure OTP, async messaging, provider strategy, and production deployment posture.</p>
</section>
</main>
<footer class="progress"><span id="counter">1 / 9</span></footer>
<script src="slides.js"></script>
</body>
</html>

28
docs/slides/slides.js Normal file
View File

@@ -0,0 +1,28 @@
const slides = [...document.querySelectorAll(".slide")];
const counter = document.querySelector("#counter");
const root = document.documentElement;
let index = 0;
function render() {
slides.forEach((slide, current) => slide.classList.toggle("active", current === index));
counter.textContent = `${index + 1} / ${slides.length}`;
}
function move(delta) {
index = (index + delta + slides.length) % slides.length;
render();
}
document.querySelector("#next").addEventListener("click", () => move(1));
document.querySelector("#prev").addEventListener("click", () => move(-1));
document.querySelector("#theme-toggle").addEventListener("click", () => {
root.classList.toggle("light");
});
document.addEventListener("keydown", (event) => {
if (event.key === "ArrowRight" || event.key === " ") move(1);
if (event.key === "ArrowLeft") move(-1);
});
render();

178
docs/slides/styles.css Normal file
View File

@@ -0,0 +1,178 @@
:root {
color-scheme: dark;
--bg: #070b14;
--panel: #101827;
--text: #e6edf7;
--muted: #93a4bc;
--accent: #34d399;
--border: #223149;
}
:root.light {
color-scheme: light;
--bg: #f7fafc;
--panel: #ffffff;
--text: #172033;
--muted: #5f7088;
--accent: #0f766e;
--border: #d8e1ee;
}
* {
box-sizing: border-box;
}
body {
margin: 0;
min-height: 100vh;
font-family: Inter, ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
background: var(--bg);
color: var(--text);
}
.topbar,
.progress {
position: fixed;
left: 0;
right: 0;
z-index: 10;
display: flex;
justify-content: space-between;
align-items: center;
padding: 16px 24px;
color: var(--muted);
}
.topbar {
top: 0;
}
.progress {
bottom: 0;
}
.controls {
display: flex;
gap: 8px;
}
button {
border: 1px solid var(--border);
border-radius: 8px;
background: var(--panel);
color: var(--text);
padding: 8px 12px;
font-weight: 700;
cursor: pointer;
}
.deck {
min-height: 100vh;
display: grid;
place-items: center;
padding: 72px 5vw;
}
.slide {
display: none;
width: min(1120px, 100%);
min-height: 620px;
padding: 48px;
border: 1px solid var(--border);
border-radius: 12px;
background: var(--panel);
box-shadow: 0 24px 80px rgb(0 0 0 / 24%);
}
.slide.active {
display: grid;
align-content: center;
gap: 24px;
}
.split.active {
grid-template-columns: 0.9fr 1.1fr;
align-items: center;
}
.diagram.active {
align-content: start;
}
.eyebrow {
margin: 0;
color: var(--accent);
font-size: 14px;
font-weight: 800;
text-transform: uppercase;
}
h1,
h2 {
margin: 0;
letter-spacing: 0;
}
h1 {
max-width: 900px;
font-size: clamp(54px, 7vw, 92px);
line-height: 0.95;
}
h2 {
font-size: clamp(36px, 5vw, 62px);
line-height: 1;
}
.lede {
max-width: 820px;
margin: 0;
color: var(--muted);
font-size: 24px;
line-height: 1.45;
}
.tiles {
list-style: none;
margin: 0;
padding: 0;
display: grid;
grid-template-columns: repeat(2, minmax(0, 1fr));
gap: 14px;
}
.tiles li {
min-height: 96px;
display: grid;
place-items: center;
padding: 18px;
border: 1px solid var(--border);
border-radius: 8px;
color: var(--text);
background: color-mix(in srgb, var(--panel), var(--accent) 6%);
font-size: 20px;
font-weight: 800;
text-align: center;
}
img {
width: 100%;
max-height: 460px;
object-fit: contain;
border-radius: 8px;
background: #ffffff;
padding: 12px;
}
@media (max-width: 820px) {
.slide {
min-height: 620px;
padding: 28px;
}
.split.active,
.tiles {
grid-template-columns: 1fr;
}
}

View File

@@ -21,6 +21,7 @@ dependencies = [
[project.optional-dependencies] [project.optional-dependencies]
dev = [ dev = [
"diagrams==0.24.4",
"grpcio-tools==1.68.1", "grpcio-tools==1.68.1",
"mypy==1.14.1", "mypy==1.14.1",
"pytest==8.3.4", "pytest==8.3.4",

View File

@@ -28,6 +28,8 @@ from gapido_auth.domain.ports import (
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class AuthConfig: class AuthConfig:
"""Runtime policy values used by auth use cases."""
otp_secret: str otp_secret: str
otp_ttl_seconds: int otp_ttl_seconds: int
otp_max_attempts: int otp_max_attempts: int
@@ -38,6 +40,8 @@ class AuthConfig:
class AuthService: class AuthService:
"""Application service coordinating OTP login, token rotation, and RBAC."""
def __init__( def __init__(
self, self,
users: UserRepository, users: UserRepository,
@@ -47,6 +51,7 @@ class AuthService:
token_codec: JwtTokenCodec, token_codec: JwtTokenCodec,
config: AuthConfig, config: AuthConfig,
) -> None: ) -> None:
"""Wire repository, queue, OTP, and token ports for use-case execution."""
self._users = users self._users = users
self._refresh_sessions = refresh_sessions self._refresh_sessions = refresh_sessions
self._otp_store = otp_store self._otp_store = otp_store
@@ -55,6 +60,8 @@ class AuthService:
self._config = config self._config = config
async def request_otp(self, mobile: str, purpose: str, client_key: str) -> None: async def request_otp(self, mobile: str, purpose: str, client_key: str) -> None:
"""Validate a request, store an OTP hash, and queue SMS delivery."""
_validate_mobile(mobile) _validate_mobile(mobile)
_validate_purpose(purpose) _validate_purpose(purpose)
mobile_key = f"otp-request:mobile:{mobile}:{purpose}" mobile_key = f"otp-request:mobile:{mobile}:{purpose}"
@@ -73,6 +80,7 @@ class AuthService:
raise RateLimitExceeded("too many OTP requests") raise RateLimitExceeded("too many OTP requests")
code = generate_otp_code() code = generate_otp_code()
# Store only a bound HMAC hash; the plaintext code leaves through the SMS queue only.
otp_hash = hash_otp(self._config.otp_secret, mobile, purpose, code) otp_hash = hash_otp(self._config.otp_secret, mobile, purpose, code)
await self._otp_store.store_otp( await self._otp_store.store_otp(
mobile=mobile, mobile=mobile,
@@ -86,6 +94,8 @@ class AuthService:
) )
async def verify_otp(self, mobile: str, code: str, purpose: str) -> TokenPair: async def verify_otp(self, mobile: str, code: str, purpose: str) -> TokenPair:
"""Verify the submitted OTP and issue a token pair for the mobile identity."""
_validate_mobile(mobile) _validate_mobile(mobile)
_validate_purpose(purpose) _validate_purpose(purpose)
if not code.isdigit() or len(code) != 6: if not code.isdigit() or len(code) != 6:
@@ -100,6 +110,8 @@ class AuthService:
return await self._issue_token_pair(user) return await self._issue_token_pair(user)
async def refresh_token(self, refresh_token: str) -> TokenPair: async def refresh_token(self, refresh_token: str) -> TokenPair:
"""Rotate a valid refresh token and return a new access/refresh pair."""
token_hash = hash_refresh_token(refresh_token) token_hash = hash_refresh_token(refresh_token)
now = utc_now() now = utc_now()
session = await self._refresh_sessions.get_active_by_hash(token_hash, now) session = await self._refresh_sessions.get_active_by_hash(token_hash, now)
@@ -110,6 +122,7 @@ class AuthService:
if user is None or not user.is_active: if user is None or not user.is_active:
raise AuthenticationError("invalid refresh token") raise AuthenticationError("invalid refresh token")
# Rotation revokes the old token hash and persists a fresh session hash.
new_refresh_token = generate_refresh_token() new_refresh_token = generate_refresh_token()
new_hash = hash_refresh_token(new_refresh_token) new_hash = hash_refresh_token(new_refresh_token)
expires_at = now + timedelta(seconds=self._config.refresh_token_ttl_seconds) expires_at = now + timedelta(seconds=self._config.refresh_token_ttl_seconds)
@@ -125,9 +138,13 @@ class AuthService:
) )
async def revoke_refresh_token(self, refresh_token: str) -> None: async def revoke_refresh_token(self, refresh_token: str) -> None:
"""Revoke a refresh token session if it is still active."""
await self._refresh_sessions.revoke(hash_refresh_token(refresh_token), utc_now()) await self._refresh_sessions.revoke(hash_refresh_token(refresh_token), utc_now())
async def get_authenticated_user(self, access_token: str) -> User: async def get_authenticated_user(self, access_token: str) -> User:
"""Resolve an access token into an active user entity."""
claims = self._token_codec.decode_access_token(access_token) claims = self._token_codec.decode_access_token(access_token)
user = await self._users.get_by_id(claims.user_id) user = await self._users.get_by_id(claims.user_id)
if user is None: if user is None:
@@ -137,12 +154,16 @@ class AuthService:
return user return user
async def require_role(self, access_token: str, role: Role) -> User: async def require_role(self, access_token: str, role: Role) -> User:
"""Resolve a user and ensure the requested role is present."""
user = await self.get_authenticated_user(access_token) user = await self.get_authenticated_user(access_token)
if user.role != role: if user.role != role:
raise PermissionDenied("insufficient permissions") raise PermissionDenied("insufficient permissions")
return user return user
async def _issue_token_pair(self, user: User) -> TokenPair: async def _issue_token_pair(self, user: User) -> TokenPair:
"""Create a JWT access token and persisted refresh session for a user."""
if not user.is_active: if not user.is_active:
raise InactiveUser("user is inactive") raise InactiveUser("user is inactive")
@@ -160,11 +181,15 @@ class AuthService:
def _validate_mobile(mobile: str) -> None: def _validate_mobile(mobile: str) -> None:
"""Validate the E.164-like mobile format accepted by the challenge service."""
normalized = mobile.removeprefix("+") normalized = mobile.removeprefix("+")
if not normalized.isdigit() or len(normalized) < 10 or len(normalized) > 15: if not normalized.isdigit() or len(normalized) < 10 or len(normalized) > 15:
raise ValidationError("mobile must be an E.164-like phone number") raise ValidationError("mobile must be an E.164-like phone number")
def _validate_purpose(purpose: str) -> None: def _validate_purpose(purpose: str) -> None:
"""Validate the OTP purpose used to namespace OTP hashes and rate limits."""
if not purpose or not purpose.replace("-", "").replace("_", "").isalnum(): if not purpose or not purpose.replace("-", "").replace("_", "").isalnum():
raise ValidationError("purpose is invalid") raise ValidationError("purpose is invalid")

View File

@@ -11,37 +11,54 @@ from gapido_auth.domain.errors import AuthenticationError
def utc_now() -> datetime: def utc_now() -> datetime:
"""Return timezone-aware UTC time for token/session timestamps."""
return datetime.now(UTC) return datetime.now(UTC)
def generate_otp_code() -> str: def generate_otp_code() -> str:
"""Generate a cryptographically random six-digit OTP string."""
return f"{secrets.randbelow(1_000_000):06d}" return f"{secrets.randbelow(1_000_000):06d}"
def hash_otp(secret: str, mobile: str, purpose: str, code: str) -> str: def hash_otp(secret: str, mobile: str, purpose: str, code: str) -> str:
"""Bind an OTP to its mobile and purpose before storing only an HMAC hash."""
message = f"{mobile}:{purpose}:{code}".encode() message = f"{mobile}:{purpose}:{code}".encode()
return hmac.new(secret.encode(), message, hashlib.sha256).hexdigest() return hmac.new(secret.encode(), message, hashlib.sha256).hexdigest()
def generate_refresh_token() -> str: def generate_refresh_token() -> str:
"""Generate an opaque refresh token suitable for returning to clients."""
return secrets.token_urlsafe(48) return secrets.token_urlsafe(48)
def hash_refresh_token(token: str) -> str: def hash_refresh_token(token: str) -> str:
"""Hash a refresh token before persistence so plaintext tokens are never stored."""
return hashlib.sha256(token.encode()).hexdigest() return hashlib.sha256(token.encode()).hexdigest()
class JwtTokenCodec: class JwtTokenCodec:
"""Encode and validate short-lived access JWTs for authenticated gRPC calls."""
def __init__(self, secret_key: str, issuer: str, access_ttl_seconds: int) -> None: def __init__(self, secret_key: str, issuer: str, access_ttl_seconds: int) -> None:
"""Store signing configuration used for all access-token operations."""
self._secret_key = secret_key self._secret_key = secret_key
self._issuer = issuer self._issuer = issuer
self._access_ttl_seconds = access_ttl_seconds self._access_ttl_seconds = access_ttl_seconds
@property @property
def access_ttl_seconds(self) -> int: def access_ttl_seconds(self) -> int:
"""Return the configured access-token TTL exposed to clients."""
return self._access_ttl_seconds return self._access_ttl_seconds
def encode_access_token(self, user_id: str, role: Role) -> str: def encode_access_token(self, user_id: str, role: Role) -> str:
"""Create a signed access JWT containing user identity, role, and expiry."""
now = utc_now() now = utc_now()
expires_at = now + timedelta(seconds=self._access_ttl_seconds) expires_at = now + timedelta(seconds=self._access_ttl_seconds)
payload = { payload = {
@@ -56,6 +73,8 @@ class JwtTokenCodec:
return jwt.encode(payload, self._secret_key, algorithm="HS256") return jwt.encode(payload, self._secret_key, algorithm="HS256")
def decode_access_token(self, token: str) -> AccessClaims: def decode_access_token(self, token: str) -> AccessClaims:
"""Validate an access JWT and return typed claims used by RBAC checks."""
try: try:
payload = jwt.decode( payload = jwt.decode(
token, token,
@@ -80,4 +99,3 @@ class JwtTokenCodec:
role=role, role=role,
expires_at=datetime.fromtimestamp(int(payload["exp"]), UTC), expires_at=datetime.fromtimestamp(int(payload["exp"]), UTC),
) )

View File

@@ -6,6 +6,8 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings): class Settings(BaseSettings):
"""Environment-driven settings shared by auth, worker, and demo services."""
app_env: str = "local" app_env: str = "local"
mongo_uri: str = "mongodb://localhost:27017" mongo_uri: str = "mongodb://localhost:27017"
@@ -46,4 +48,6 @@ class Settings(BaseSettings):
@lru_cache @lru_cache
def get_settings() -> Settings: def get_settings() -> Settings:
"""Return cached process settings loaded from environment and optional .env files."""
return Settings() return Settings()

View File

@@ -4,12 +4,16 @@ from enum import StrEnum
class Role(StrEnum): class Role(StrEnum):
"""User roles used by the auth service for access-control checks."""
ADMIN = "admin" ADMIN = "admin"
USER = "user" USER = "user"
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class User: class User:
"""Persisted account identity created after a successful OTP verification."""
id: str id: str
mobile: str mobile: str
role: Role role: Role
@@ -20,6 +24,8 @@ class User:
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class RefreshSession: class RefreshSession:
"""Persisted refresh-token session stored as a token hash, never plaintext."""
id: str id: str
user_id: str user_id: str
token_hash: str token_hash: str
@@ -31,6 +37,8 @@ class RefreshSession:
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class SmsJob: class SmsJob:
"""RabbitMQ payload for delivering an OTP through the configured SMS provider."""
mobile: str mobile: str
code: str code: str
template: str template: str
@@ -39,6 +47,8 @@ class SmsJob:
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class TokenPair: class TokenPair:
"""Access and refresh tokens returned to a client after login or refresh."""
access_token: str access_token: str
refresh_token: str refresh_token: str
token_type: str token_type: str
@@ -48,7 +58,8 @@ class TokenPair:
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class AccessClaims: class AccessClaims:
"""Trusted claims extracted from a validated access JWT."""
user_id: str user_id: str
role: Role role: Role
expires_at: datetime expires_at: datetime

View File

@@ -3,36 +3,54 @@ class AppError(Exception):
class RateLimitExceeded(AppError): class RateLimitExceeded(AppError):
"""Raised when OTP request or verification limits are exceeded."""
pass pass
class ValidationError(AppError): class ValidationError(AppError):
"""Raised when client input cannot be accepted by the application layer."""
pass pass
class InvalidOtp(AppError): class InvalidOtp(AppError):
"""Raised when an OTP exists but the submitted code does not match."""
pass pass
class OtpExpired(AppError): class OtpExpired(AppError):
"""Raised when an OTP is missing because it expired or was never requested."""
pass pass
class OtpAttemptsExceeded(AppError): class OtpAttemptsExceeded(AppError):
"""Raised when OTP verification attempts exceed the configured limit."""
pass pass
class AuthenticationError(AppError): class AuthenticationError(AppError):
"""Raised when credentials or tokens cannot authenticate a caller."""
pass pass
class PermissionDenied(AppError): class PermissionDenied(AppError):
"""Raised when an authenticated caller lacks the required role."""
pass pass
class InactiveUser(AppError): class InactiveUser(AppError):
"""Raised when an existing user is disabled and cannot authenticate."""
pass pass
class ExternalServiceError(AppError): class ExternalServiceError(AppError):
"""Raised when an external provider fails or returns an error response."""
pass pass

View File

@@ -5,29 +5,51 @@ from gapido_auth.domain.entities import RefreshSession, Role, SmsJob, User
class UserRepository(Protocol): class UserRepository(Protocol):
async def get_by_id(self, user_id: str) -> User | None: ... """Persistence port for user lookup, creation, and admin bootstrapping."""
async def get_by_mobile(self, mobile: str) -> User | None: ... async def get_by_id(self, user_id: str) -> User | None:
"""Return a user by internal id, or None when it does not exist."""
...
async def get_or_create_user(self, mobile: str, role: Role = Role.USER) -> User: ... async def get_by_mobile(self, mobile: str) -> User | None:
"""Return a user by mobile number, or None when it does not exist."""
...
async def ensure_admin_user(self, mobile: str) -> User: ... async def get_or_create_user(self, mobile: str, role: Role = Role.USER) -> User:
"""Return the existing user for a mobile or create one with the default role."""
...
async def ensure_admin_user(self, mobile: str) -> User:
"""Seed or promote the configured admin mobile idempotently."""
...
class RefreshSessionRepository(Protocol): class RefreshSessionRepository(Protocol):
"""Persistence port for refresh-token session creation and rotation."""
async def create( async def create(
self, user_id: str, token_hash: str, expires_at: datetime self, user_id: str, token_hash: str, expires_at: datetime
) -> RefreshSession: ... ) -> RefreshSession:
"""Persist a hashed refresh-token session."""
...
async def get_active_by_hash(self, token_hash: str, now: datetime) -> RefreshSession | None: ... async def get_active_by_hash(self, token_hash: str, now: datetime) -> RefreshSession | None:
"""Return a non-expired active refresh session by token hash."""
...
async def revoke( async def revoke(
self, token_hash: str, now: datetime, replaced_by_hash: str | None = None self, token_hash: str, now: datetime, replaced_by_hash: str | None = None
) -> None: ... ) -> None:
"""Mark a refresh session as revoked, optionally linking its replacement."""
...
class OtpStore(Protocol): class OtpStore(Protocol):
async def allow_request(self, key: str, limit: int, window_seconds: int) -> bool: ... """Temporary OTP storage port with TTL, attempt, and rate-limit behavior."""
async def allow_request(self, key: str, limit: int, window_seconds: int) -> bool:
"""Check and increment a named rate-limit bucket."""
...
async def store_otp( async def store_otp(
self, self,
@@ -36,14 +58,26 @@ class OtpStore(Protocol):
otp_hash: str, otp_hash: str,
ttl_seconds: int, ttl_seconds: int,
max_attempts: int, max_attempts: int,
) -> None: ... ) -> None:
"""Store a hashed OTP with TTL and attempt policy."""
...
async def verify_otp(self, mobile: str, purpose: str, candidate_hash: str) -> bool: ... async def verify_otp(self, mobile: str, purpose: str, candidate_hash: str) -> bool:
"""Verify an OTP hash and apply attempt/expiry behavior."""
...
class SmsPublisher(Protocol): class SmsPublisher(Protocol):
async def publish(self, job: SmsJob) -> None: ... """Queue publishing port used by auth use cases to request SMS delivery."""
async def publish(self, job: SmsJob) -> None:
"""Publish an OTP SMS job to the messaging boundary."""
...
class SmsClient(Protocol): class SmsClient(Protocol):
async def send_otp(self, mobile: str, code: str, template: str) -> None: ... """Provider strategy port implemented by Kavenegar, SMS.ir, and debug SMS."""
async def send_otp(self, mobile: str, code: str, template: str) -> None:
"""Send an OTP through a concrete SMS provider."""
...

View File

@@ -1,3 +1,5 @@
import asyncio
import logging
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any from typing import Any
@@ -16,24 +18,34 @@ from gapido_auth.infrastructure.rabbitmq import RabbitMqSmsPublisher
from gapido_auth.infrastructure.redis_otp_store import RedisOtpStore from gapido_auth.infrastructure.redis_otp_store import RedisOtpStore
from gapido_auth.infrastructure.sms_provider import get_sms_template from gapido_auth.infrastructure.sms_provider import get_sms_template
logger = logging.getLogger(__name__)
STARTUP_RETRY_ATTEMPTS = 30
STARTUP_RETRY_DELAY_SECONDS = 2
@dataclass(slots=True) @dataclass(slots=True)
class AppContainer: class AppContainer:
"""Runtime dependencies that need explicit shutdown after the gRPC server stops."""
auth_service: AuthService auth_service: AuthService
mongo_client: AsyncIOMotorClient[Any] mongo_client: AsyncIOMotorClient[Any]
redis: Redis redis: Redis
async def build_auth_service(settings: Settings, rabbitmq_channel: AbstractChannel) -> AppContainer: async def build_auth_service(settings: Settings, rabbitmq_channel: AbstractChannel) -> AppContainer:
mongo_client: AsyncIOMotorClient[Any] = AsyncIOMotorClient(settings.mongo_uri) """Create repositories, adapters, policies, and the AuthService use-case object."""
mongo_client: AsyncIOMotorClient[Any] = AsyncIOMotorClient(
settings.mongo_uri,
serverSelectionTimeoutMS=2000,
)
db = mongo_client[settings.mongo_db_name] db = mongo_client[settings.mongo_db_name]
users = MongoUserRepository(db) users = MongoUserRepository(db)
refresh_sessions = MongoRefreshSessionRepository(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) redis = Redis.from_url(settings.redis_url, decode_responses=False)
await _wait_for_state_stores(users, refresh_sessions, redis, settings.admin_mobile)
otp_store = RedisOtpStore(redis) otp_store = RedisOtpStore(redis)
sms_publisher = RabbitMqSmsPublisher(rabbitmq_channel) sms_publisher = RabbitMqSmsPublisher(rabbitmq_channel)
token_codec = JwtTokenCodec( token_codec = JwtTokenCodec(
@@ -58,3 +70,30 @@ async def build_auth_service(settings: Settings, rabbitmq_channel: AbstractChann
), ),
) )
return AppContainer(auth_service=auth_service, mongo_client=mongo_client, redis=redis) return AppContainer(auth_service=auth_service, mongo_client=mongo_client, redis=redis)
async def _wait_for_state_stores(
users: MongoUserRepository,
refresh_sessions: MongoRefreshSessionRepository,
redis: Redis,
admin_mobile: str,
) -> None:
"""Wait for MongoDB and Redis before exposing the gRPC server."""
for attempt in range(1, STARTUP_RETRY_ATTEMPTS + 1):
try:
await users.create_indexes()
await refresh_sessions.create_indexes()
await users.ensure_admin_user(admin_mobile)
await redis.ping()
return
except Exception:
if attempt >= STARTUP_RETRY_ATTEMPTS:
logger.exception("State stores did not become ready during auth-service startup")
raise
logger.info(
"Waiting for MongoDB/Redis before auth-service startup attempt=%s/%s",
attempt,
STARTUP_RETRY_ATTEMPTS,
)
await asyncio.sleep(STARTUP_RETRY_DELAY_SECONDS)

View File

@@ -7,20 +7,30 @@ logger = logging.getLogger(__name__)
class DebugSmsStore(Protocol): class DebugSmsStore(Protocol):
async def setex(self, name: str, time: int, value: str) -> object: ... """Minimal Redis-like store used by the local debug SMS provider."""
async def setex(self, name: str, time: int, value: str) -> object:
"""Store a value with a TTL using the Redis-compatible signature."""
...
class DebugSmsClient(SmsClient): class DebugSmsClient(SmsClient):
"""Local-only SMS strategy that stores the latest OTP for demo retrieval."""
def __init__(self, store: DebugSmsStore, ttl_seconds: int) -> None: def __init__(self, store: DebugSmsStore, ttl_seconds: int) -> None:
"""Configure the Redis-like store and short OTP debug retention."""
self._store = store self._store = store
self._ttl_seconds = ttl_seconds self._ttl_seconds = ttl_seconds
async def send_otp(self, mobile: str, code: str, template: str) -> None: async def send_otp(self, mobile: str, code: str, template: str) -> None:
"""Persist the OTP under a short-lived debug key instead of sending real SMS."""
key = debug_sms_key(mobile) key = debug_sms_key(mobile)
await self._store.setex(key, self._ttl_seconds, code) await self._store.setex(key, self._ttl_seconds, code)
logger.info("Debug SMS stored for mobile=%s template=%s", mobile, template) logger.info("Debug SMS stored for mobile=%s template=%s", mobile, template)
def debug_sms_key(mobile: str) -> str: def debug_sms_key(mobile: str) -> str:
return f"debug:sms:last:{mobile}" """Return the Redis key used by the demo UI to read the latest local OTP."""
return f"debug:sms:last:{mobile}"

View File

@@ -10,17 +10,22 @@ logger = logging.getLogger(__name__)
class KavenegarSmsClient(SmsClient): class KavenegarSmsClient(SmsClient):
"""Kavenegar verify/lookup implementation of the SMS provider strategy."""
def __init__( def __init__(
self, self,
api_key: str, api_key: str,
timeout_seconds: float = 10.0, timeout_seconds: float = 10.0,
transport: httpx.AsyncBaseTransport | None = None, transport: httpx.AsyncBaseTransport | None = None,
) -> None: ) -> None:
"""Configure credentials, timeout, and optional test transport."""
self._api_key = api_key self._api_key = api_key
self._timeout_seconds = timeout_seconds self._timeout_seconds = timeout_seconds
self._transport = transport self._transport = transport
async def send_otp(self, mobile: str, code: str, template: str) -> None: async def send_otp(self, mobile: str, code: str, template: str) -> None:
"""Send an OTP through Kavenegar and raise on transport or API failure."""
url = f"https://api.kavenegar.com/v1/{self._api_key}/verify/lookup.json" url = f"https://api.kavenegar.com/v1/{self._api_key}/verify/lookup.json"
payload = {"receptor": mobile, "template": template, "token": code, "type": "sms"} payload = {"receptor": mobile, "template": template, "token": code, "type": "sms"}
try: try:

View File

@@ -10,10 +10,12 @@ from gapido_auth.domain.ports import RefreshSessionRepository, UserRepository
def _now() -> datetime: def _now() -> datetime:
"""Return timezone-aware UTC now for Mongo document timestamps."""
return datetime.now(UTC) return datetime.now(UTC)
def _user_from_doc(doc: dict[str, Any]) -> User: def _user_from_doc(doc: dict[str, Any]) -> User:
"""Map a Mongo user document into the domain entity."""
return User( return User(
id=str(doc["_id"]), id=str(doc["_id"]),
mobile=str(doc["mobile"]), mobile=str(doc["mobile"]),
@@ -25,6 +27,7 @@ def _user_from_doc(doc: dict[str, Any]) -> User:
def _session_from_doc(doc: dict[str, Any]) -> RefreshSession: def _session_from_doc(doc: dict[str, Any]) -> RefreshSession:
"""Map a Mongo refresh-session document into the domain entity."""
return RefreshSession( return RefreshSession(
id=str(doc["_id"]), id=str(doc["_id"]),
user_id=str(doc["user_id"]), user_id=str(doc["user_id"]),
@@ -37,24 +40,35 @@ def _session_from_doc(doc: dict[str, Any]) -> RefreshSession:
class MongoUserRepository(UserRepository): class MongoUserRepository(UserRepository):
"""MongoDB adapter for user documents and admin bootstrap."""
def __init__(self, db: AsyncIOMotorDatabase[Any]) -> None: def __init__(self, db: AsyncIOMotorDatabase[Any]) -> None:
"""Bind the repository to the users collection."""
self._collection = db.users self._collection = db.users
async def create_indexes(self) -> None: async def create_indexes(self) -> None:
"""Create indexes required for unique mobile lookup and role scans."""
await self._collection.create_index([("mobile", ASCENDING)], unique=True) await self._collection.create_index([("mobile", ASCENDING)], unique=True)
await self._collection.create_index([("role", ASCENDING)]) await self._collection.create_index([("role", ASCENDING)])
async def get_by_id(self, user_id: str) -> User | None: async def get_by_id(self, user_id: str) -> User | None:
"""Return a user by Mongo ObjectId string, or None for invalid/missing ids."""
if not ObjectId.is_valid(user_id): if not ObjectId.is_valid(user_id):
return None return None
doc = await self._collection.find_one({"_id": ObjectId(user_id)}) doc = await self._collection.find_one({"_id": ObjectId(user_id)})
return _user_from_doc(doc) if doc else None return _user_from_doc(doc) if doc else None
async def get_by_mobile(self, mobile: str) -> User | None: async def get_by_mobile(self, mobile: str) -> User | None:
"""Return the user registered for a mobile number if one exists."""
doc = await self._collection.find_one({"mobile": mobile}) doc = await self._collection.find_one({"mobile": mobile})
return _user_from_doc(doc) if doc else None return _user_from_doc(doc) if doc else None
async def get_or_create_user(self, mobile: str, role: Role = Role.USER) -> User: async def get_or_create_user(self, mobile: str, role: Role = Role.USER) -> User:
"""Create a default active user for a verified mobile, or return the existing one."""
now = _now() now = _now()
await self._collection.update_one( await self._collection.update_one(
{"mobile": mobile}, {"mobile": mobile},
@@ -75,6 +89,8 @@ class MongoUserRepository(UserRepository):
return user return user
async def ensure_admin_user(self, mobile: str) -> User: async def ensure_admin_user(self, mobile: str) -> User:
"""Idempotently seed or promote the configured admin mobile."""
now = _now() now = _now()
await self._collection.update_one( await self._collection.update_one(
{"mobile": mobile}, {"mobile": mobile},
@@ -91,15 +107,22 @@ class MongoUserRepository(UserRepository):
class MongoRefreshSessionRepository(RefreshSessionRepository): class MongoRefreshSessionRepository(RefreshSessionRepository):
"""MongoDB adapter for hashed refresh-token sessions."""
def __init__(self, db: AsyncIOMotorDatabase[Any]) -> None: def __init__(self, db: AsyncIOMotorDatabase[Any]) -> None:
"""Bind the repository to the refresh_sessions collection."""
self._collection = db.refresh_sessions self._collection = db.refresh_sessions
async def create_indexes(self) -> None: async def create_indexes(self) -> None:
"""Create indexes used for token lookup, user session scans, and revocation."""
await self._collection.create_index([("token_hash", ASCENDING)], unique=True) 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([("user_id", ASCENDING), ("expires_at", ASCENDING)])
await self._collection.create_index([("revoked_at", ASCENDING)]) await self._collection.create_index([("revoked_at", ASCENDING)])
async def create(self, user_id: str, token_hash: str, expires_at: datetime) -> RefreshSession: async def create(self, user_id: str, token_hash: str, expires_at: datetime) -> RefreshSession:
"""Persist a new active refresh session for a user."""
now = _now() now = _now()
result = await self._collection.insert_one( result = await self._collection.insert_one(
{ {
@@ -117,6 +140,8 @@ class MongoRefreshSessionRepository(RefreshSessionRepository):
return _session_from_doc(doc) return _session_from_doc(doc)
async def get_active_by_hash(self, token_hash: str, now: datetime) -> RefreshSession | None: async def get_active_by_hash(self, token_hash: str, now: datetime) -> RefreshSession | None:
"""Return a non-revoked, non-expired session by token hash."""
doc = await self._collection.find_one( doc = await self._collection.find_one(
{"token_hash": token_hash, "revoked_at": None, "expires_at": {"$gt": now}} {"token_hash": token_hash, "revoked_at": None, "expires_at": {"$gt": now}}
) )
@@ -125,6 +150,8 @@ class MongoRefreshSessionRepository(RefreshSessionRepository):
async def revoke( async def revoke(
self, token_hash: str, now: datetime, replaced_by_hash: str | None = None self, token_hash: str, now: datetime, replaced_by_hash: str | None = None
) -> None: ) -> None:
"""Mark a refresh session revoked, optionally linking its replacement hash."""
update: dict[str, Any] = {"revoked_at": now} update: dict[str, Any] = {"revoked_at": now}
if replaced_by_hash is not None: if replaced_by_hash is not None:
update["replaced_by_hash"] = replaced_by_hash update["replaced_by_hash"] = replaced_by_hash
@@ -132,4 +159,3 @@ class MongoRefreshSessionRepository(RefreshSessionRepository):
{"token_hash": token_hash, "revoked_at": None}, {"token_hash": token_hash, "revoked_at": None},
{"$set": update}, {"$set": update},
) )

View File

@@ -16,10 +16,14 @@ SMS_DLX = "gapido.sms.dlx"
async def connect_robust(url: str) -> AbstractRobustConnection: async def connect_robust(url: str) -> AbstractRobustConnection:
"""Open a reconnecting RabbitMQ connection for publishers and workers."""
return await aio_pika.connect_robust(url) return await aio_pika.connect_robust(url)
async def declare_sms_topology(channel: AbstractChannel) -> tuple[Any, AbstractQueue, Any]: async def declare_sms_topology(channel: AbstractChannel) -> tuple[Any, AbstractQueue, Any]:
"""Declare durable SMS exchange, queue, and dead-letter queue topology."""
exchange = await channel.declare_exchange( exchange = await channel.declare_exchange(
SMS_EXCHANGE, aio_pika.ExchangeType.DIRECT, durable=True SMS_EXCHANGE, aio_pika.ExchangeType.DIRECT, durable=True
) )
@@ -36,11 +40,16 @@ async def declare_sms_topology(channel: AbstractChannel) -> tuple[Any, AbstractQ
class RabbitMqSmsPublisher(SmsPublisher): class RabbitMqSmsPublisher(SmsPublisher):
"""RabbitMQ publisher that sends durable OTP SMS jobs to the worker queue."""
def __init__(self, channel: AbstractChannel) -> None: def __init__(self, channel: AbstractChannel) -> None:
"""Store the channel used for SMS job publishing."""
self._channel = channel self._channel = channel
self._exchange: Any | None = None self._exchange: Any | None = None
async def publish(self, job: SmsJob) -> None: async def publish(self, job: SmsJob) -> None:
"""Serialize and publish an OTP SMS job to the configured routing key."""
if self._exchange is None: if self._exchange is None:
self._exchange, _, _ = await declare_sms_topology(self._channel) self._exchange, _, _ = await declare_sms_topology(self._channel)

View File

@@ -7,10 +7,15 @@ from gapido_auth.domain.ports import OtpStore
class RedisOtpStore(OtpStore): class RedisOtpStore(OtpStore):
"""Redis adapter for OTP hashes, verification attempts, and request throttling."""
def __init__(self, redis: Redis) -> None: def __init__(self, redis: Redis) -> None:
"""Bind the store to an async Redis client."""
self._redis = redis self._redis = redis
async def allow_request(self, key: str, limit: int, window_seconds: int) -> bool: 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) count = await self._redis.incr(key)
if count == 1: if count == 1:
await self._redis.expire(key, window_seconds) await self._redis.expire(key, window_seconds)
@@ -24,6 +29,8 @@ class RedisOtpStore(OtpStore):
ttl_seconds: int, ttl_seconds: int,
max_attempts: int, max_attempts: int,
) -> None: ) -> None:
"""Store a hashed OTP and reset its attempt counter with the same TTL window."""
key = self._otp_key(mobile, purpose) key = self._otp_key(mobile, purpose)
attempts_key = self._attempts_key(mobile, purpose) attempts_key = self._attempts_key(mobile, purpose)
async with self._redis.pipeline(transaction=True) as pipe: async with self._redis.pipeline(transaction=True) as pipe:
@@ -34,6 +41,8 @@ class RedisOtpStore(OtpStore):
await pipe.execute() await pipe.execute()
async def verify_otp(self, mobile: str, purpose: str, candidate_hash: str) -> bool: 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) key = self._otp_key(mobile, purpose)
attempts_key = self._attempts_key(mobile, purpose) attempts_key = self._attempts_key(mobile, purpose)
@@ -60,8 +69,10 @@ class RedisOtpStore(OtpStore):
@staticmethod @staticmethod
def _otp_key(mobile: str, purpose: str) -> str: def _otp_key(mobile: str, purpose: str) -> str:
"""Return the Redis hash key for an OTP challenge."""
return f"otp:{mobile}:{purpose}" return f"otp:{mobile}:{purpose}"
@staticmethod @staticmethod
def _attempts_key(mobile: str, purpose: str) -> str: def _attempts_key(mobile: str, purpose: str) -> str:
"""Return the Redis counter key for OTP verification attempts."""
return f"otp-attempts:{mobile}:{purpose}" return f"otp-attempts:{mobile}:{purpose}"

View File

@@ -10,6 +10,8 @@ logger = logging.getLogger(__name__)
class SmsIrSmsClient(SmsClient): class SmsIrSmsClient(SmsClient):
"""SMS.ir verify API implementation of the SMS provider strategy."""
_endpoint = "https://api.sms.ir/v1/send/verify" _endpoint = "https://api.sms.ir/v1/send/verify"
def __init__( def __init__(
@@ -18,11 +20,14 @@ class SmsIrSmsClient(SmsClient):
timeout_seconds: float = 10.0, timeout_seconds: float = 10.0,
transport: httpx.AsyncBaseTransport | None = None, transport: httpx.AsyncBaseTransport | None = None,
) -> None: ) -> None:
"""Configure credentials, timeout, and optional test transport."""
self._api_key = api_key self._api_key = api_key
self._timeout_seconds = timeout_seconds self._timeout_seconds = timeout_seconds
self._transport = transport self._transport = transport
async def send_otp(self, mobile: str, code: str, template: str) -> None: async def send_otp(self, mobile: str, code: str, template: str) -> None:
"""Send an OTP through SMS.ir using the configured verify template id."""
headers = { headers = {
"Content-Type": "application/json", "Content-Type": "application/json",
"Accept": "application/json", "Accept": "application/json",
@@ -53,4 +58,3 @@ class SmsIrSmsClient(SmsClient):
raise ExternalServiceError("SMS.ir API error") raise ExternalServiceError("SMS.ir API error")
logger.info("OTP SMS sent successfully to mobile=%s via SMS.ir", mobile) logger.info("OTP SMS sent successfully to mobile=%s via SMS.ir", mobile)

View File

@@ -12,6 +12,8 @@ def create_sms_client(
transport: httpx.AsyncBaseTransport | None = None, transport: httpx.AsyncBaseTransport | None = None,
debug_store: DebugSmsStore | None = None, debug_store: DebugSmsStore | None = None,
) -> SmsClient: ) -> SmsClient:
"""Build the configured SMS provider strategy for the worker process."""
match settings.sms_provider: match settings.sms_provider:
case "kavenegar": case "kavenegar":
return KavenegarSmsClient(settings.kavenegar_api_key, transport=transport) return KavenegarSmsClient(settings.kavenegar_api_key, transport=transport)
@@ -24,6 +26,8 @@ def create_sms_client(
def get_sms_template(settings: Settings) -> str: def get_sms_template(settings: Settings) -> str:
"""Return the provider-specific OTP template identifier used in queued SMS jobs."""
match settings.sms_provider: match settings.sms_provider:
case "kavenegar": case "kavenegar":
return settings.kavenegar_login_template return settings.kavenegar_login_template

View File

@@ -27,6 +27,8 @@ async def handle_message(
exchange: Any, exchange: Any,
dlx: Any, dlx: Any,
) -> None: ) -> None:
"""Process one SMS job, retry bounded failures, and dead-letter permanent failures."""
async with message.process(ignore_processed=True, requeue=False): async with message.process(ignore_processed=True, requeue=False):
payload = json.loads(message.body.decode()) payload = json.loads(message.body.decode())
job = SmsJob(**payload) job = SmsJob(**payload)
@@ -48,6 +50,7 @@ async def handle_message(
) )
return return
# Re-publish instead of requeueing indefinitely so retries remain bounded.
logger.warning( logger.warning(
"SMS job failed for mobile=%s retry=%s", job.mobile, retry_count + 1, exc_info=True "SMS job failed for mobile=%s retry=%s", job.mobile, retry_count + 1, exc_info=True
) )
@@ -63,6 +66,8 @@ async def handle_message(
async def main() -> None: async def main() -> None:
"""Start the RabbitMQ consumer and bind it to the configured SMS strategy."""
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
settings = get_settings() settings = get_settings()
connection = await connect_robust(settings.rabbitmq_url) connection = await connect_robust(settings.rabbitmq_url)

View File

@@ -4,6 +4,7 @@ from grpc_tools import protoc
def main() -> None: def main() -> None:
"""Regenerate Python gRPC stubs from the checked-in auth protobuf file."""
root = Path(__file__).resolve().parents[3] root = Path(__file__).resolve().parents[3]
proto_root = root / "proto" proto_root = root / "proto"
src_root = root / "src" src_root = root / "src"
@@ -23,4 +24,3 @@ def main() -> None:
if __name__ == "__main__": if __name__ == "__main__":
main() main()

View File

@@ -23,10 +23,15 @@ logger = logging.getLogger(__name__)
class AuthGrpcServicer(auth_pb2_grpc.AuthServiceServicer): class AuthGrpcServicer(auth_pb2_grpc.AuthServiceServicer):
"""gRPC transport adapter that maps protobuf calls to AuthService use cases."""
def __init__(self, auth_service: AuthService) -> None: def __init__(self, auth_service: AuthService) -> None:
"""Bind the servicer to the application auth service."""
self._auth_service = auth_service self._auth_service = auth_service
async def RequestOtp(self, request, context): # type: ignore[no-untyped-def] async def RequestOtp(self, request, context): # type: ignore[no-untyped-def]
"""Handle public OTP request calls."""
try: try:
await self._auth_service.request_otp( await self._auth_service.request_otp(
mobile=request.mobile, mobile=request.mobile,
@@ -38,6 +43,8 @@ class AuthGrpcServicer(auth_pb2_grpc.AuthServiceServicer):
await _abort_for_app_error(context, exc) await _abort_for_app_error(context, exc)
async def VerifyOtp(self, request, context): # type: ignore[no-untyped-def] async def VerifyOtp(self, request, context): # type: ignore[no-untyped-def]
"""Handle public OTP verification and token issuance calls."""
try: try:
token_pair = await self._auth_service.verify_otp( token_pair = await self._auth_service.verify_otp(
mobile=request.mobile, mobile=request.mobile,
@@ -49,6 +56,8 @@ class AuthGrpcServicer(auth_pb2_grpc.AuthServiceServicer):
await _abort_for_app_error(context, exc) await _abort_for_app_error(context, exc)
async def RefreshToken(self, request, context): # type: ignore[no-untyped-def] async def RefreshToken(self, request, context): # type: ignore[no-untyped-def]
"""Handle refresh-token rotation calls."""
try: try:
token_pair = await self._auth_service.refresh_token(request.refresh_token) token_pair = await self._auth_service.refresh_token(request.refresh_token)
return _token_response(token_pair) return _token_response(token_pair)
@@ -56,6 +65,8 @@ class AuthGrpcServicer(auth_pb2_grpc.AuthServiceServicer):
await _abort_for_app_error(context, exc) await _abort_for_app_error(context, exc)
async def RevokeRefreshToken(self, request, context): # type: ignore[no-untyped-def] async def RevokeRefreshToken(self, request, context): # type: ignore[no-untyped-def]
"""Handle authenticated refresh-token revocation calls."""
try: try:
token = _extract_bearer_token(context.invocation_metadata()) token = _extract_bearer_token(context.invocation_metadata())
await self._auth_service.get_authenticated_user(token) await self._auth_service.get_authenticated_user(token)
@@ -65,9 +76,13 @@ class AuthGrpcServicer(auth_pb2_grpc.AuthServiceServicer):
await _abort_for_app_error(context, exc) await _abort_for_app_error(context, exc)
async def PublicPing(self, request, context): # type: ignore[no-untyped-def] async def PublicPing(self, request, context): # type: ignore[no-untyped-def]
"""Return a public response without authentication."""
return auth_pb2.PingResponse(message="public ok") return auth_pb2.PingResponse(message="public ok")
async def UserOnly(self, request, context): # type: ignore[no-untyped-def] async def UserOnly(self, request, context): # type: ignore[no-untyped-def]
"""Return a response for any active authenticated user."""
try: try:
user = await self._auth_service.get_authenticated_user( user = await self._auth_service.get_authenticated_user(
_extract_bearer_token(context.invocation_metadata()) _extract_bearer_token(context.invocation_metadata())
@@ -77,6 +92,8 @@ class AuthGrpcServicer(auth_pb2_grpc.AuthServiceServicer):
await _abort_for_app_error(context, exc) await _abort_for_app_error(context, exc)
async def AdminOnly(self, request, context): # type: ignore[no-untyped-def] async def AdminOnly(self, request, context): # type: ignore[no-untyped-def]
"""Return a response only for authenticated admin users."""
try: try:
user = await self._auth_service.require_role( user = await self._auth_service.require_role(
_extract_bearer_token(context.invocation_metadata()), Role.ADMIN _extract_bearer_token(context.invocation_metadata()), Role.ADMIN
@@ -87,6 +104,8 @@ class AuthGrpcServicer(auth_pb2_grpc.AuthServiceServicer):
def _token_response(token_pair: TokenPair): # type: ignore[no-untyped-def] def _token_response(token_pair: TokenPair): # type: ignore[no-untyped-def]
"""Convert an application token pair into a protobuf response."""
return auth_pb2.TokenResponse( return auth_pb2.TokenResponse(
access_token=token_pair.access_token, access_token=token_pair.access_token,
refresh_token=token_pair.refresh_token, refresh_token=token_pair.refresh_token,
@@ -97,10 +116,14 @@ def _token_response(token_pair: TokenPair): # type: ignore[no-untyped-def]
def _protected_response(user: User, message: str): # type: ignore[no-untyped-def] def _protected_response(user: User, message: str): # type: ignore[no-untyped-def]
"""Convert an authenticated user into a protected-method protobuf response."""
return auth_pb2.ProtectedResponse(user_id=user.id, role=user.role.value, message=message) return auth_pb2.ProtectedResponse(user_id=user.id, role=user.role.value, message=message)
def _extract_bearer_token(metadata: Sequence[tuple[str, str]]) -> str: def _extract_bearer_token(metadata: Sequence[tuple[str, str]]) -> str:
"""Read bearer token metadata from a protected gRPC invocation."""
for key, value in metadata: for key, value in metadata:
if key.lower() == "authorization" and value.startswith("Bearer "): if key.lower() == "authorization" and value.startswith("Bearer "):
return value.removeprefix("Bearer ").strip() return value.removeprefix("Bearer ").strip()
@@ -108,6 +131,8 @@ def _extract_bearer_token(metadata: Sequence[tuple[str, str]]) -> str:
async def _abort_for_app_error(context: grpc.aio.ServicerContext, exc: AppError) -> None: async def _abort_for_app_error(context: grpc.aio.ServicerContext, exc: AppError) -> None:
"""Translate expected application errors into meaningful gRPC status codes."""
if isinstance(exc, AuthenticationError | InvalidOtp | OtpExpired | OtpAttemptsExceeded): if isinstance(exc, AuthenticationError | InvalidOtp | OtpExpired | OtpAttemptsExceeded):
await context.abort(grpc.StatusCode.UNAUTHENTICATED, str(exc)) await context.abort(grpc.StatusCode.UNAUTHENTICATED, str(exc))
if isinstance(exc, InactiveUser): if isinstance(exc, InactiveUser):

View File

@@ -16,6 +16,7 @@ logger = logging.getLogger(__name__)
async def serve() -> None: async def serve() -> None:
"""Start the async gRPC auth server and own infrastructure lifecycle."""
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
settings = get_settings() settings = get_settings()
rabbitmq = await connect_robust(settings.rabbitmq_url) rabbitmq = await connect_robust(settings.rabbitmq_url)
@@ -26,6 +27,7 @@ async def serve() -> None:
auth_servicer = AuthGrpcServicer(container.auth_service) auth_servicer = AuthGrpcServicer(container.auth_service)
auth_pb2_grpc.add_AuthServiceServicer_to_server(auth_servicer, server) auth_pb2_grpc.add_AuthServiceServicer_to_server(auth_servicer, server)
# Health and reflection make the service easier to inspect with grpcurl.
health_servicer = health.HealthServicer() health_servicer = health.HealthServicer()
health_pb2_grpc.add_HealthServicer_to_server(health_servicer, server) health_pb2_grpc.add_HealthServicer_to_server(health_servicer, server)
service_names = ( service_names = (

View File

@@ -20,46 +20,73 @@ T = TypeVar("T")
class AuthClient(Protocol): class AuthClient(Protocol):
async def request_otp(self, mobile: str, purpose: str) -> dict[str, bool]: ... """Minimal client contract used by FastAPI routes and tests."""
async def verify_otp(self, mobile: str, code: str, purpose: str) -> DemoTokenResponse: ... async def request_otp(self, mobile: str, purpose: str) -> dict[str, bool]:
"""Ask the auth service to create and dispatch an OTP."""
...
async def refresh_token(self, refresh_token: str) -> DemoTokenResponse: ... async def verify_otp(self, mobile: str, code: str, purpose: str) -> DemoTokenResponse:
"""Verify an OTP and return the token pair produced by auth-service."""
...
async def refresh_token(self, refresh_token: str) -> DemoTokenResponse:
"""Rotate a refresh token through auth-service."""
...
async def revoke_refresh_token( async def revoke_refresh_token(
self, access_token: str, refresh_token: str self, access_token: str, refresh_token: str
) -> dict[str, bool]: ... ) -> dict[str, bool]:
"""Revoke a refresh session using bearer-token metadata."""
...
async def public_ping(self) -> dict[str, str]: ... async def public_ping(self) -> dict[str, str]:
"""Call the public demo RPC without authentication."""
...
async def user_only(self, access_token: str) -> dict[str, str]: ... async def user_only(self, access_token: str) -> dict[str, str]:
"""Call the user-protected demo RPC with an access token."""
...
async def admin_only(self, access_token: str) -> dict[str, str]: ... async def admin_only(self, access_token: str) -> dict[str, str]:
"""Call the admin-protected demo RPC with an access token."""
...
class RequestOtpBody(BaseModel): class RequestOtpBody(BaseModel):
"""HTTP body for starting the mobile OTP login flow."""
mobile: str = Field(min_length=10, max_length=16) mobile: str = Field(min_length=10, max_length=16)
purpose: str = "login" purpose: str = "login"
class VerifyOtpBody(RequestOtpBody): class VerifyOtpBody(RequestOtpBody):
"""HTTP body for verifying a received OTP code."""
code: str = Field(min_length=6, max_length=6) code: str = Field(min_length=6, max_length=6)
class RefreshBody(BaseModel): class RefreshBody(BaseModel):
"""HTTP body carrying the opaque refresh token."""
refresh_token: str = Field(min_length=20) refresh_token: str = Field(min_length=20)
class TokenBody(BaseModel): class TokenBody(BaseModel):
"""HTTP body carrying a bearer access token for demo actions."""
access_token: str = Field(min_length=20) access_token: str = Field(min_length=20)
class RevokeBody(TokenBody): class RevokeBody(TokenBody):
"""HTTP body for revoking the current refresh session."""
refresh_token: str = Field(min_length=20) refresh_token: str = Field(min_length=20)
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]: async def lifespan(app: FastAPI) -> AsyncIterator[None]:
"""Create and close shared outbound clients for the demo service."""
settings = get_settings() settings = get_settings()
app.state.settings = settings app.state.settings = settings
app.state.auth_client = AuthGrpcClient(settings.auth_grpc_target) app.state.auth_client = AuthGrpcClient(settings.auth_grpc_target)
@@ -81,24 +108,29 @@ app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
def get_auth_client(request: Request) -> AuthClient: def get_auth_client(request: Request) -> AuthClient:
"""Resolve the configured auth-service client from application state."""
return cast(AuthClient, request.app.state.auth_client) return cast(AuthClient, request.app.state.auth_client)
def get_app_settings(request: Request) -> Settings: def get_app_settings(request: Request) -> Settings:
"""Resolve immutable runtime settings for request handlers."""
return cast(Settings, request.app.state.settings) return cast(Settings, request.app.state.settings)
def get_debug_redis(request: Request) -> Redis | None: def get_debug_redis(request: Request) -> Redis | None:
"""Resolve the optional Redis client used only by local debug OTP mode."""
return cast(Redis | None, request.app.state.debug_redis) return cast(Redis | None, request.app.state.debug_redis)
@app.get("/") @app.get("/")
async def index() -> FileResponse: async def index() -> FileResponse:
"""Serve the single-page browser demo."""
return FileResponse(STATIC_DIR / "index.html") return FileResponse(STATIC_DIR / "index.html")
@app.get("/healthz") @app.get("/healthz")
async def healthz() -> dict[str, str]: async def healthz() -> dict[str, str]:
"""Return a lightweight readiness response for Compose and Caddy checks."""
return {"status": "ok"} return {"status": "ok"}
@@ -107,6 +139,7 @@ async def request_otp(
body: RequestOtpBody, body: RequestOtpBody,
client: Annotated[AuthClient, Depends(get_auth_client)], client: Annotated[AuthClient, Depends(get_auth_client)],
) -> dict[str, object]: ) -> dict[str, object]:
"""Forward an OTP request from the browser to the gRPC auth service."""
return cast( return cast(
dict[str, object], dict[str, object],
await _call_grpc(lambda: client.request_otp(body.mobile, body.purpose)), await _call_grpc(lambda: client.request_otp(body.mobile, body.purpose)),
@@ -118,6 +151,7 @@ async def verify_otp(
body: VerifyOtpBody, body: VerifyOtpBody,
client: Annotated[AuthClient, Depends(get_auth_client)], client: Annotated[AuthClient, Depends(get_auth_client)],
) -> dict[str, object]: ) -> dict[str, object]:
"""Verify an OTP and return token data to the browser demo."""
response = await _call_grpc(lambda: client.verify_otp(body.mobile, body.code, body.purpose)) response = await _call_grpc(lambda: client.verify_otp(body.mobile, body.code, body.purpose))
return cast(dict[str, object], asdict(response)) return cast(dict[str, object], asdict(response))
@@ -127,6 +161,7 @@ async def refresh_token(
body: RefreshBody, body: RefreshBody,
client: Annotated[AuthClient, Depends(get_auth_client)], client: Annotated[AuthClient, Depends(get_auth_client)],
) -> dict[str, object]: ) -> dict[str, object]:
"""Rotate a refresh token and return a new token pair."""
response = await _call_grpc(lambda: client.refresh_token(body.refresh_token)) response = await _call_grpc(lambda: client.refresh_token(body.refresh_token))
return cast(dict[str, object], asdict(response)) return cast(dict[str, object], asdict(response))
@@ -136,6 +171,7 @@ async def revoke_refresh_token(
body: RevokeBody, body: RevokeBody,
client: Annotated[AuthClient, Depends(get_auth_client)], client: Annotated[AuthClient, Depends(get_auth_client)],
) -> dict[str, object]: ) -> dict[str, object]:
"""Revoke the current refresh session through auth-service."""
return cast( return cast(
dict[str, object], dict[str, object],
await _call_grpc( await _call_grpc(
@@ -146,6 +182,7 @@ async def revoke_refresh_token(
@app.post("/api/demo/public") @app.post("/api/demo/public")
async def public_demo(client: Annotated[AuthClient, Depends(get_auth_client)]) -> dict[str, object]: async def public_demo(client: Annotated[AuthClient, Depends(get_auth_client)]) -> dict[str, object]:
"""Call the public gRPC endpoint to prove unauthenticated access."""
return cast(dict[str, object], await _call_grpc(client.public_ping)) return cast(dict[str, object], await _call_grpc(client.public_ping))
@@ -154,6 +191,7 @@ async def user_demo(
body: TokenBody, body: TokenBody,
client: Annotated[AuthClient, Depends(get_auth_client)], client: Annotated[AuthClient, Depends(get_auth_client)],
) -> dict[str, object]: ) -> dict[str, object]:
"""Call the user-protected gRPC endpoint with bearer metadata."""
return cast(dict[str, object], await _call_grpc(lambda: client.user_only(body.access_token))) return cast(dict[str, object], await _call_grpc(lambda: client.user_only(body.access_token)))
@@ -162,6 +200,7 @@ async def admin_demo(
body: TokenBody, body: TokenBody,
client: Annotated[AuthClient, Depends(get_auth_client)], client: Annotated[AuthClient, Depends(get_auth_client)],
) -> dict[str, object]: ) -> dict[str, object]:
"""Call the admin-protected gRPC endpoint with bearer metadata."""
return cast(dict[str, object], await _call_grpc(lambda: client.admin_only(body.access_token))) return cast(dict[str, object], await _call_grpc(lambda: client.admin_only(body.access_token)))
@@ -171,6 +210,7 @@ async def debug_otp(
settings: Annotated[Settings, Depends(get_app_settings)], settings: Annotated[Settings, Depends(get_app_settings)],
redis: Annotated[Redis | None, Depends(get_debug_redis)], redis: Annotated[Redis | None, Depends(get_debug_redis)],
) -> dict[str, str | None]: ) -> dict[str, str | None]:
"""Expose the last debug OTP in local demo mode only."""
if not settings.demo_enable_debug_otp or redis is None: if not settings.demo_enable_debug_otp or redis is None:
raise HTTPException(status_code=404, detail="debug OTP endpoint is disabled") raise HTTPException(status_code=404, detail="debug OTP endpoint is disabled")
code = await redis.get(debug_sms_key(mobile)) code = await redis.get(debug_sms_key(mobile))
@@ -178,6 +218,7 @@ async def debug_otp(
async def _call_grpc(call: Callable[[], Awaitable[T]]) -> T: async def _call_grpc(call: Callable[[], Awaitable[T]]) -> T:
"""Execute a gRPC call and translate transport errors to HTTP errors."""
try: try:
return await call() return await call()
except grpc.aio.AioRpcError as exc: except grpc.aio.AioRpcError as exc:
@@ -187,6 +228,7 @@ async def _call_grpc(call: Callable[[], Awaitable[T]]) -> T:
def _grpc_to_http_status(code: grpc.StatusCode) -> int: def _grpc_to_http_status(code: grpc.StatusCode) -> int:
"""Map auth-service gRPC status codes to browser-friendly HTTP statuses."""
match code: match code:
case grpc.StatusCode.INVALID_ARGUMENT: case grpc.StatusCode.INVALID_ARGUMENT:
return 400 return 400

View File

@@ -9,6 +9,8 @@ from gapido_auth.generated import auth_pb2, auth_pb2_grpc
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class DemoTokenResponse: class DemoTokenResponse:
"""Token payload shape returned by the demo BFF to the browser."""
access_token: str access_token: str
refresh_token: str refresh_token: str
token_type: str token_type: str
@@ -17,32 +19,40 @@ class DemoTokenResponse:
class AuthGrpcClient: class AuthGrpcClient:
"""Thin async gRPC client used by the FastAPI demo service."""
def __init__(self, target: str) -> None: def __init__(self, target: str) -> None:
"""Open an async channel to auth-service."""
self._channel = grpc.aio.insecure_channel(target) self._channel = grpc.aio.insecure_channel(target)
self._stub = auth_pb2_grpc.AuthServiceStub(self._channel) self._stub = auth_pb2_grpc.AuthServiceStub(self._channel)
async def close(self) -> None: async def close(self) -> None:
"""Close the underlying gRPC channel during FastAPI shutdown."""
await self._channel.close() await self._channel.close()
async def request_otp(self, mobile: str, purpose: str) -> dict[str, bool]: async def request_otp(self, mobile: str, purpose: str) -> dict[str, bool]:
"""Forward an OTP request and normalize the protobuf response."""
response = await self._stub.RequestOtp( response = await self._stub.RequestOtp(
auth_pb2.RequestOtpRequest(mobile=mobile, purpose=purpose) auth_pb2.RequestOtpRequest(mobile=mobile, purpose=purpose)
) )
return {"accepted": bool(response.accepted)} return {"accepted": bool(response.accepted)}
async def verify_otp(self, mobile: str, code: str, purpose: str) -> DemoTokenResponse: async def verify_otp(self, mobile: str, code: str, purpose: str) -> DemoTokenResponse:
"""Verify an OTP and convert the protobuf token response."""
response = await self._stub.VerifyOtp( response = await self._stub.VerifyOtp(
auth_pb2.VerifyOtpRequest(mobile=mobile, code=code, purpose=purpose) auth_pb2.VerifyOtpRequest(mobile=mobile, code=code, purpose=purpose)
) )
return _token_response(response) return _token_response(response)
async def refresh_token(self, refresh_token: str) -> DemoTokenResponse: async def refresh_token(self, refresh_token: str) -> DemoTokenResponse:
"""Refresh and rotate an opaque refresh token."""
response = await self._stub.RefreshToken( response = await self._stub.RefreshToken(
auth_pb2.RefreshTokenRequest(refresh_token=refresh_token) auth_pb2.RefreshTokenRequest(refresh_token=refresh_token)
) )
return _token_response(response) return _token_response(response)
async def revoke_refresh_token(self, access_token: str, refresh_token: str) -> dict[str, bool]: async def revoke_refresh_token(self, access_token: str, refresh_token: str) -> dict[str, bool]:
"""Revoke a refresh token using access-token authorization metadata."""
response = await self._stub.RevokeRefreshToken( response = await self._stub.RevokeRefreshToken(
auth_pb2.RevokeRefreshTokenRequest(refresh_token=refresh_token), auth_pb2.RevokeRefreshTokenRequest(refresh_token=refresh_token),
metadata=_auth_metadata(access_token), metadata=_auth_metadata(access_token),
@@ -50,16 +60,19 @@ class AuthGrpcClient:
return {"revoked": bool(response.revoked)} return {"revoked": bool(response.revoked)}
async def public_ping(self) -> dict[str, str]: async def public_ping(self) -> dict[str, str]:
"""Call the public demonstration endpoint."""
response = await self._stub.PublicPing(auth_pb2.PingRequest()) response = await self._stub.PublicPing(auth_pb2.PingRequest())
return {"message": str(response.message)} return {"message": str(response.message)}
async def user_only(self, access_token: str) -> dict[str, str]: async def user_only(self, access_token: str) -> dict[str, str]:
"""Call the user-only demonstration endpoint."""
response = await self._stub.UserOnly( response = await self._stub.UserOnly(
auth_pb2.ProtectedRequest(), metadata=_auth_metadata(access_token) auth_pb2.ProtectedRequest(), metadata=_auth_metadata(access_token)
) )
return _protected_response(response) return _protected_response(response)
async def admin_only(self, access_token: str) -> dict[str, str]: async def admin_only(self, access_token: str) -> dict[str, str]:
"""Call the admin-only demonstration endpoint."""
response = await self._stub.AdminOnly( response = await self._stub.AdminOnly(
auth_pb2.ProtectedRequest(), metadata=_auth_metadata(access_token) auth_pb2.ProtectedRequest(), metadata=_auth_metadata(access_token)
) )
@@ -67,10 +80,12 @@ class AuthGrpcClient:
def _auth_metadata(access_token: str) -> tuple[tuple[str, str], ...]: def _auth_metadata(access_token: str) -> tuple[tuple[str, str], ...]:
"""Build gRPC bearer metadata expected by auth-service."""
return (("authorization", f"Bearer {access_token}"),) return (("authorization", f"Bearer {access_token}"),)
def _token_response(response: Any) -> DemoTokenResponse: def _token_response(response: Any) -> DemoTokenResponse:
"""Convert a protobuf token message into a dataclass."""
return DemoTokenResponse( return DemoTokenResponse(
access_token=str(response.access_token), access_token=str(response.access_token),
refresh_token=str(response.refresh_token), refresh_token=str(response.refresh_token),
@@ -81,6 +96,7 @@ def _token_response(response: Any) -> DemoTokenResponse:
def _protected_response(response: Any) -> dict[str, str]: def _protected_response(response: Any) -> dict[str, str]:
"""Convert a protected protobuf response into a JSON-ready dict."""
return { return {
"user_id": str(response.user_id), "user_id": str(response.user_id),
"role": str(response.role), "role": str(response.role),