Building a Home Secretary #5 - Server: Signatures and the Gatekeeper

"""HMAC request signatures. The server and the example client use the same functions.

String to sign = METHOD \\n PATH[?QUERY] \\n TIMESTAMP \\n NONCE \\n SHA256(body) in hex
Signature = base64url(HMAC-SHA256(secret, string to sign))
"""
import base64
import hashlib
import hmac
import secrets
import time

HEADER_KEY = "X-Access-Key"
HEADER_TIMESTAMP = "X-Timestamp"
HEADER_NONCE = "X-Nonce"
HEADER_SIGNATURE = "X-Signature"


def canonical_string(method: str, path: str, query: str, timestamp: str, nonce: str, body: bytes) -> str:
    """Build the string to sign. The server and client must build it identically, down to the last character, for signatures to match.

    e.g. GET /v1/me, no body →
        GET
        /v1/me
        1762300000
        3f9c...(nonce)
        e3b0c442...(SHA-256 of an empty body)
    The body isn't included whole, only its hash. Change even one character of the body and the hash changes.
    """
    target = f"{path}?{query}" if query else path
    body_hash = hashlib.sha256(body).hexdigest()
    return "\n".join([method.upper(), target, timestamp, nonce, body_hash])


def sign(secret: str, canonical: str) -> str:
    """Sign the string to sign with the Secret using HMAC-SHA256, returned as a base64url string.

    base64url uses -, _ instead of +, / so it can go straight into an HTTP header, and the trailing = padding is stripped.
    """
    digest = hmac.new(secret.encode(), canonical.encode(), hashlib.sha256).digest()
    return base64.urlsafe_b64encode(digest).decode().rstrip("=")


def build_headers(access_key: str, secret: str, method: str, path: str, query: str = "", body: bytes = b"") -> dict:
    """Build the 4 auth headers to attach to one request (for tests and the example client).

    The timestamp is the current time (seconds), and the Nonce is a new random value every time, so even the same request gets a different signature each time.
    """
    timestamp = str(int(time.time()))
    nonce = secrets.token_hex(16)
    canonical = canonical_string(method, path, query, timestamp, nonce, body)
    return {
        HEADER_KEY: access_key,
        HEADER_TIMESTAMP: timestamp,
        HEADER_NONCE: nonce,
        HEADER_SIGNATURE: sign(secret, canonical),
    }

"""Authentication dependency that runs in front of every /v1 request.

Check order:
  1. Whether it came through Cloudflare Access (the JWT header cloudflared adds)
  2. Header format, body size
  3. Whether the key is locked out (keys with piling failures are blocked for a while)
  4. Whether the device exists and isn't revoked, and whether the device owner (user) is active
  5. Timestamp tolerance
  6. Signature match (constant-time comparison)
  7. Nonce reuse
  8. scope (device scopes AND scopes the user's role allows)

The failure reason is recorded only in the audit log; the response is always the same single 401.
Revealing where it was blocked would give attackers a hint.
"""
import hmac
import re
import time
from dataclasses import dataclass

from fastapi import HTTPException, Request, status

from .registry import Device, User
from .scopes import allows, effective_scopes
from .signing import (
    HEADER_KEY,
    HEADER_NONCE,
    HEADER_SIGNATURE,
    HEADER_TIMESTAMP,
    canonical_string,
    sign,
)

_KEY_RE = re.compile(r"^[A-Za-z0-9_-]{3,64}$")
_NONCE_RE = re.compile(r"^[A-Za-z0-9_-]{16,64}$")
_TS_RE = re.compile(r"^\d{9,11}$")

UNAUTHORIZED = HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Authentication failed.")
FORBIDDEN = HTTPException(status.HTTP_403_FORBIDDEN, detail="Permission denied.")


@dataclass
class Caller:
    """A caller that passed authentication. Router functions receive this to know who (user), with which device (device), and what they can do (scopes)."""

    user: User
    device: Device
    scopes: set[str]
    ip: str | None

    def can(self, scope: str) -> bool:
        """Whether this caller can use a single scope. Wildcards (mail:*, modules:*) are checked too."""
        return allows(self.scopes, scope)

    def forbid(self, request: Request, scope: str) -> HTTPException:
        """Record the permission denial in the audit log and build the 403 exception to return. The caller raises it."""
        request.app.state.audit.write(
            "auth.forbidden", user=self.user.id, device=self.device.access_key, ip=self.ip,
            scope=scope, path=request.url.path,
        )
        return FORBIDDEN


def _client_ip(request: Request) -> str | None:
    """Find the requester's real IP. Used for the audit log."""
    # For requests that came through Cloudflare, the real client IP is in this header.
    return request.headers.get("Cf-Connecting-Ip") or (request.client.host if request.client else None)


