Skip to content

Lesson 9 — protected endpoints

What you will build

The other half of lesson 8. Knowing who is calling is not the same as knowing what they may do, and this lesson is where the application starts refusing people it has successfully identified: 403, roles, and the deployment obligation that keeps an assertion from being forgeable.

Nothing in the framework changes. Authorization is a product decision (ADR 0014), so what this lesson shows is a shape you can keep in a route table and audit in a review.

What you need

Nothing new.

Two refusals, two questions

Status Class The question it answers What a client does next
401 UnauthenticatedError I do not know who you are authenticate, then retry
403 ForbiddenError I know who you are, and no nothing: retrying will not help

A client that cannot tell them apart retries forever or gives up too early. Both classes are in identity.py, next to the verifier that produces the principal they are about:

examples/user_management/identity.py
"""Who is calling, established by something that checked the assertion.

Xyberos ships no identity model, and will not (see
``docs/adr/0014-no-identity-or-authorization-model.md``). What it provides is the shape this module
is written in: an `HttpError` with the status you choose, a component that verifies rather than
believes, and a principal that travels as an argument.

The rule this module exists to enforce, in one line: a header is trusted because something
*verified* it, never because it arrived. Until `verify` returns, nothing in the request is a fact
about the caller.
"""

from __future__ import annotations

import hashlib
import hmac
import os
from dataclasses import dataclass

from starlette.requests import Request

from xyberos import HttpError

SUBJECT_HEADER = "X-Authenticated-Subject"
ROLES_HEADER = "X-Authenticated-Roles"
SIGNATURE_HEADER = "X-Principal-Signature"
SECRET_ENV = "XYBEROS_IDENTITY_SECRET"


@dataclass(frozen=True)
class Principal:
    """Who is calling, and what the verifier established about them."""

    subject: str
    roles: frozenset[str] = frozenset()


class UnauthenticatedError(HttpError):
    """The assertion was missing, unsigned, or signed by something else; exposed as a 401."""

    def __init__(self, reason: str) -> None:
        super().__init__(f"the request is not authenticated: {reason}", status_code=401)


class ForbiddenError(HttpError):
    """The caller is known and is not allowed to do this; exposed as a 403.

    The two statuses answer different questions, which is why they are separate classes:
    `401` means "I do not know who you are", `403` means "I know who you are, and no". A client can
    act on the first — authenticate more convincingly — and cannot act on the second.
    """

    def __init__(self, requirement: str) -> None:
        super().__init__(f"this operation requires {requirement}", status_code=403)


def require_role(principal: Principal, role: str) -> None:
    """Refuse a principal that does not carry this role, as a `403`."""
    if role not in principal.roles:
        raise ForbiddenError(f"the {role} role")


class PrincipalVerifier:
    """Turns a signed assertion into a principal, or refuses.

    The secret is configuration, and a deployment without one refuses every request rather than
    trusting the request: the failure is closed, and it is a 401 rather than a crash.
    """

    def __init__(self, secret: str | None = None) -> None:
        self._secret = (secret if secret is not None else os.environ.get(SECRET_ENV, "")).encode()

    def verify(self, request: Request) -> Principal:
        """Return the verified principal for a request, or raise `UnauthenticatedError`."""
        if not self._secret:
            raise UnauthenticatedError(f"no identity secret is configured in {SECRET_ENV}")
        subject = request.headers.get(SUBJECT_HEADER, "")
        roles = request.headers.get(ROLES_HEADER, "")
        signature = request.headers.get(SIGNATURE_HEADER, "")
        if not subject:
            raise UnauthenticatedError(f"{SUBJECT_HEADER} is required")
        if not hmac.compare_digest(signature, self.sign(subject, roles)):
            raise UnauthenticatedError("the assertion signature does not match")
        return Principal(subject=subject, roles=_roles(roles))

    def sign(self, subject: str, roles: str = "") -> str:
        """Return the signature a caller must present for this subject and these roles."""
        message = f"{subject}\n{roles}".encode()
        return hmac.new(self._secret, message, hashlib.sha256).hexdigest()

    def headers(self, subject: str, roles: str = "") -> dict[str, str]:
        """Return the headers a client presents after authenticating successfully."""
        return {
            SUBJECT_HEADER: subject,
            ROLES_HEADER: roles,
            SIGNATURE_HEADER: self.sign(subject, roles),
        }


def _roles(raw: str) -> frozenset[str]:
    return frozenset(role.strip() for role in raw.split(",") if role.strip())

