PyBook 06: Tenancy

Organisation ──► Project (live | test)  ──► EndUser ──► Connection
                    │  signing_secret        (usr_…)      (conn_…)
                    └► ApiKey (adk_live_… , scopes)

from auradefi.clock import FrozenClock
from auradefi.tenancy.models import ConnectionKind, Environment
from auradefi.tenancy.store import TenancyStore

T0 = 1_767_225_600_000  # 2026-01-01T00:00:00Z
EXTERNAL_ID = "host-user-1"           # the SAME opaque id in both tenants
DESCRIPTOR = "0xAbCd000000000000000000000000000000000001"  # the SAME address

SECRET_A, SECRET_B = "aa" * 32, "bb" * 32


def scripted_entropy(ids, secrets):
    "Width-keyed: n=8 pops an id, n=32 pops a signing secret."
    id_queue, secret_queue = list(ids), list(secrets)

    def entropy(n: int) -> str:
        if n == 8:
            return id_queue.pop(0)
        if n == 32:
            return secret_queue.pop(0)
        return "7" * (2 * n)

    return entropy


clock = FrozenClock(T0)
store = TenancyStore(scripted_entropy(["11" * 8, "a", "22" * 8, "b"], [SECRET_A, SECRET_B]))

org_a = store.create_organisation("Org A", clock)
proj_a = store.create_project(org_a.id, "tenant-a", Environment.LIVE, clock)
org_b = store.create_organisation("Org B", clock)
proj_b = store.create_project(org_b.id, "tenant-b", Environment.LIVE, clock)

assert (proj_a.id, proj_b.id) == ("proj_a", "proj_b")
assert proj_a.signing_secret == SECRET_A
assert proj_b.signing_secret == SECRET_B
assert proj_a.signing_secret != proj_b.signing_secret  # THE isolation root
print(proj_a)

Project(id='proj_a', org_id='org_1111111111111111', name='tenant-a', environment=<Environment.LIVE: 'live'>, signing_secret='aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', created_at=1767225600000)

from auradefi.tenancy.models import connection_id, end_user_id

user_a = store.get_or_create_user(proj_a.id, EXTERNAL_ID, clock)
user_b = store.get_or_create_user(proj_b.id, EXTERNAL_ID, clock)
conn_a = store.create_connection(proj_a.id, user_a.id, ConnectionKind.ADDRESS, DESCRIPTOR, clock)
conn_b = store.create_connection(proj_b.id, user_b.id, ConnectionKind.ADDRESS, DESCRIPTOR, clock)

assert user_a.external_user_id == user_b.external_user_id == EXTERNAL_ID
assert conn_a.descriptor == conn_b.descriptor == DESCRIPTOR.lower()  # canonical form
assert user_a.id != user_b.id and conn_a.id != conn_b.id

# Derived, so recomputable from first principles:
assert user_a.id == end_user_id(proj_a.id, EXTERNAL_ID)
assert conn_a.id == connection_id(proj_a.id, user_a.id, ConnectionKind.ADDRESS, DESCRIPTOR)
# ...and idempotent: a second get-or-create returns the same row.
assert store.get_or_create_user(proj_a.id, EXTERNAL_ID, clock) == user_a

print(f"tenant A: {user_a.id}  {conn_a.id}")
print(f"tenant B: {user_b.id}  {conn_b.id}   <- same inputs, different tenant")

tenant A: usr_2b67bf34d2444625  conn_eaa947a5814c21e3
tenant B: usr_2414ea20b3d09c41  conn_3d9ab7d47cb3733d   <- same inputs, different tenant

from auradefi.errors import AuthError
from auradefi.tenancy.keys import ApiKeyStore, has_scope
from auradefi.tenancy.models import Scope

keys = ApiKeyStore(scripted_entropy(["cc" * 8, "dd" * 8], ["ee" * 32]))
key, plaintext = keys.issue(proj_a.id, Environment.LIVE, (Scope.USERS_ADMIN, Scope.ACCOUNTS_READ), clock)

assert plaintext.startswith("adk_live_") and len(plaintext) == 57
assert key.secret_hash != plaintext           # only the hash is retained
assert has_scope(key, Scope.ACCOUNTS_READ)
assert not has_scope(key, Scope.ACCOUNTS_WRITE)
assert keys.authenticate(plaintext, clock).id == key.id
print("issued", plaintext[:14] + "…", "scopes:", sorted(str(s) for s in key.scopes))

keys.revoke(proj_a.id, key.id, clock)  # tenant-gated (0.1.1 #25)
try:
    keys.authenticate(plaintext, clock)
except AuthError as exc:
    print("after revoke:", exc)

issued adk_live_77777… scopes: ['accounts:read', 'users:admin']
after revoke: api key failed authentication

from auradefi.tenancy.audit import AuditLog
from auradefi.tenancy.tokens import RevocationSet, require_scope, verify_token

audit = AuditLog()
minting_key, _ = ApiKeyStore(scripted_entropy(["ab" * 8], ["cd" * 32])).issue(
    proj_a.id, Environment.LIVE, (Scope.USERS_ADMIN,), clock
)

