Compare commits

..

3 Commits

45 changed files with 1131 additions and 34 deletions

View File

@@ -73,6 +73,15 @@ pytest
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
- `RequestOtp`: public; creates a short-lived OTP and publishes an SMS job.

View File

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

View File

@@ -35,9 +35,9 @@ services:
start_period: 20s
depends_on:
mongo:
condition: service_started
condition: service_healthy
redis:
condition: service_started
condition: service_healthy
rabbitmq:
condition: service_healthy
@@ -54,6 +54,8 @@ services:
depends_on:
rabbitmq:
condition: service_healthy
redis:
condition: service_healthy
demo-app:
build: .
@@ -72,15 +74,26 @@ services:
auth-service:
condition: service_healthy
redis:
condition: service_started
condition: service_healthy
mongo:
image: mongo:7
volumes:
- mongo-data:/data/db
healthcheck:
test: ["CMD", "mongosh", "--quiet", "--eval", "db.adminCommand('ping').ok"]
interval: 5s
timeout: 5s
retries: 20
start_period: 10s
redis:
image: redis:7-alpine
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 5s
timeout: 5s
retries: 20
rabbitmq:
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: 36 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 49 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 46 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 27 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 51 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 18 KiB

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

@@ -0,0 +1,215 @@
from pathlib import Path
from diagrams import Cluster, Diagram, Edge
from diagrams.generic.blank import Blank
from diagrams.onprem.database import Mongodb
from diagrams.onprem.inmemory import Redis
from diagrams.onprem.network import Internet
from diagrams.onprem.queue import Rabbitmq
ROOT = Path(__file__).resolve().parents[2]
OUT = ROOT / "docs" / "assets" / "diagrams"
def graph_attr() -> dict[str, str]:
return {
"bgcolor": "transparent",
"pad": "0.35",
"ranksep": "0.8",
"nodesep": "0.55",
"fontname": "Inter",
}
def node_attr() -> dict[str, str]:
return {
"fontname": "Inter",
"fontsize": "13",
"style": "rounded,filled",
"fillcolor": "#111827",
"fontcolor": "#E5E7EB",
"color": "#334155",
}
def edge_attr() -> dict[str, str]:
return {"fontname": "Inter", "fontsize": "11", "color": "#64748B", "fontcolor": "#475569"}
def render_service_architecture() -> None:
with Diagram(
"Service Architecture",
filename=str(OUT / "service_architecture"),
show=False,
direction="LR",
graph_attr=graph_attr(),
node_attr=node_attr(),
edge_attr=edge_attr(),
):
browser = Internet("Browser")
demo = Blank("demo-app\nFastAPI UI")
auth = Blank("auth-service\ngRPC")
worker = Blank("sms-worker")
mongo = Mongodb("MongoDB")
redis = Redis("Redis")
rabbit = Rabbitmq("RabbitMQ")
providers = Blank("Kavenegar / SMS.ir")
browser >> Edge(label="HTTPS / local HTTP") >> demo
demo >> Edge(label="gRPC") >> auth
auth >> Edge(label="users + refresh sessions") >> mongo
auth >> Edge(label="OTP hash + rate limits") >> redis
auth >> Edge(label="SMS job") >> rabbit >> worker >> providers
def render_clean_architecture() -> None:
with Diagram(
"Clean Architecture Layers",
filename=str(OUT / "clean_architecture"),
show=False,
direction="TB",
graph_attr=graph_attr(),
node_attr=node_attr(),
edge_attr=edge_attr(),
):
with Cluster("Transport"):
grpc = Blank("gRPC servicer")
demo = Blank("Demo BFF")
with Cluster("Application"):
auth = Blank("AuthService\nuse cases")
security = Blank("Token + OTP helpers")
with Cluster("Domain"):
entities = Blank("Entities")
ports = Blank("Ports")
with Cluster("Infrastructure"):
mongo = Blank("Mongo repositories")
redis = Blank("Redis OTP store")
rabbit = Blank("RabbitMQ publisher")
sms = Blank("SMS strategies")
grpc >> auth
demo >> grpc
auth >> entities
auth >> ports
auth >> security
ports >> mongo
ports >> redis
ports >> rabbit
ports >> sms
def render_otp_flow() -> None:
with Diagram(
"OTP Login Flow",
filename=str(OUT / "otp_login_flow"),
show=False,
direction="LR",
graph_attr=graph_attr(),
node_attr=node_attr(),
edge_attr=edge_attr(),
):
user = Blank("User")
demo = Blank("demo-app")
auth = Blank("auth-service")
redis = Redis("Redis")
rabbit = Rabbitmq("RabbitMQ")
worker = Blank("sms-worker")
sms = Blank("SMS provider")
mongo = Mongodb("MongoDB")
user >> Edge(label="mobile") >> demo >> Edge(label="RequestOtp") >> auth
auth >> Edge(label="store OTP hash") >> redis
auth >> Edge(label="publish job") >> rabbit >> worker >> sms
user >> Edge(label="code") >> demo >> Edge(label="VerifyOtp") >> auth
auth >> Edge(label="compare hash") >> redis
auth >> Edge(label="upsert user + session") >> mongo
auth >> Edge(label="token pair") >> demo >> user
def render_refresh_flow() -> None:
with Diagram(
"Refresh Token Rotation",
filename=str(OUT / "refresh_rotation_flow"),
show=False,
direction="LR",
graph_attr=graph_attr(),
node_attr=node_attr(),
edge_attr=edge_attr(),
):
client = Blank("Client")
auth = Blank("auth-service")
mongo = Mongodb("MongoDB")
old = Blank("Old session\nrevoked")
new = Blank("New session\nactive")
client >> Edge(label="refresh token") >> auth
auth >> Edge(label="hash lookup") >> mongo
mongo >> Edge(label="active session") >> auth
auth >> old
auth >> new
new >> Edge(label="new token pair") >> client
def render_sms_strategy() -> None:
with Diagram(
"SMS Provider Strategy",
filename=str(OUT / "sms_provider_strategy"),
show=False,
direction="LR",
graph_attr=graph_attr(),
node_attr=node_attr(),
edge_attr=edge_attr(),
):
worker = Blank("sms-worker")
factory = Blank("create_sms_client")
port = Blank("SmsClient port")
kav = Blank("Kavenegar")
smsir = Blank("SMS.ir")
debug = Blank("Debug local")
worker >> factory >> port
port >> kav
port >> smsir
port >> debug
def render_production_deployment() -> None:
with Diagram(
"Production Deployment",
filename=str(OUT / "production_deployment"),
show=False,
direction="LR",
graph_attr=graph_attr(),
node_attr=node_attr(),
edge_attr=edge_attr(),
):
internet = Internet("Internet")
caddy = Blank("Caddy\nHTTPS")
demo = Blank("demo-app")
auth = Blank("auth-service")
private = Blank("Private Docker network")
mongo = Mongodb("MongoDB")
redis = Redis("Redis")
rabbit = Rabbitmq("RabbitMQ")
worker = Blank("sms-worker")
internet >> Edge(label="gapido.amiirkhl.ir") >> caddy >> demo >> auth
auth >> private
private >> mongo
private >> redis
private >> rabbit >> worker
def main() -> None:
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.

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