The route table is the policy

examples/user_management/app.py gains one nested function and one call at the top of every administrative route:

examples/user_management/app.py
"""Compose the application and expose it as an ASGI app."""

from __future__ import annotations

from collections.abc import Mapping
from pathlib import Path
from typing import Annotated

from starlette.applications import Starlette
from starlette.datastructures import QueryParams
from starlette.requests import Request
from starlette.responses import RedirectResponse, Response
from starlette.templating import Jinja2Templates

from user_management.errors import DuplicateEmailError, UserValidationError
from user_management.identity import PrincipalVerifier, require_role
from user_management.service import DEFAULT_LIMIT, UserService
from xyberos import App, Inject, Settings
from xyberos.database import DatabaseProvider, SqliteSettings, sqlite_provider
from xyberos.http import HttpAdapter, read_json_object

#: The role the administrative routes require. One constant, so the route table reads as a list.
ADMIN = "admin"

#: Where the pages live. Templates are application code: the framework routes and serializes, and a
#: template engine is something an application adds.
TEMPLATES = Path(__file__).parent / "templates"

#: The header HTMX sends. It is how one route answers a browser and a script differently.
HX_REQUEST = "HX-Request"


def create_app(
    *,
    settings: SqliteSettings | None = None,
    database: DatabaseProvider | None = None,
    secret: str | None = None,
) -> App:
    """Compose the application.

    Pass ``settings`` to let the application own and start its database, or ``database`` to supply
    an already-started provider, which the application then owns and closes on shutdown. ``secret``
    is the identity provider's shared secret; without one the verifier refuses every request.
    """
    app = App(settings=Settings.load(environ={}, overrides={"app": {"name": "user_management"}}))
    if database is None:
        app.provide(DatabaseProvider, factory=sqlite_provider(settings), eager=True)
    else:
        app.provide(DatabaseProvider, instance=database)
    app.provide(UserService)
    app.provide(PrincipalVerifier, instance=PrincipalVerifier(secret))

    def administrator(request: Request, verifier: PrincipalVerifier) -> None:
        """Refuse a caller who is not an administrator, before anything else happens.

        One call at the top of a route, and its cost is that it can be forgotten: there is no marker
        on the handler that a reviewer can grep for except this line. That is the trade the ADR
        describes — the application owns authorization, so the route table is where it is audited.
        """
        require_role(verifier.verify(request), ADMIN)

    @app.get("/health")
    async def health(database: Annotated[DatabaseProvider, Inject()]) -> dict[str, str]:
        await database.health()
        return {"status": "ready"}

    @app.post("/users")
    async def create_user(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        administrator(request, verifier)
        payload = await read_json_object(request)
        return await users.create_user(payload.get("email"), payload.get("display_name"))

    @app.get("/users")
    async def list_users(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        administrator(request, verifier)
        query = request.query_params
        return await users.list_users(
            limit=_whole_number(query, "limit", default=DEFAULT_LIMIT),
            offset=_whole_number(query, "offset", default=0),
        )

    @app.get("/users/{user_id}")
    async def get_user(
        user_id: int,
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        administrator(request, verifier)
        return await users.get_user(user_id)

    @app.patch("/users/{user_id}")
    async def update_user(
        user_id: int,
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        administrator(request, verifier)
        payload = await read_json_object(request)
        return await users.update_user(user_id, payload)

    @app.delete("/users/{user_id}")
    async def delete_user(
        user_id: int,
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> None:
        administrator(request, verifier)
        await users.delete_user(user_id)

    @app.get("/me")
    async def read_me(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        principal = verifier.verify(request)
        return await users.find_own_user(principal)

    @app.post("/me")
    async def claim_account(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        principal = verifier.verify(request)
        payload = await read_json_object(request)
        return await users.create_user(
            payload.get("email"),
            payload.get("display_name"),
            external_subject=principal.subject,
        )

    @app.patch("/me")
    async def update_me(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> dict[str, object]:
        principal = verifier.verify(request)
        payload = await read_json_object(request)
        return await users.update_own_profile(principal, payload)

    templates = Jinja2Templates(directory=str(TEMPLATES))

    async def users_page(
        request: Request, users: UserService, *, status_code: int = 200, error: str | None = None
    ) -> Response:
        page = await users.list_users()
        return templates.TemplateResponse(
            request,
            "users.html",
            {"users": page["items"], "error": error, "form": await _submitted_form(request)},
            status_code=status_code,
        )

    @app.get("/pages/users")
    async def show_users(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> Response:
        administrator(request, verifier)
        return await users_page(request, users)

    @app.post("/pages/users")
    async def create_user_from_form(
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> Response:
        administrator(request, verifier)
        form = await request.form()
        try:
            await users.create_user(form.get("email"), form.get("display_name"))
        except (DuplicateEmailError, UserValidationError) as error:
            return await users_page(request, users, status_code=422, error=str(error))
        return RedirectResponse("/pages/users", status_code=303)

    @app.post("/pages/users/{user_id}/toggle")
    async def toggle_user(
        user_id: int,
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        users: Annotated[UserService, Inject()],
    ) -> Response:
        administrator(request, verifier)
        current = await users.get_user(user_id)
        updated = await users.update_user(user_id, {"enabled": not current["enabled"]})
        if request.headers.get(HX_REQUEST):
            return templates.TemplateResponse(request, "_row.html", {"user": updated})
        return RedirectResponse("/pages/users", status_code=303)

    return app


async def _submitted_form(request: Request) -> Mapping[str, object] | None:
    """Return what the client typed, so a refused form can be shown back to them."""
    if request.method != "POST":
        return None
    form = await request.form()
    return {key: str(value) for key, value in form.items()}


def _whole_number(query: QueryParams, name: str, *, default: int) -> int:
    """Read one query string value as a whole number, or refuse the request.

    A query string is text: `?limit=abc` arrives as `"abc"`. The conversion belongs here, where the
    request is untrusted, and the *bounds* stay in the service, which knows what a page may be.
    """
    raw = query.get(name)
    if raw is None:
        return default
    try:
        return int(raw)
    except ValueError:
        raise UserValidationError(f"{name} must be a whole number") from None


def build_asgi_app(app: App) -> Starlette:
    """Build an ASGI application for a composed application."""
    return HttpAdapter(app).asgi_app


def app_factory() -> Starlette:
    """Create the application's ASGI app for an ASGI server.

    The identity secret is read from the environment by the verifier, and a deployment that has not
    set it refuses every request rather than trusting one.

    Usage: ``uvicorn --app-dir examples --factory user_management.app:app_factory``
    """
    return build_asgi_app(create_app())

The whole policy is now readable in one place:

Route Requirement
GET /health nothing — a health check that needs identity is a health check a load balancer cannot run
GET /users, GET /users/{user_id}, POST /users, PATCH /users/{user_id}, DELETE /users/{user_id} the admin role
GET /me, POST /me, PATCH /me any verified principal, and only about itself

That table is the audit surface. administrator(request, verifier) is one line, and its cost is that a new administrative route can forget it — which is why the route table, not the handler, is what a reviewer reads. A framework that owned authorization would give you a decorator and the same obligation; it would also own the security of the decision, which is the trade the ADR refuses.

Every earlier script changed, and nothing they print did

Four earlier lessons drive these routes without an assertion, so each of their scripts needed the same two lines: a SECRET constant, and ADMIN_HEADERS = PrincipalVerifier(SECRET).headers("admin@idp", "admin") — which every request in those scripts now carries through client.headers.update(ADMIN_HEADERS). The lesson modules on lesson 4's page and the pages after it show the lines in place.

Every transcript in lessons 4–8 is otherwise byte-identical, including lesson 7's migration run. That is checked rather than claimed: the suite runs each lesson module and compares its output with the block on the page, so a change that quietly altered a documented result would fail the build. Lesson 8's own script is the one that already had an assertion; it keeps working because the roles it sends have not changed.

The edge obligation

Everything in this lesson depends on one operational fact, and it is the one the framework cannot enforce for you: the assertion headers must be stripped from client requests at the edge. If your proxy forwards a client-supplied X-Authenticated-Subject or X-Authenticated-Roles, a client chooses their own identity and every check above becomes decoration.

  • The proxy removes those headers from inbound requests, and sets them only after authenticating.
  • The secret lives in the deployment's environment, not in the repository, and never in a URL: the signature is what makes the header a fact.
  • Nothing here is durable. A verified principal is per request, which is why nothing is cached in a session, a cookie, or contextvars (security guide).

What the lesson runs

examples/learn_traditional/traditional09_protected.py
"""Lesson 9 — roles, and the difference between 401 and 403.

Run it with:

    python -m learn_traditional.traditional09_protected
"""

from __future__ import annotations

import tempfile
from pathlib import Path

from httpx import Response
from user_management.app import create_app
from user_management.identity import PrincipalVerifier
from user_management.migrate import migrate

from xyberos.database import SqliteSettings
from xyberos.http import HttpAdapter
from xyberos.http.testing import create_test_client

SECRET = "a-secret-the-deployment-keeps"
VERIFIER = PrincipalVerifier(SECRET)
MEMBER = VERIFIER.headers("ada@idp", "member")
ADMIN = VERIFIER.headers("admin@idp", "admin")
#: A member's assertion with the roles header edited. The signature covers the roles, so it fails.
FORGED = {**MEMBER, "X-Authenticated-Roles": "admin"}


def show(label: str, response: Response) -> None:
    """Print one response as the lesson reads it: the status, then the body."""
    body = response.json() if response.content else None
    print(f"{label} ->", response.status_code, body)


def main() -> None:
    with tempfile.TemporaryDirectory() as directory:
        settings = SqliteSettings(path=Path(directory) / "users.db")
        migrate(settings)

        app = create_app(settings=settings, secret=SECRET)
        with create_test_client(HttpAdapter(app)) as client:
            show("no assertion", client.get("/users"))
            show("a member asks for the collection", client.get("/users", headers=MEMBER))
            show("a member editing the roles header", client.get("/users", headers=FORGED))

            show(
                "an administrator creates a user",
                client.post(
                    "/users",
                    json={"email": "grace@example.com", "display_name": "Grace Hopper"},
                    headers=ADMIN,
                ),
            )
            listed = client.get("/users", headers=ADMIN)
            print(
                "an administrator sees", len(listed.json()["items"]), "of", listed.json()["total"]
            )

            show(
                "a member claims an account",
                client.post(
                    "/me",
                    json={"email": "ada@example.com", "display_name": "Ada Lovelace"},
                    headers=MEMBER,
                ),
            )
            show("a member reads their own account", client.get("/me", headers=MEMBER))
            show("a member reads the admin's account", client.get("/users/1", headers=MEMBER))


if __name__ == "__main__":
    main()

How to run it

$env:PYTHONPATH = "src;examples"
python -m learn_traditional.traditional09_protected

PYTHONPATH=src:examples python -m learn_traditional.traditional09_protected

What it prints

no assertion -> 401 {'error': {'type': 'UnauthenticatedError', 'message': 'the request is not authenticated: X-Authenticated-Subject is required'}}
a member asks for the collection -> 403 {'error': {'type': 'ForbiddenError', 'message': 'this operation requires the admin role'}}
a member editing the roles header -> 401 {'error': {'type': 'UnauthenticatedError', 'message': 'the request is not authenticated: the assertion signature does not match'}}
an administrator creates a user -> 200 {'id': 1, 'email': 'grace@example.com', 'display_name': 'Grace Hopper', 'enabled': True, 'external_subject': None}
an administrator sees 1 of 1
a member claims an account -> 200 {'id': 2, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': True, 'external_subject': 'ada@idp'}
a member reads their own account -> 200 {'id': 2, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': True, 'external_subject': 'ada@idp'}
a member reads the admin's account -> 403 {'error': {'type': 'ForbiddenError', 'message': 'this operation requires the admin role'}}

Three of those lines are the lesson:

  • a member asks for the collection -> 403. The caller is known and the collection is not theirs.
  • a member editing the roles header -> 401, not 403. The signature covers the subject and the roles, so an edited header is not a member who overreached: it is an assertion that failed. This is the difference between an authorization bug and a broken client, and the status says which happened.
  • a member reads the admin's account -> 403. GET /users/{user_id} is an administrative route whoever you are; owning an account never grants it.

Common mistakes

  • Answering 404 where you mean 403. Replacing a refusal with "not found" hides existence, which is sometimes the right call and always a decision — make it deliberately, as the exercise does.
  • Letting a client supply a role. A role in a header is only a role because the signature covered it. Roles from a query parameter, a body, or a cookie are a client asserting its own permissions.
  • Checking the role in the service. The service should not know what a route is: keep the refusal at the edge of the operation and pass a principal that already satisfies it. When a rule is genuinely about the data — "this record belongs to that subject" — it belongs in the service, scoped by the principal it is given.
  • Auditing only the happy path. The routes that need a check are the ones that change state, and the ones that leak data are the ones that read. Both lists belong in the review.
  • Assuming the gateway stripped the headers. Verify it in a test, or in staging, by sending X-Authenticated-Subject yourself and expecting a 401. A deployment that forwards it looks exactly like a deployment that works.

Exercise

Add GET /accounts/{user_id} with the rule nearly every application ends up needing: a caller may read their own account, an administrator may read anybody's, and everybody else is refused. Pay attention to what the refusal says and to when it happens.

Show the answer
examples/learn_traditional/traditional09_exercise.py
"""Lesson 9, exercise — owner or administrator, and no third case.

The answer to the lesson's exercise. The check is the one every application eventually writes: a
caller may read their own record, and an administrator may read anybody's, and everybody else is
refused with the same `403` — the same message, so the refusal does not become a way to enumerate
which accounts exist.
"""

from __future__ import annotations

import tempfile
from pathlib import Path
from typing import Annotated

from httpx import Response
from starlette.requests import Request
from user_management.app import ADMIN, create_app
from user_management.identity import ForbiddenError, Principal, PrincipalVerifier
from user_management.migrate import migrate
from user_management.service import UserService

from xyberos import App, Inject
from xyberos.database import SqliteSettings
from xyberos.http import HttpAdapter
from xyberos.http.testing import create_test_client

SECRET = "a-secret-the-deployment-keeps"
VERIFIER = PrincipalVerifier(SECRET)
MEMBER = VERIFIER.headers("ada@idp", "member")
ADMIN_HEADERS = VERIFIER.headers("admin@idp", ADMIN)


class UserReader(UserService):
    """``service.py``, plus the one authorization rule the exercise asks for."""

    async def read_for(self, principal: Principal, user_id: int) -> dict[str, object]:
        """Return one user, if the caller owns that account or administers the application."""
        if ADMIN not in principal.roles:
            own = await self.find_own_user(principal)
            if own["id"] != user_id:
                raise ForbiddenError(f"the {ADMIN} role or your own account")
        return await self.get_user(user_id)


def create_guarded_app(*, settings: SqliteSettings) -> App:
    """The lesson's application, with the exercise's route added to it."""
    app = create_app(settings=settings, secret=SECRET)
    app.provide(UserReader)

    @app.get("/accounts/{user_id}")
    async def read_account(
        user_id: int,
        request: Request,
        verifier: Annotated[PrincipalVerifier, Inject()],
        reader: Annotated[UserReader, Inject()],
    ) -> dict[str, object]:
        return await reader.read_for(verifier.verify(request), user_id)

    return app


def show(label: str, response: Response) -> None:
    """Print one response as the lesson reads it: the status, then the body."""
    body = response.json() if response.content else None
    print(f"{label} ->", response.status_code, body)


def main() -> None:
    """One member, one administrator, and the three answers the rule can give."""
    with tempfile.TemporaryDirectory() as directory:
        settings = SqliteSettings(path=Path(directory) / "users.db")
        migrate(settings)

        app = create_guarded_app(settings=settings)
        with create_test_client(HttpAdapter(app)) as client:
            claimed = client.post(
                "/me",
                json={"email": "ada@example.com", "display_name": "Ada Lovelace"},
                headers=MEMBER,
            )
            print("the member's own id:", claimed.json()["id"])

            show("their own account", client.get("/accounts/1", headers=MEMBER))
            show("somebody else's account", client.get("/accounts/99", headers=MEMBER))
            show(
                "an administrator reads account 99",
                client.get("/accounts/99", headers=ADMIN_HEADERS),
            )
            show(
                "an administrator reads account 1", client.get("/accounts/1", headers=ADMIN_HEADERS)
            )


if __name__ == "__main__":
    main()

the member's own id: 1
their own account -> 200 {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': True, 'external_subject': 'ada@idp'}
somebody else's account -> 403 {'error': {'type': 'ForbiddenError', 'message': 'this operation requires the admin role or your own account'}}
an administrator reads account 99 -> 404 {'error': {'type': 'UserNotFoundError', 'message': 'user 99 was not found'}}
an administrator reads account 1 -> 200 {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': True, 'external_subject': 'ada@idp'}
The last four lines are the rule, and the third is the one to think about: - **`their own account -> 200`**, because the account is theirs. - **`somebody else's account -> 403`**, and the message is the *requirement*, not a fact about the account. Refusing before the lookup is deliberate: a `404` for somebody else's record would let a member walk the id space and learn which accounts exist. - **`an administrator reads account 99 -> 404`.** The administrator passes the authorization check, so they get the honest answer: no such user. Adding the admin rule did not change lesson 6's error, which is what "authorization is a separate decision from lookup" looks like in practice.

Next

Lesson 10 gives the application an interface: server-rendered pages over the same handlers, and the first lesson where the reader supplies a piece of the stack.