Lesson 2 — a verified principal, and what it may do¶
What you will build¶
Who is calling. Xyberos ships no identity model and no authorization model, and
ADR 0014 says why: a framework cannot
guess what your assertion looks like. What it provides is the shape — a component that verifies
rather than believes, a principal that travels as a required argument, and HttpError subclasses whose
status codes you choose.
The one thing this lesson adds to the traditional path's version is the tenant, and it is here rather than in the request for a reason: a payload that names a tenant cannot move a lookup, because the lookup never reads the payload.
What you need¶
Nothing new.
examples/support_desk/security.py
"""Who is calling, established by something that checked the assertion.
Xyberos ships no identity model and no authorization model (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 a required argument.
The tenant is part of the principal rather than part of the request, which is the difference
between a multi-tenant application and a claim a client can edit: a payload that names a tenant
cannot change which rows the service looks at.
"""
from __future__ import annotations
import hashlib
import hmac
import os
from collections.abc import Iterable
from dataclasses import dataclass
from starlette.requests import Request
from xyberos import HttpError
SUBJECT_HEADER = "X-Authenticated-Subject"
TENANT_HEADER = "X-Authenticated-Tenant"
ROLES_HEADER = "X-Authenticated-Roles"
SIGNATURE_HEADER = "X-Principal-Signature"
SECRET_ENV = "XYBEROS_IDENTITY_SECRET"
REQUESTER = "requester"
SUPPORT_AGENT = "support_agent"
SUPPORT_LEAD = "support_lead"
#: The two roles that may work on tickets they did not open.
STAFF = (SUPPORT_AGENT, SUPPORT_LEAD)
@dataclass(frozen=True)
class Principal:
"""Who is calling, which tenant they are in, and what the verifier established about them."""
subject: str
tenant: str
roles: frozenset[str] = frozenset()
def holds(self, *roles: str) -> bool:
"""Whether the principal carries any of these roles."""
return any(role in self.roles for role in roles)
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 may not do this; exposed as a 403."""
def __init__(self, requirement: str) -> None:
super().__init__(f"this operation requires {requirement}", status_code=403)
def require_role(principal: Principal, *roles: str) -> None:
"""Refuse a principal that carries none of these roles, as a `403`."""
if not principal.holds(*roles):
raise ForbiddenError(" or ".join(roles))
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, "")
tenant = request.headers.get(TENANT_HEADER, "")
roles = request.headers.get(ROLES_HEADER, "")
signature = request.headers.get(SIGNATURE_HEADER, "")
if not subject or not tenant:
raise UnauthenticatedError(f"{SUBJECT_HEADER} and {TENANT_HEADER} are both required")
if not hmac.compare_digest(signature, self.sign(subject, tenant, roles)):
raise UnauthenticatedError("the assertion signature does not match")
return Principal(subject=subject, tenant=tenant, roles=_roles(roles))
def sign(self, subject: str, tenant: str, roles: str = "") -> str:
"""Return the signature a caller must present for this subject, tenant, and roles."""
message = "\n".join((subject, tenant, roles)).encode()
return hmac.new(self._secret, message, hashlib.sha256).hexdigest()
def headers(self, subject: str, tenant: str, roles: Iterable[str] = ()) -> dict[str, str]:
"""Return the headers a client presents after authenticating successfully."""
joined = ",".join(roles)
return {
SUBJECT_HEADER: subject,
TENANT_HEADER: tenant,
ROLES_HEADER: joined,
SIGNATURE_HEADER: self.sign(subject, tenant, joined),
}
def _roles(raw: str) -> frozenset[str]:
return frozenset(role.strip() for role in raw.split(",") if role.strip())
examples/learn_hybrid/hybrid02_principal.py
"""Lesson 2 — a verified principal, and the permission matrix it is measured against.
Run it with:
python -m learn_hybrid.hybrid02_principal
"""
from __future__ import annotations
import tempfile
from pathlib import Path
from support_desk.app import create_app
from support_desk.migrate import migrate
from support_desk.security import PrincipalVerifier
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)
ADA = VERIFIER.headers("ada@example.com", "acme", ["requester"])
GRACE = VERIFIER.headers("grace@example.com", "acme", ["requester"])
SAM = VERIFIER.headers("sam@desk", "acme", ["support_agent"])
#: The same agent in somebody else's tenant: the same roles, and no sight of acme's tickets.
OTHER_TENANT = VERIFIER.headers("sam@desk", "globex", ["support_agent"])
def main() -> None:
"""Ask three principals for the same rows, and one of them for something they may not do."""
with tempfile.TemporaryDirectory() as directory:
settings = SqliteSettings(path=Path(directory) / "support.db")
migrate(settings)
app = create_app(settings=settings, secret=SECRET)
with create_test_client(HttpAdapter(app)) as client:
client.post(
"/tickets",
json={"subject": "Where is my refund?", "body": "Order 42 came back on Monday."},
headers=ADA,
)
client.post(
"/tickets",
json={"subject": "The badge reader is broken", "body": "It flashes."},
headers=ADA,
)
anonymous = client.get("/tickets")
print("no assertion ->", anonymous.status_code, anonymous.json()["error"]["message"])
forged = dict(ADA, **{"X-Principal-Signature": "0" * 64})
refused = client.get("/tickets", headers=forged)
print("a forged signature ->", refused.status_code, refused.json()["error"]["message"])
ada = client.get("/tickets", headers=ADA).json()
print("ada sees:", ada["total"], [item["subject"] for item in ada["items"]])
print("grace sees:", client.get("/tickets", headers=GRACE).json()["total"])
print("sam (an agent) sees:", client.get("/tickets", headers=SAM).json()["total"])
denied = client.post("/tickets/1/classify", headers=GRACE)
print(
"a requester classifying ->",
denied.status_code,
denied.json()["error"]["message"],
)
print(
"another tenant's agent sees:",
client.get("/tickets", headers=OTHER_TENANT).json()["total"],
)
missing = client.get("/tickets/1", headers=OTHER_TENANT)
print(
"and reading acme's ticket 1 ->",
missing.status_code,
missing.json()["error"]["message"],
)
if __name__ == "__main__":
main()
PrincipalVerifier.verify is the only place a request becomes a fact. Until it returns, nothing in the
request is known: the headers are assertions that arrived, and they are trusted because a shared secret
signed them, not because they said so.
How to run it¶
What it prints¶
no assertion -> 401 the request is not authenticated: X-Authenticated-Subject and X-Authenticated-Tenant are both required
a forged signature -> 401 the request is not authenticated: the assertion signature does not match
ada sees: 2 ['Where is my refund?', 'The badge reader is broken']
grace sees: 0
sam (an agent) sees: 2
a requester classifying -> 403 this operation requires support_agent or support_lead
another tenant's agent sees: 0
and reading acme's ticket 1 -> 404 ticket 1 was not found
Read the last two lines together. Another tenant's agent sees nothing — 0 tickets — and reading
acme's ticket by its id is a 404, not a 403. That is deliberate: a 403 would confirm that ticket
1 exists, which is the one thing a caller in another tenant must not learn. The tenant is part of the
query, so there is no row to refuse.
The 403 for the requester who tried to classify is the other half: the subject is known, the answer
is no, and the caller can tell the difference between "authenticate" and "you may not".
Common mistakes¶
- Trusting a header because it is there. A header is not a fact until something verified it. The forged-signature line is what that looks like when it fails closed.
- A check after the load.
_rowinservice.pylooks upidand tenant together, so there is no window in which a ticket from another tenant is in memory. - Roles from the request body. Roles come from the verified assertion. A role a client can name is not a role.
403where404belongs. Existence is information. For a resource in another tenant, the honest answer is that there is no such resource for you.- An authorization helper that returns a boolean. A caller can ignore a boolean.
require_roleraises, so the only way past it is through it.
Next¶
Lesson 3 writes down what each decision was: an audit trail whose rows are written in the same transaction as the change they describe.