token = store.mint_user_token(
    proj_a.id, EXTERNAL_ID, ["accounts:read"], 600_000,
    "203.0.113.7", minting_key.id, clock, audit, jti="ab" * 16,
)
claims = verify_token(token, signing_secret=proj_a.signing_secret, clock=clock)

assert len(token.split(".")) == 3                 # header.payload.signature
assert claims.project_id == proj_a.id
assert claims.external_user_id == EXTERNAL_ID
assert claims.scopes == ("accounts:read",)
assert claims.exp - claims.iat == 600_000         # ten minutes, ms epoch
require_scope(claims, "accounts:read")
print("claims:", claims)

(record,) = audit.entries(proj_a.id)
assert (record.seq, record.event, record.key_id, record.ip) == (1, "token.minted", minting_key.id, "203.0.113.7")
print("audited:", record)

revoked = RevocationSet()
revoked.revoke(claims.jti)
try:
    verify_token(token, signing_secret=proj_a.signing_secret, clock=clock, revoked=revoked)
except AuthError as exc:
    print("after revocation:", exc)

claims: TokenClaims(external_user_id='host-user-1', project_id='proj_a', scopes=('accounts:read',), iat=1767225600000, exp=1767226200000, jti='abababababababababababababababab')
audited: AuditRecord(seq=1, event='token.minted', project_id='proj_a', external_user_id='host-user-1', key_id='key_abababababababab', ip='203.0.113.7', at_ms=1767225600000, ip_source='unknown')
after revocation: token revoked

control = verify_token(token, signing_secret=proj_a.signing_secret, clock=clock)
assert control.project_id == proj_a.id  # valid under its OWN project

try:
    verify_token(token, signing_secret=proj_b.signing_secret, clock=clock)
except AuthError as exc:
    assert type(exc) is AuthError  # not a subclass: no information leaked
    print("A's token under B's secret ->", type(exc).__name__, "-", exc)
else:
    raise AssertionError("cross-tenant token verified: isolation is broken")

A's token under B's secret -> AuthError - token failed authentication

from auradefi.errors import NotFoundError

ABSENT = "conn_0000000000000000"

try:
    store.get_connection(proj_b.id, conn_a.id)
except NotFoundError as cross:
    smuggled = cross
try:
    store.get_connection(proj_b.id, ABSENT)
except NotFoundError as missing:
    absent = missing

assert type(smuggled) is type(absent) is NotFoundError
assert proj_a.id not in str(smuggled)
assert proj_a.signing_secret not in str(smuggled)
print("smuggled:", smuggled)
print("absent:  ", absent)

# Enumeration gets nowhere either: disjoint user sets, no cross-project list.
ids_a = {user.id for user in store.users(proj_a.id)}
ids_b = {user.id for user in store.users(proj_b.id)}
assert ids_a and ids_b and ids_a.isdisjoint(ids_b)
print("A's users:", ids_a, "B's users:", ids_b)

smuggled: connection not found: 'conn_eaa947a5814c21e3'
absent:   connection not found: 'conn_0000000000000000'
A's users: {'usr_2b67bf34d2444625'} B's users: {'usr_2414ea20b3d09c41'}

from auradefi.errors import QuotaExceededError
from auradefi.tenancy.quota import QuotaCounter, QuotaLimits

shared_audit = AuditLog()
quota = QuotaCounter(QuotaLimits(per_second=2, per_day=100, per_month=100), clock)


def mint(project_id, jti):
    return store.mint_user_token(
        project_id, EXTERNAL_ID, ["accounts:read"], 600_000,
        "203.0.113.7", minting_key.id, clock, shared_audit, quota=quota, jti=jti,
    )


mint(proj_a.id, "11" * 16)
mint(proj_a.id, "22" * 16)
try:
    mint(proj_a.id, "33" * 16)
except QuotaExceededError as exc:
    print("A's window is spent:", exc)

assert len(shared_audit.entries(proj_a.id)) == 2   # the refused mint is NOT audited
assert shared_audit.entries(proj_b.id) == ()       # B's trail is empty

b_token = mint(proj_b.id, "44" * 16)               # B is unthrottled
assert verify_token(b_token, signing_secret=proj_b.signing_secret, clock=clock).project_id == proj_b.id
print("B mints fine; snapshot:", {k: v.remaining for k, v in quota.snapshot(proj_b.id).items()})

A's window is spent: quota exceeded in the 'second' window for 'proj_a': limit 2, window resets at 1767225601000
B mints fine; snapshot: {'second': 1, 'day': 99, 'month': 99}

import inspect

public = {name for name in dir(store) if not name.startswith("_")}
assert public == {
    "create_organisation", "create_project", "get_or_create_user", "users",
    "create_connection", "get_connection", "connections", "mint_user_token",
}
for name in ("get_or_create_user", "users", "create_connection", "get_connection",
             "connections", "mint_user_token"):
    assert list(inspect.signature(getattr(store, name)).parameters)[0] == "project_id", name
print("project-scoped first argument on every tenant-data method:", sorted(public))

project-scoped first argument on every tenant-data method: ['connections', 'create_connection', 'create_organisation', 'create_project', 'get_connection', 'get_or_create_user', 'mint_user_token', 'users']