@@ -0,0 +1,103 @@
<!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 split">
<div>
<p class="eyebrow">Constraints</p>
<h2>What had to be proven</h2>
</div>
<ul class="tiles">
<li>gRPC auth boundary</li>
<li>secure OTP</li>
<li>refresh rotation</li>
<li>role-based access</li>
<li>async SMS delivery</li>
<li>production deployment path</li>
</ul>
</section>
<section class="slide diagram">
<p class="eyebrow">Solution shape</p>
<h2>Service architecture</h2>
<img src="../assets/diagrams/service_architecture.png" alt="Service architecture" />
</section>
<section class="slide diagram">
<p class="eyebrow">Code organization</p>
<h2>Clean architecture layers</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 without storing plaintext codes</h2>
<img src="../assets/diagrams/otp_login_flow.png" alt="OTP login flow" />
</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">Extensibility</p>
<h2>Selectable SMS providers</h2>
<img src="../assets/diagrams/sms_provider_strategy.png" alt="SMS provider strategy" />
</section>
<section class="slide diagram">
<p class="eyebrow">Deployment</p>
<h2>Only Caddy is public</h2>
<img src="../assets/diagrams/production_deployment.png" alt="Production deployment" />
</section>
<section class="slide split">
<div>
<p class="eyebrow">Reliability</p>
<h2>What makes it reviewable</h2>
</div>
<ul class="tiles">
<li>Mocked SMS providers</li>
<li>gRPC tests</li>
<li>RBAC coverage</li>
<li>Compose validation</li>
<li>Health checks</li>
<li>Postman guide</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 / 10</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]
dev = [
"diagrams==0.24.4",
"grpcio-tools==1.68.1",
"mypy==1.14.1",
"pytest==8.3.4",

View File

@@ -28,6 +28,8 @@ from gapido_auth.domain.ports import (
@dataclass(frozen=True, slots=True)
class AuthConfig:
"""Runtime policy values used by auth use cases."""
otp_secret: str
otp_ttl_seconds: int
otp_max_attempts: int
@@ -38,6 +40,8 @@ class AuthConfig:
class AuthService:
"""Application service coordinating OTP login, token rotation, and RBAC."""
def __init__(
self,
users: UserRepository,
@@ -47,6 +51,7 @@ class AuthService:
token_codec: JwtTokenCodec,
config: AuthConfig,
) -> None:
"""Wire repository, queue, OTP, and token ports for use-case execution."""
self._users = users
self._refresh_sessions = refresh_sessions
self._otp_store = otp_store
@@ -55,6 +60,8 @@ class AuthService:
self._config = config
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_purpose(purpose)
mobile_key = f"otp-request:mobile:{mobile}:{purpose}"
@@ -73,6 +80,7 @@ class AuthService:
raise RateLimitExceeded("too many OTP requests")
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)
await self._otp_store.store_otp(
mobile=mobile,
@@ -86,6 +94,8 @@ class AuthService:
)
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_purpose(purpose)
if not code.isdigit() or len(code) != 6:
@@ -100,6 +110,8 @@ class AuthService:
return await self._issue_token_pair(user)
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)
now = utc_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:
raise AuthenticationError("invalid refresh token")
# Rotation revokes the old token hash and persists a fresh session hash.
new_refresh_token = generate_refresh_token()
new_hash = hash_refresh_token(new_refresh_token)
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:
"""Revoke a refresh token session if it is still active."""
await self._refresh_sessions.revoke(hash_refresh_token(refresh_token), utc_now())
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)
user = await self._users.get_by_id(claims.user_id)
if user is None:
@@ -137,12 +154,16 @@ class AuthService:
return 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)
if user.role != role:
raise PermissionDenied("insufficient permissions")
return user
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:
raise InactiveUser("user is inactive")
@@ -160,11 +181,15 @@ class AuthService:
def _validate_mobile(mobile: str) -> None:
"""Validate the E.164-like mobile format accepted by the challenge service."""
normalized = mobile.removeprefix("+")
if not normalized.isdigit() or len(normalized) < 10 or len(normalized) > 15:
raise ValidationError("mobile must be an E.164-like phone number")
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():
raise ValidationError("purpose is invalid")

