← Home
Technical Report · ASL Pipeline

REST APIs & Security

Complete endpoint documentation, authentication mechanisms, and security controls across all three FastAPI microservices.

Complete API Reference

Auth Service (authService:8001)

Method Path Auth Body Response Notes
POST /register none {username, password, role} {userId} Creates new user, hashes password with bcrypt
POST /login none {username, password} {accessToken, refreshToken} Returns 60-min access token + 7-day refresh token
POST /refresh none {refreshToken} {accessToken} Extends session
GET /me Bearer — {userId, role} Returns current user info
GET /health none — {status, service} Readiness check

Inference Service (inferenceService:8002)

Method Path Auth Query / Body Response Notes
POST /predict Bearer {landmarks, modelName} {sign, confidence, latencyMs, eventId, modelVersion} Runs inference, writes event to Delta
GET /models Bearer — [{name, version, stage, accuracy}] Lists registered MLflow models
GET /stats Bearer + analyst — {predictionsLastHour, latencyP95, topSigns} Aggregated metrics (requires analyst role)
GET /health none — {status, service} Readiness check

Agent Service (agentService:8003)

Method Path Auth Body Response Notes
POST /translate Bearer {signs: [{label, confidence}]} {gloss, english, trace, corrections} Runs Assembler + Translator agents
GET /health none — {status, service} Readiness check

Authentication & JWT Tokens

Scheme: HS256 (HMAC SHA-256, symmetric — shared secret across all services)

Token Structure:

{
  "userId": "user-123",
  "role": "user",
  "exp": 1717287600,
  "iat": 1717284000
}

Token Lifetime: 60 minutes (access) · 7 days (refresh)

Header Format:

Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

Validation (on each service):

def validate_token(token: str) -> dict:
    """Decode and validate JWT"""
    try:
        payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=["HS256"])
        if payload["exp"] < time.time():
            raise HTTPException(401, "Token expired")
        return payload
    except jwt.InvalidSignatureError:
        raise HTTPException(401, "Invalid signature")

Role-Based Access Control (RBAC)

Roles:

  • user — Can call /predict, /translate, /me
  • analyst — Can additionally call /stats (aggregated metrics)
  • admin — Reserved for future use

403 Forbidden is returned on insufficient role.

Enforcement:

def require_role(*allowed_roles):
    def decorator(func):
        @wraps(func)
        async def wrapper(request: Request, *args, **kwargs):
            token = request.headers.get("Authorization", "").replace("Bearer ", "")
            payload = validate_token(token)

            if payload["role"] not in allowed_roles:
                raise HTTPException(403, "Insufficient permissions")

            return await func(request, *args, **kwargs)
        return wrapper
    return decorator

@app.get("/stats")
@require_role("analyst", "admin")
async def get_stats():
    # Only accessible to analysts and admins
    ...

Security Controls

1. Input Validation

Landmark Payload (inferenceService):

class PredictRequest(BaseModel):
    landmarks: list[list[float]]  # (T, 54) or (T, 27, 2)
    modelName: str

    @validator('landmarks')
    def validate_landmarks(cls, v):
        # Max 200 frames
        if len(v) > 200:
            raise ValueError("Max 200 frames")

        # Each frame exactly 54 floats (27 keypoints × 2 coords)
        for frame in v:
            if len(frame) != 54:
                raise ValueError("Each frame must have 54 floats")

        return v

Pydantic validates every request body across every service.

2. Password Security

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

3. Rate Limiting

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post("/predict")
@limiter.limit("10/minute")  # Max 10 predictions per minute per IP
async def predict(request: Request, payload: PredictRequest):
    ...

slowapi middleware caps /predict to prevent abuse.

4. Error Handling

❌ Do NOT leak internal details to clients 
except Exception as e:
    return {"error": str(e)}  # e.g., "CUDA out of memory" — leaks internals
✓ Generic user-safe message + structured logging 
except Exception as e:
    logger.error("Prediction failed", extra={"error": str(e), "userId": user_id})
    raise HTTPException(500, "Inference failed. Please try again.")

5. Structured Logging

import json
import sys
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO, format='%(message)s', stream=sys.stdout)
logger = logging.getLogger(__name__)

def log_prediction(user_id, model_name, sign, confidence, latency_ms, event_id):
    logger.info(json.dumps({
        "event": "prediction",
        "userId": user_id,
        "modelName": model_name,
        "sign": sign,
        "confidence": round(confidence, 3),
        "latencyMs": latency_ms,
        "eventId": event_id,
        "timestamp": datetime.utcnow().isoformat()
    }))

Every prediction carries an eventId for log correlation.

6. CORS Configuration

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://your-frontend-domain.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["Authorization", "Content-Type"],
    max_age=3600
)

Continuous Security in CI/CD

Trivy scans all four container images for HIGH/CRITICAL CVEs on each build:

# .github/workflows/securityScan.yml — runs weekly + on every PR
- name: Scan Docker Images with Trivy
  run: |
    for image in authService inferenceService agentService frontend; do
      trivy image --severity HIGH,CRITICAL \
        $ACR_NAME.azurecr.io/$image:latest
    done

Secrets are kept on GitHub (repo secrets) and Azure (Key Vault / Kubernetes Secrets) so they are only accessed at runtime — never committed to the repo.

Secrets Management

Never commit secrets. Use environment variables locally and platform secret stores in CI/CD and production:

# .env (local, not committed)
JWT_SECRET_KEY=<generated via: openssl rand -hex 32>
JWT_ALGORITHM=HS256
JWT_EXPIRY_MINUTES=60
OPENAI_API_KEY=sk-...
DATABRICKS_TOKEN=dapi...

For GitHub Actions (via repo settings):

  • AZURE_CLIENT_ID
  • AZURE_TENANT_ID
  • AZURE_SUBSCRIPTION_ID
  • ACR_NAME

All services read from os.getenv():

JWT_SECRET_KEY = os.getenv("JWT_SECRET_KEY")
if not JWT_SECRET_KEY:
    raise RuntimeError("JWT_SECRET_KEY not set")

OpenAPI Specs

Each FastAPI service automatically exposes OpenAPI docs:

  • http://localhost:8001/docs — authService interactive docs
  • http://localhost:8002/docs — inferenceService interactive docs
  • http://localhost:8003/docs — agentService interactive docs

Export for external consumption:

curl http://localhost:8001/openapi.json > docs/openapi.authService.json
curl http://localhost:8002/openapi.json > docs/openapi.inferenceService.json
curl http://localhost:8003/openapi.json > docs/openapi.agentService.json