async def authenticate(request: Request) -> Caller:
    """Check the signature in the request headers and return the caller (Caller). 401 if anything is wrong.

    See the description at the top of this file for the check order. Cheap checks (header format) come first,
    expensive checks (DB lookups, signature computation) come later.
    """
    state = request.app.state
    settings = state.settings
    audit = state.audit
    ip = _client_ip(request)

    def fail(reason: str, key: str | None = None) -> HTTPException:
        """Record an authentication failure in the audit log and the failure table, and build the 401 exception to return.

        The caller uses it as raise fail("reason"). The reason goes only to the log, never into the response.
        Failure records are added only when the key format is valid (so the table can't be filled with junk values).
        """
        audit.write("auth.fail", device=key, ip=ip, reason=reason, path=request.url.path)
        if key and _KEY_RE.match(key):
            with state.db.tx() as c:
                c.execute("INSERT INTO auth_failures (access_key, at) VALUES (?, ?)", (key, int(time.time())))
        return UNAUTHORIZED

    if settings.require_cf_access and not request.headers.get("Cf-Access-Jwt-Assertion"):
        raise fail("no_cf_access")

    key = request.headers.get(HEADER_KEY, "")
    ts = request.headers.get(HEADER_TIMESTAMP, "")
    nonce = request.headers.get(HEADER_NONCE, "")
    signature = request.headers.get(HEADER_SIGNATURE, "")
    if not (_KEY_RE.match(key) and _TS_RE.match(ts) and _NONCE_RE.match(nonce) and signature):
        raise fail("bad_headers", key or None)

    body = await request.body()
    if len(body) > settings.max_body_bytes:
        raise fail("body_too_large", key)

    now = int(time.time())
    with state.db.tx() as c:
        c.execute("DELETE FROM auth_failures WHERE at < ?", (now - settings.lockout_window_seconds,))
        failures = c.execute(
            "SELECT COUNT(*) FROM auth_failures WHERE access_key = ? AND at > ?",
            (key, now - settings.lockout_window_seconds),
        ).fetchone()[0]
    if failures >= settings.lockout_threshold:
        audit.write("auth.locked", device=key, ip=ip, path=request.url.path)
        raise UNAUTHORIZED

    device = state.registry.device(key)
    if device is None:
        raise fail("unknown_key", key)
    if device.revoked:
        raise fail("revoked_key", key)
    user = state.registry.user(device.user)
    if user is None or user.disabled:
        raise fail("user_disabled", key)

    if abs(now - int(ts)) > settings.clock_skew_seconds:
        raise fail("timestamp_skew", key)

    canonical = canonical_string(request.method, request.url.path, request.url.query, ts, nonce, body)
    expected = sign(device.secret, canonical)
    if not hmac.compare_digest(expected, signature):
        raise fail("bad_signature", key)

    # Nonces are remembered for twice the timestamp tolerance. After that, the timestamp check catches them.
    with state.db.tx() as c:
        c.execute("DELETE FROM nonces WHERE expires_at < ?", (now,))
        exists = c.execute(
            "SELECT 1 FROM nonces WHERE access_key = ? AND nonce = ?", (key, nonce)
        ).fetchone()
        if not exists:
            c.execute(
                "INSERT INTO nonces (access_key, nonce, expires_at) VALUES (?, ?, ?)",
                (key, nonce, now + settings.clock_skew_seconds * 2),
            )
    if exists:
        raise fail("nonce_replay", key)

    return Caller(user=user, device=device, scopes=effective_scopes(device.scopes, user.role), ip=ip)


def require(scope: str):
    """Take one scope and build and return a dependency function that "passes only with this permission."

        @router.get("/me")
        async def me(caller: Caller = Depends(require(SYSTEM_READ))):

    Authentication (401) comes first, then 403 if the scope is missing. It's a function that stamps out a check function per scope.
    """
    async def dependency(request: Request) -> Caller:
        """The function FastAPI actually calls per request. Checks the scope after authentication."""
        caller = await authenticate(request)
        if not caller.can(scope):
            raise caller.forbid(request, scope)
        return caller

    return dependency

1
import asyncio
2
import time
3

4

5
async def work(name: str, seconds: float) -> str:
6
    await asyncio.sleep(seconds)  # Go do something else while waiting
7
    return f"{name} done"
8

9

10
async def main():
11
    start = time.perf_counter()
12
    results = await asyncio.gather(work("mail", 1), work("note", 1), work("lights", 1))
13
    print(results, f"{time.perf_counter() - start:.1f}s")
14

15

16
asyncio.run(main())

1
from fastapi import APIRouter, Depends
2

3
from ..auth.guard import Caller, require
4
from ..auth.scopes import SYSTEM_READ
5

6
router = APIRouter(prefix="/v1", tags=["system"])
7

8

9
@router.get("/me")
10
async def me(caller: Caller = Depends(require(SYSTEM_READ))):
11
    """GET /v1/me : Return whose device this key is, which device, and what it can do.
12

13
    Meant to be the first call after setting up a new device. If you get a 200, both the signature and the Cloudflare Access setup are correct.
14
    """
15
    d = caller.device
16
    return {
17
        "user": caller.user.public(),
18
        "device": {"access_key": d.access_key, "name": d.name},
19
        "scopes": sorted(caller.scopes),
20
    }

1
from fastapi import Depends, FastAPI, Header, HTTPException
2
from fastapi.testclient import TestClient
3