View File

@@ -11,37 +11,54 @@ from gapido_auth.domain.errors import AuthenticationError
def utc_now() -> datetime:
"""Return timezone-aware UTC time for token/session timestamps."""
return datetime.now(UTC)
def generate_otp_code() -> str:
"""Generate a cryptographically random six-digit OTP string."""
return f"{secrets.randbelow(1_000_000):06d}"
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()
return hmac.new(secret.encode(), message, hashlib.sha256).hexdigest()
def generate_refresh_token() -> str:
"""Generate an opaque refresh token suitable for returning to clients."""
return secrets.token_urlsafe(48)
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()
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:
"""Store signing configuration used for all access-token operations."""
self._secret_key = secret_key
self._issuer = issuer
self._access_ttl_seconds = access_ttl_seconds
@property
def access_ttl_seconds(self) -> int:
"""Return the configured access-token TTL exposed to clients."""
return self._access_ttl_seconds
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()
expires_at = now + timedelta(seconds=self._access_ttl_seconds)
payload = {
@@ -56,6 +73,8 @@ class JwtTokenCodec:
return jwt.encode(payload, self._secret_key, algorithm="HS256")
def decode_access_token(self, token: str) -> AccessClaims:
"""Validate an access JWT and return typed claims used by RBAC checks."""
try:
payload = jwt.decode(
token,
@@ -80,4 +99,3 @@ class JwtTokenCodec:
role=role,
expires_at=datetime.fromtimestamp(int(payload["exp"]), UTC),
)

View File

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

View File

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

View File

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

View File

@@ -5,29 +5,51 @@ from gapido_auth.domain.entities import RefreshSession, Role, SmsJob, User
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):
"""Persistence port for refresh-token session creation and rotation."""
async def create(
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(
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):
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(
self,
@@ -36,14 +58,26 @@ class OtpStore(Protocol):
otp_hash: str,
ttl_seconds: 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):
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):
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 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.sms_provider import get_sms_template
logger = logging.getLogger(__name__)
STARTUP_RETRY_ATTEMPTS = 30
STARTUP_RETRY_DELAY_SECONDS = 2
@dataclass(slots=True)
class AppContainer:
"""Runtime dependencies that need explicit shutdown after the gRPC server stops."""
auth_service: AuthService
mongo_client: AsyncIOMotorClient[Any]
redis: Redis
async def build_auth_service(settings: Settings, rabbitmq_channel: AbstractChannel) -> AppContainer:
mongo_client: AsyncIOMotorClient[Any] = AsyncIOMotorClient(settings.mongo_uri)
"""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]
users = MongoUserRepository(db)
refresh_sessions = MongoRefreshSessionRepository(db)
await users.create_indexes()
await refresh_sessions.create_indexes()
await users.ensure_admin_user(settings.admin_mobile)
redis = Redis.from_url(settings.redis_url, decode_responses=False)
await _wait_for_state_stores(users, refresh_sessions, redis, settings.admin_mobile)
otp_store = RedisOtpStore(redis)
sms_publisher = RabbitMqSmsPublisher(rabbitmq_channel)
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)
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):
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):
"""Local-only SMS strategy that stores the latest OTP for demo retrieval."""
def __init__(self, store: DebugSmsStore, ttl_seconds: int) -> None:
"""Configure the Redis-like store and short OTP debug retention."""
self._store = store
self._ttl_seconds = ttl_seconds
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)
await self._store.setex(key, self._ttl_seconds, code)
logger.info("Debug SMS stored for mobile=%s template=%s", mobile, template)
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):
"""Kavenegar verify/lookup implementation of the SMS provider strategy."""
def __init__(
self,
api_key: str,
timeout_seconds: float = 10.0,
transport: httpx.AsyncBaseTransport | None = None,
) -> None:
"""Configure credentials, timeout, and optional test transport."""
self._api_key = api_key
self._timeout_seconds = timeout_seconds
self._transport = transport
async def send_otp(self, mobile: str, code: str, template: str) -> None:
"""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"
payload = {"receptor": mobile, "template": template, "token": code, "type": "sms"}
try:

