Lesson 8 — identity from upstream¶
What you will build¶
The first lesson about something a request carries that the application must not trust. Xyberos ships no identity, session, tenancy, or authorization model and will not (ADR 0014), so this lesson is not about a framework feature: it is about the shape the ADR sanctions, and the four rules that make it safe.
Two of the endpoints you already have are the wrong way round for this. GET /users/{user_id} asks the
caller to name a user; after this lesson the caller cannot name one at all.
What you need¶
Nothing new. The interesting part of this lesson is what it does not install: no auth library, no session store, no role model.
The four rules¶
| Rule | What it means here |
|---|---|
| A header is trusted only because something verified it | PrincipalVerifier re-computes an HMAC over the subject and roles; a header that merely arrived proves nothing |
| The verified principal is passed as an argument | every /me service call takes a Principal, so there is no code path that reaches a user without one |
| The subject never comes from the payload | PATCH /me refuses id, enabled, and external_subject: the client does not get to say who it is |
| Context values are not identity | bind_context is diagnostic only (security guide); a subject in contextvars is ambient authority |
examples/user_management/identity.py is the whole verifier. Everything else in this lesson follows
from it:
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())
Note what the secret does when it is missing. A deployment with no XYBEROS_IDENTITY_SECRET refuses
every request with a 401 rather than trusting one, and it fails closed without crashing: the
verifier is constructed at import time and asks the question on the first request.
The link is a migration¶
A verified principal is a fact about the caller, not an account. Linking the two needs somewhere to keep the link, and that is lesson 7's machinery doing its job: a new column, nullable and unique, following the revision that created the table.
examples/user_management/migrations/versions/0002_add_external_subject.py
"""link users to the subject an identity provider verifies
The subject is a separate column rather than the email address, because the two change for different
reasons: an email address can be changed by the person who owns it, while a subject is what the
identity provider will keep asserting. A local account linked by email stops being reachable the
first time somebody's address changes upstream.
The column is nullable and unique: most rows may have no link yet, and no two accounts may claim the
same subject.
Revision ID: 0002_add_external_subject
Revises: 0001_create_users
Create Date: 2026-10-12
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
revision = "0002_add_external_subject"
down_revision = "0001_create_users"
branch_labels = None
depends_on = None
def upgrade() -> None:
with op.batch_alter_table("users") as batch:
batch.add_column(sa.Column("external_subject", sa.String(length=200), nullable=True))
batch.create_unique_constraint("uq_users_external_subject", ["external_subject"])
def downgrade() -> None:
with op.batch_alter_table("users") as batch:
batch.drop_constraint("uq_users_external_subject", type_="unique")
batch.drop_column("external_subject")
Two decisions are recorded in that file. The subject is a separate column rather than the email address, because an email is what a person may change and a subject is what the identity provider will keep asserting; and the column is unique, so two accounts cannot claim one subject even if two requests race past the same check.
Where the principal goes¶
examples/user_management/app.py gains a provider for the verifier and three routes. Each one verifies
first, then calls the service with the principal — never with a user id:
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())
UserService grew the methods those routes call: find_own_user(principal) and
update_own_profile(principal, changes). Neither has a user_id parameter. That is the design in one
sentence: the caller cannot ask for an account, only for its account, and the query that finds it is
WHERE external_subject = :subject from the principal.
create_user gained one keyword argument — external_subject — so POST /me reuses every rule lesson 4
established: the same email validation, the same duplicate check, the same transaction. A repeat claim
is a 409 decided before the payload is judged, because the conflict is about the request's target
rather than about the body.
What the lesson runs¶
examples/learn_traditional/traditional08_identity.py
"""Lesson 8 — identity from upstream, and the account it belongs to.
Run it with:
python -m learn_traditional.traditional08_identity
"""
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"
SELF_CHANGES = [
{"display_name": "Ada, Countess of Lovelace "},
{"enabled": False},
{"external_subject": "linus@idp"},
{"id": 2},
]
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)
with create_test_client(HttpAdapter(create_app(settings=settings))) as unconfigured:
show("no secret configured", unconfigured.get("/me"))
app = create_app(settings=settings, secret=SECRET)
verifier = PrincipalVerifier(SECRET)
ada = verifier.headers("ada@idp")
linus = verifier.headers("linus@idp")
with create_test_client(HttpAdapter(app)) as client:
show("unsigned", client.get("/me"))
show(
"signed for the wrong roles",
client.get("/me", headers={**ada, "X-Authenticated-Roles": "admin"}),
)
show("before claiming an account", client.get("/me", headers=ada))
show(
"claim",
client.post(
"/me",
json={"email": "Ada@Example.com", "display_name": "Ada Lovelace"},
headers=ada,
),
)
show("read back", client.get("/me", headers=ada))
show("claim again", client.post("/me", json={"email": "a@b.com"}, headers=ada))
for changes in SELF_CHANGES:
show(f"PATCH /me {changes}", client.patch("/me", json=changes, headers=ada))
show("somebody else", client.get("/me", headers=linus))
if __name__ == "__main__":
main()
How to run it¶
What it prints¶
no secret configured -> 401 {'error': {'type': 'UnauthenticatedError', 'message': 'the request is not authenticated: no identity secret is configured in XYBEROS_IDENTITY_SECRET'}}
unsigned -> 401 {'error': {'type': 'UnauthenticatedError', 'message': 'the request is not authenticated: X-Authenticated-Subject is required'}}
signed for the wrong roles -> 401 {'error': {'type': 'UnauthenticatedError', 'message': 'the request is not authenticated: the assertion signature does not match'}}
before claiming an account -> 404 {'error': {'type': 'AccountNotLinkedError', 'message': 'no local account is linked to ada@idp'}}
claim -> 200 {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': True, 'external_subject': 'ada@idp'}
read back -> 200 {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': True, 'external_subject': 'ada@idp'}
claim again -> 409 {'error': {'type': 'AccountAlreadyLinkedError', 'message': 'ada@idp is already linked to an account'}}
PATCH /me {'display_name': 'Ada, Countess of Lovelace '} -> 200 {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada, Countess of Lovelace', 'enabled': True, 'external_subject': 'ada@idp'}
PATCH /me {'enabled': False} -> 422 {'error': {'type': 'UserValidationError', 'message': 'unknown field: enabled'}}
PATCH /me {'external_subject': 'linus@idp'} -> 422 {'error': {'type': 'UserValidationError', 'message': 'unknown field: external_subject'}}
PATCH /me {'id': 2} -> 422 {'error': {'type': 'UserValidationError', 'message': 'unknown field: id'}}
somebody else -> 404 {'error': {'type': 'AccountNotLinkedError', 'message': 'no local account is linked to linus@idp'}}
| Line | What it establishes |
|---|---|
no secret configured -> 401 |
a deployment that forgot its secret refuses everything, and says why without quoting the secret |
unsigned -> 401 |
headers alone are not identity: there is no signature to check |
signed for the wrong roles -> 401 |
the signature covers the roles, so a caller cannot add one by editing a header |
before claiming an account -> 404 |
being authenticated is not the same as having an account here |
claim -> 200 |
the verified subject becomes external_subject on a row created by the same rules as lesson 4 |
claim again -> 409 |
one subject, one account |
PATCH /me {enabled: False} -> 422 |
a user does not get to enable or disable themselves |
PATCH /me {external_subject: ...} -> 422 |
and does not get to re-point the link |
PATCH /me {id: 2} -> 422 |
nor to name somebody else's account |
somebody else -> 404 |
another verified subject has their own account, and sees nothing of this one |
What the framework provides, and what it refuses to¶
| Provided | Refused |
|---|---|
HttpError with the status you choose (401 here, 403 in lesson 9) |
any identity, session, or role model |
| request-scoped providers, if a verifier should cache its work inside one request | a login form, or a place to keep a password |
app.with_overrides to fake a verifier in a test |
a tenant in contextvars, or any ambient authority |
| an error policy that never leaks an internal message to a client | enforcement of tenancy, or a warning when you forget to verify |
That list is short on purpose. A framework that owned authorization would be responsible for security it cannot see, and a half-built one would be worse than none.
Common mistakes¶
- Trusting a header because it is present.
X-Authenticated-Subject: ada@idpis a string anyone can send. It becomes a fact only after something checked it, and the check is what this lesson's HMAC stands in for: replace it with a token signature or an introspection call. - Putting the principal in a session or a cookie. A session is a second identity system with its own lifetime, revocation, and failure modes; if the assertion arrives on every request, verify it on every request and keep nothing.
- Scoping by a value from the payload. A client that can name the account it reads is a client that
can name somebody else's.
PATCH /merefusesidandexternal_subjectfor exactly this reason. - Using
bind_contextas authorization. It is for logs and correlation. Authorization decisions read the principal, which is an argument. - Forgetting to strip the assertion headers at the edge. This is the one operational failure mode
that turns this shape into a vulnerability: if the proxy forwards a client-supplied
X-Authenticated-Subject, a client chooses their own subject. The proxy must remove these headers from inbound requests and set them only after authenticating. Xyberos does no header hardening for you (security guide).
Exercise¶
Write the test-side version: build the application, replace the verifier with one that answers from a
fixed subject using app.with_overrides, and claim and read an account without any assertion headers
at all. Then explain to yourself why that is a test affordance and not something to deploy.
Show the answer
examples/learn_traditional/traditional08_exercise.py
"""Lesson 8, exercise — a verifier replaced by a fixed principal, for a test.
The answer to the lesson's exercise. `app.with_overrides` swaps the verifier for one that answers
from a constant, which is what a test wants: the subject is the thing under test's control, and no
signature has to be computed to express it.
"""
from __future__ import annotations
import tempfile
from pathlib import Path
from starlette.requests import Request
from user_management.app import create_app
from user_management.identity import Principal, PrincipalVerifier
from user_management.migrate import migrate
from xyberos import App
from xyberos.database import SqliteSettings
from xyberos.http import HttpAdapter
from xyberos.http.testing import create_test_client
class FixedVerifier(PrincipalVerifier):
"""The test double: no signing, no secret, one principal."""
def __init__(self, subject: str) -> None:
super().__init__("the-secret-is-irrelevant-for-a-fixed-principal")
self._subject = subject
def verify(self, request: Request) -> Principal:
"""Answer with the same principal for every request."""
return Principal(subject=self._subject, roles=frozenset({"member"}))
def app_with_fixed_principal(*, settings: SqliteSettings, subject: str) -> App:
"""The application under test, with the verifier it would use in production replaced."""
app = create_app(settings=settings, secret="unused")
return app.with_overrides({PrincipalVerifier: FixedVerifier(subject)})
def main() -> None:
"""Claim and read an account without a single assertion header."""
with tempfile.TemporaryDirectory() as directory:
settings = SqliteSettings(path=Path(directory) / "users.db")
migrate(settings)
app = app_with_fixed_principal(settings=settings, subject="ada@idp")
with create_test_client(HttpAdapter(app)) as client:
claimed = client.post(
"/me",
json={"email": "ada@example.com", "display_name": "Ada Lovelace"},
)
print(
"claimed with no headers:", claimed.status_code, claimed.json()["external_subject"]
)
shown = client.get("/me")
print("read with no headers:", shown.status_code, shown.json()["email"])
if __name__ == "__main__":
main()
Next¶
Lesson 9 is the other half of the decision: what a verified principal may do. Roles, 401 against
403, and the deployment obligation that keeps an assertion from being forgeable at the edge.