4
app = FastAPI()
5

6

7
def who_are_you(x_name: str = Header("")) -> str:
8
    if not x_name:
9
        raise HTTPException(401, "Who are you?")
10
    return x_name
11

12

13
@app.get("/hello")
14
def hello(name: str = Depends(who_are_you)):
15
    return {"hello": name}
16

17

18
client = TestClient(app)
19
print(client.get("/hello").status_code)
20
print(client.get("/hello", headers={"X-Name": "dad"}).json())

1
import logging
2
import uuid
3
from contextlib import asynccontextmanager
4

5
from fastapi import FastAPI, Request
6
from fastapi.exceptions import RequestValidationError
7
from fastapi.responses import JSONResponse
8
from starlette.exceptions import HTTPException
9

10
from .audit import AuditLog
11
from .auth.registry import Registry
12
from .config import Settings, get_settings
13
from .db import Database
14
from .approvals import ApprovalQueue
15
from .catalog import ModuleCatalog
16
from .events import EventStore
17
from .routers import approvals, commands, events, system
18

19
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
20
logger = logging.getLogger("secretary")
21

22

23
def create_app(settings: Settings | None = None, module_transport=None) -> FastAPI:
24
    """Create the FastAPI app. Tests create their own by passing in settings and a fake module (module_transport).
25

26
    The production server has uvicorn load app = create_app() at the very bottom.
27
    """
28
    settings = settings or get_settings()
29

30
    @asynccontextmanager
31
    async def lifespan(app: FastAPI):
32
        """When the server starts (before yield), create the needed objects and put them on app.state; when it stops (after yield), clean up.
33

34
        Routers fetch them like request.app.state.db.
35
        """
36
        app.state.settings = settings
37
        app.state.db = Database(settings.db_path)
38
        app.state.registry = Registry(settings.registry_path, settings.max_users)
39
        app.state.audit = AuditLog(settings.audit_path)
40
        app.state.approvals = ApprovalQueue(app.state.db, settings.approval_ttl_seconds)
41
        app.state.events = EventStore(app.state.db)
42
        app.state.catalog = ModuleCatalog(settings.module_endpoints, settings.module_timeout_seconds,
43
                                          transport=module_transport)
44
        app.state.audit.write("server.start", modules=sorted(settings.module_endpoints))
45
        yield
46
        await app.state.catalog.close()
47
        app.state.db.close()
48

49
    # Don't open the docs pages in production. If you need the spec, generate it locally.
50
    app = FastAPI(title="Secretary", lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None)
51

52
    @app.middleware("http")
53
    async def request_id(request: Request, call_next):
54
        """Middleware that attaches X-Request-Id to every response.
55

56
        If an unexpected error occurs, the details go only to the server log, and the client gets just a 500 with the same ID.
57
        If a user tells you the ID, you can find that request in the log right away.
58
        """
59
        rid = uuid.uuid4().hex[:12]
60
        try:
61
            response = await call_next(request)
62
        except Exception:
63
            logger.exception("unhandled error rid=%s path=%s", rid, request.url.path)
64
            response = JSONResponse({"code": "INTERNAL_ERROR", "message": "A temporary error occurred."}, 500)
65
        response.headers["X-Request-Id"] = rid
66
        return response
67

68
    @app.exception_handler(HTTPException)
69
    async def http_error(request: Request, exc: HTTPException):
70
        """Unify HTTP errors (401, 403, 404 ...) into JSON shaped like {code, message}."""
71
        return JSONResponse({"code": f"HTTP_{exc.status_code}", "message": exc.detail}, exc.status_code)
72

73
    @app.exception_handler(RequestValidationError)
74
    async def validation_error(request: Request, exc: RequestValidationError):
75
        """Turn request format errors (422) into the shape {code, message, fields}."""
76
        # Only say which fields were wrong and why; never send back the input values themselves.
77
        fields = [".".join(str(p) for p in e["loc"][1:]) for e in exc.errors()]
78
        return JSONResponse({"code": "INVALID_REQUEST", "message": "Invalid request format.", "fields": fields}, 422)
79

80
    @app.get("/healthz", include_in_schema=False)
81
    async def healthz():
82
        """GET /healthz : Whether the server is alive. Open without auth, called by the container health check."""
83
        return {"ok": True}
84

85
    app.include_router(system.router)
86
    app.include_router(commands.router)
87
    app.include_router(events.router)
88
    app.include_router(events.internal)
89
    app.include_router(approvals.router)
90
    return app
91

92

93
app = create_app()

1
from contextlib import asynccontextmanager
2

3
from fastapi import FastAPI
4
from fastapi.testclient import TestClient
5

6

7
@asynccontextmanager
8
async def lifespan(app: FastAPI):
9
    print("Server start: connect DB")
10
    yield
11
    print("Server stop: close DB")
12

13

14
app = FastAPI(lifespan=lifespan)
15

16

17
@app.get("/")
18
def root():
19
    print("Handling request")
20
    return {}
21

22

23
with TestClient(app) as client:
24
    client.get("/")