View File

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

View File

@@ -16,10 +16,14 @@ SMS_DLX = "gapido.sms.dlx"
async def connect_robust(url: str) -> AbstractRobustConnection:
"""Open a reconnecting RabbitMQ connection for publishers and workers."""
return await aio_pika.connect_robust(url)
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(
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):
"""RabbitMQ publisher that sends durable OTP SMS jobs to the worker queue."""
def __init__(self, channel: AbstractChannel) -> None:
"""Store the channel used for SMS job publishing."""
self._channel = channel
self._exchange: Any | None = 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:
self._exchange, _, _ = await declare_sms_topology(self._channel)

View File

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

View File

@@ -10,6 +10,8 @@ logger = logging.getLogger(__name__)
class SmsIrSmsClient(SmsClient):
"""SMS.ir verify API implementation of the SMS provider strategy."""
_endpoint = "https://api.sms.ir/v1/send/verify"
def __init__(
@@ -18,11 +20,14 @@ class SmsIrSmsClient(SmsClient):
timeout_seconds: float = 10.0,
transport: httpx.AsyncBaseTransport | None = None,
) -> None:
"""Configure credentials, timeout, and optional test transport."""
self._api_key = api_key
self._timeout_seconds = timeout_seconds
self._transport = transport
async def send_otp(self, mobile: str, code: str, template: str) -> None:
"""Send an OTP through SMS.ir using the configured verify template id."""
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
@@ -53,4 +58,3 @@ class SmsIrSmsClient(SmsClient):
raise ExternalServiceError("SMS.ir API error")
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,
debug_store: DebugSmsStore | None = None,
) -> SmsClient:
"""Build the configured SMS provider strategy for the worker process."""
match settings.sms_provider:
case "kavenegar":
return KavenegarSmsClient(settings.kavenegar_api_key, transport=transport)
@@ -24,6 +26,8 @@ def create_sms_client(
def get_sms_template(settings: Settings) -> str:
"""Return the provider-specific OTP template identifier used in queued SMS jobs."""
match settings.sms_provider:
case "kavenegar":
return settings.kavenegar_login_template

View File

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

View File

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

View File

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

View File

@@ -16,6 +16,7 @@ logger = logging.getLogger(__name__)
async def serve() -> None:
"""Start the async gRPC auth server and own infrastructure lifecycle."""
logging.basicConfig(level=logging.INFO)
settings = get_settings()
rabbitmq = await connect_robust(settings.rabbitmq_url)
@@ -26,6 +27,7 @@ async def serve() -> None:
auth_servicer = AuthGrpcServicer(container.auth_service)
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_pb2_grpc.add_HealthServicer_to_server(health_servicer, server)
service_names = (

View File

@@ -20,46 +20,73 @@ T = TypeVar("T")
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(
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):
"""HTTP body for starting the mobile OTP login flow."""
mobile: str = Field(min_length=10, max_length=16)
purpose: str = "login"
class VerifyOtpBody(RequestOtpBody):
"""HTTP body for verifying a received OTP code."""
code: str = Field(min_length=6, max_length=6)
class RefreshBody(BaseModel):
"""HTTP body carrying the opaque refresh token."""
refresh_token: str = Field(min_length=20)
class TokenBody(BaseModel):
"""HTTP body carrying a bearer access token for demo actions."""
access_token: str = Field(min_length=20)
class RevokeBody(TokenBody):
"""HTTP body for revoking the current refresh session."""
refresh_token: str = Field(min_length=20)
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
"""Create and close shared outbound clients for the demo service."""
settings = get_settings()
app.state.settings = settings
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:
"""Resolve the configured auth-service client from application state."""
return cast(AuthClient, request.app.state.auth_client)
def get_app_settings(request: Request) -> Settings:
"""Resolve immutable runtime settings for request handlers."""
return cast(Settings, request.app.state.settings)
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)
@app.get("/")
async def index() -> FileResponse:
"""Serve the single-page browser demo."""
return FileResponse(STATIC_DIR / "index.html")
@app.get("/healthz")
async def healthz() -> dict[str, str]:
"""Return a lightweight readiness response for Compose and Caddy checks."""
return {"status": "ok"}
@@ -107,6 +139,7 @@ async def request_otp(
body: RequestOtpBody,
client: Annotated[AuthClient, Depends(get_auth_client)],
) -> dict[str, object]:
"""Forward an OTP request from the browser to the gRPC auth service."""
return cast(
dict[str, object],
await _call_grpc(lambda: client.request_otp(body.mobile, body.purpose)),
@@ -118,6 +151,7 @@ async def verify_otp(
body: VerifyOtpBody,
client: Annotated[AuthClient, Depends(get_auth_client)],
) -> 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))
return cast(dict[str, object], asdict(response))
@@ -127,6 +161,7 @@ async def refresh_token(
body: RefreshBody,
client: Annotated[AuthClient, Depends(get_auth_client)],
) -> dict[str, object]:
"""Rotate a refresh token and return a new token pair."""
response = await _call_grpc(lambda: client.refresh_token(body.refresh_token))
return cast(dict[str, object], asdict(response))
@@ -136,6 +171,7 @@ async def revoke_refresh_token(
body: RevokeBody,
client: Annotated[AuthClient, Depends(get_auth_client)],
) -> dict[str, object]:
"""Revoke the current refresh session through auth-service."""
return cast(
dict[str, object],
await _call_grpc(
@@ -146,6 +182,7 @@ async def revoke_refresh_token(
@app.post("/api/demo/public")
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))
@@ -154,6 +191,7 @@ async def user_demo(
body: TokenBody,
client: Annotated[AuthClient, Depends(get_auth_client)],
) -> 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)))
@@ -162,6 +200,7 @@ async def admin_demo(
body: TokenBody,
client: Annotated[AuthClient, Depends(get_auth_client)],
) -> 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)))
@@ -171,6 +210,7 @@ async def debug_otp(
settings: Annotated[Settings, Depends(get_app_settings)],
redis: Annotated[Redis | None, Depends(get_debug_redis)],
) -> 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:
raise HTTPException(status_code=404, detail="debug OTP endpoint is disabled")
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:
"""Execute a gRPC call and translate transport errors to HTTP errors."""
try:
return await call()
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:
"""Map auth-service gRPC status codes to browser-friendly HTTP statuses."""
match code:
case grpc.StatusCode.INVALID_ARGUMENT:
return 400

View File

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