From 2209e2395c6412906c769b6f1fdbc6419aebddc7 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Thu, 1 Oct 2026 05:40:32 +0000 Subject: [PATCH] Satellite v2: upstream-minted scoped sync tokens, per-session contexts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Registration simplified to a single sync URL: minted in the upstream's admin UI (Satellite access section of the domain dialog), carrying the token as userinfo — https://@. Tokens live hashed in the upstream database, scoped to one rp-id, revocable (drops live connections). PASKIA_SYNC_TOKENS env is gone. Data minimization: the sync feed now carries only the scope's sessions and credentials; the user graph never leaves the upstream. Each session upsert embeds its master-computed ApiSessionContext (identity display data + effective permissions for the session's host); user-graph mutations eagerly re-push affected sessions with fresh contexts, so revocations land within ~1 RTT. /check and /user-info are proxied to the upstream (they need the user graph). Satellite side: replicas keyed by rp-id (no more shared-replica config conflicts — each remote domain has its own token), the replica refuses to go live on a ready-message scope mismatch, write-backs are confined to the token's scope server-side. Admin dialog: one sync-URL paste field (token split from userinfo, write-only), uplink verification required to save (reachability, rp-id match, no nesting, auth host present); a verified uplink takes over the domain's origins (**. + upstream auth host) and auth host. New domains adopt the upstream's rp-id and name. --- docs/RemoteProxy.md | 106 ++++-- .../src/admin/dialogs/DomainEditDialog.vue | 323 ++++++++++++---- frontend/src/assets/style.css | 6 + paskia/db/__init__.py | 6 + paskia/db/operations.py | 25 ++ paskia/db/structs.py | 37 +- paskia/domains.py | 24 -- paskia/fastapi/admin/domains.py | 93 ++++- paskia/fastapi/api.py | 8 + paskia/fastapi/sync.py | 79 +++- paskia/satellite.py | 131 +++++-- paskia/syncfeed.py | 121 ++++-- tests/test_remote.py | 344 ++++++++++++++---- tests/test_satellite_e2e.py | 186 +++++++++- 14 files changed, 1159 insertions(+), 330 deletions(-) diff --git a/docs/RemoteProxy.md b/docs/RemoteProxy.md index da53206..10c3132 100644 --- a/docs/RemoteProxy.md +++ b/docs/RemoteProxy.md @@ -20,9 +20,25 @@ alternatives is at the end of this document. ## Configuration -A domain becomes remote in the admin domains UI (master admin, on the -primary server's auth host — this configuration itself never touches a -remote): enable *Remote instance* and set the remote URL and sync token. +Setup is a single paste. On the **upstream** (the authoritative instance), +open the domain in the admin UI and use *Satellite access → Generate sync +URL*. That mints a token scoped to the domain's rp-id and returns one URL +carrying everything the satellite needs: + +``` +https://@auth.example.com +``` + +On the **satellite**, open (or add) the same domain in its admin UI, +enable *Remote instance*, and paste the sync URL. The dialog probes the +upstream's `/auth/api/settings` as you type: it must be reachable, serve +this very rp-id authoritatively (a satellite cannot upstream another +satellite), and have an auth host configured — only such an instance can +serve satellites at all. Once the uplink verifies, the remote +configuration takes over completely: the domain's origins become +`**.` plus the upstream's auth host, and saving requires the +verified uplink. + In the stored config (`DomainConfig.remote`): ```json @@ -34,32 +50,31 @@ In the stored config (`DomainConfig.remote`): } ``` -- `url` — the remote instance's base URL. The satellite connects to - `{url}/auth/api/sync/ws`; the connection is server-to-server and not - host-dispatched, so internal addresses work. The admin dialog probes - `{url}/auth/api/settings` as you type: it must be reachable, serve this - very rp-id, and not be a satellite for it itself (nested satellites are - not supported — always point at the authoritative instance). Probe - findings are warnings; the server-side validation is authoritative. -- `token` — bearer token for the sync channel. The **remote accepts tokens - via its `PASKIA_SYNC_TOKENS` environment variable** (comma-separated); - nothing is stored in the remote's database, and with the variable unset - the sync endpoint stays closed. The token is write-only over the admin - API (an empty field keeps the stored one). +- `url` — the upstream's auth host URL (the userinfo token is split off + before storing). The satellite connects to `{url}/auth/api/sync/ws`; + the connection is server-to-server and not host-dispatched, so internal + addresses work. +- `token` — bearer token for the sync channel, stored **hashed in the + upstream's database** and scoped to one rp-id. Managed in the upstream + admin UI (the domain's *Satellite access* section: generate, list, + revoke); the raw token is shown once, inside the sync URL. On the + satellite it is write-only over the admin API (a sync URL without + userinfo keeps the stored token). Revoking a token drops the satellite's + live connections immediately. - `cache_ttl` — seconds the replica remains trusted after the sync channel goes down; then checks fail closed (503). Set it large (up to the 24 h session lifetime) for fail-open behavior. Minimum 1. - `refresh_interval` — seconds between reconnects; every connect starts from a full snapshot, which reconciles any drift. Minimum 30. -Several domains may share one remote (they share a single replica); their -remote settings must then be identical — conflicting configurations are -rejected at validation and aligned with a warning at load. +Each remote domain gets its own replica and its own token; several domains +may point at the same upstream instance independently. -A remote domain **must mark an auth host** (validated cross-domain and in -the UI): the profile, admin and sign-in pages live there, so browsers and -WebSockets go directly to the remote. Other domains on the same satellite -remain fully local — the multi-domain config mixes both kinds freely. +A remote domain **must mark an auth host** (the uplink's own — the dialog +derives it): the profile, admin and sign-in pages live there, so browsers +and WebSockets go directly to the upstream. Other domains on the same +satellite remain fully local — the multi-domain config mixes both kinds +freely. ## How it works @@ -73,12 +88,20 @@ write-behind for remote, `db.update_session` for local; `satellite.evict_session` on logout). `satellite.forward_request(request)` returns the proxied response for remote domains or `None` for local ones. -**The replica** is a plain `DB` struct instance in RAM, never persisted. -On connect the remote sends a snapshot of the replicated tables -(permissions, orgs, roles, users, credentials, sessions), then live -upsert/delete events emitted from the struct `store()`/`delete()` hooks -(which also cover cascade deletes) and field-mutating operations. A -single ordered WebSocket cannot gap; a slow subscriber is dropped and +**The replica** is a `ReplicaDB` instance in RAM, never persisted. On +connect the upstream sends a snapshot of the replicated tables +(the domain's credentials and sessions), then live upsert/delete events +emitted from the struct `store()`/`delete()` hooks (which also cover +cascade deletes) and field-mutating operations. The feed is scoped to the +token's rp-id, and the user graph never leaves the upstream: every session +upsert carries an embedded `ctx` — the same `ApiSessionContext` sent to +users — holding the session's user/org/role display data and effective +permissions, computed for the session's bound host. Mutations of users, +orgs, roles or permissions are not published; instead the upstream eagerly +re-pushes the affected sessions with freshly computed contexts, so renames +and permission revocations take effect within ~1 RTT. The replica refuses +to go live when the scope in the ready message does not match its domain. +A single ordered WebSocket cannot gap; a slow subscriber is dropped and resyncs. The feed carries no usable secrets: sessions are keyed by `hash_secret` output, credentials carry public keys only, and the OIDC signing key is never replicated. @@ -87,8 +110,9 @@ signing key is never replicated. | Endpoint | Handling | |---|---| -| `GET /auth/api/forward`, `GET /check`, `GET /user-info`, `GET /settings` | served from the replica (sub-ms) | +| `GET /auth/api/forward`, `GET /settings` | served from the replica (sub-ms) | | `POST /auth/api/validate` | verified from the replica; the throttled refresh updates the replica and is written back over the sync channel; cookie renewed locally | +| `GET /check`, `GET /user-info` | proxied — these need the user graph (per-session contexts do not cover arbitrary user lookups or full profile data) | | `POST /auth/api/logout` | proxied to the remote (original Host preserved) and evicted from the replica immediately | | `POST /auth/api/set-session`, `GET /token-info` | proxied (the exchange code/reset token lives on the remote); the session arrives via sync event | | `/auth/oidc/*` | proxied (signing key and OIDC sessions stay on the remote) | @@ -112,18 +136,22 @@ Freshness hierarchy: ## The remote side -Strictly additive and RAM-only: `syncfeed` (a subscriber set fed by the -commit hooks) and the token-gated `/auth/api/sync/ws` endpoint serving -snapshot + live events and accepting `session_refresh` write-backs. With -no satellites connected, the hooks are a no-op. +Strictly additive and RAM-only besides the token table: `syncfeed` (a +subscriber set fed by the commit hooks) and the `/auth/api/sync/ws` +endpoint serving snapshot + live events and accepting `session_refresh` +write-backs within the token's scope. With no satellites connected, the +hooks are a no-op. ## Trust and caveats -- The satellite host holds a full copy of the remote's auth data in RAM - (minus the OIDC key) — treat it as trusted as the remote. -- Avatars are stored on the remote's disk; `user-info` from a replica - reports no avatar URL. -- OIDC sessions in a replica-backed `user-info` show the client UUID +- The satellite host holds only its own domain's sessions (with their + identity contexts) and credentials in RAM — it learns the users that + actually sign in to its domain, never the directory, and never the OIDC + key. Still treat it as trusted: it is authoritative for session checks + within `cache_ttl`. +- Avatars are stored on the remote's disk; profile data such as + `user-info` is proxied to the remote rather than answered locally. +- OIDC sessions in a proxied `user-info` show the client UUID rather than its name (OIDC clients are not replicated). --- @@ -175,7 +203,7 @@ within one RTT instead of at TTL. meaning shifts under the caller. Dispatch is instead keyed explicitly by the request host.) - A sync protocol (snapshot + live events + reconnect reconciliation). -- A trusted satellite host (full data copy in RAM). +- A trusted satellite host (its domain's sessions and credentials in RAM). - Additive remote code (sync endpoint + commit hooks), where A needs none. - Replica housekeeping (expiry sweeper, write-behind, optimistic diff --git a/frontend/src/admin/dialogs/DomainEditDialog.vue b/frontend/src/admin/dialogs/DomainEditDialog.vue index 4eb7745..9ac5bd7 100644 --- a/frontend/src/admin/dialogs/DomainEditDialog.vue +++ b/frontend/src/admin/dialogs/DomainEditDialog.vue @@ -2,7 +2,8 @@ import { computed, nextTick, onBeforeUnmount, onMounted, ref, watch } from 'vue' import AdminDialog from './AdminDialog.vue' import { useAuthStore } from '@/stores/auth' -import { compareOrigins } from '@/utils/helpers' +import { compareOrigins, formatDate } from '@/utils/helpers' +import { apiJson } from 'paskia' const props = defineProps({ dialog: { type: Object, required: true } @@ -22,9 +23,11 @@ const dialogRpId = computed(() => (props.dialog.data?.rp_id || '').trim().toLowe // // A remote domain is served from another paskia instance: this one keeps a // RAM-only read replica for fast local session checks and forwards -// mutations. The remote must accept our sync token via its -// PASKIA_SYNC_TOKENS environment variable. An auth host (the remote's) is -// required — profile, admin and sign-in pages live there. +// mutations. Setup is a single sync URL minted on the upstream (its admin +// UI, the domain's Satellite access section): it carries the upstream auth +// host and a token scoped to this domain. The remote configuration takes +// over completely — allowed origins and the auth host are derived from the +// uplink once it verifies. const remoteEnabled = computed({ get: () => !!props.dialog.data?.remote, set: on => { @@ -34,39 +37,74 @@ const remoteEnabled = computed({ }, }) -const remoteUrlInvalid = computed(() => { - const url = props.dialog.data?.remote?.url?.trim() - if (!url) return false - return !/^https?:\/\/[^\s/]+/.test(url) +// The raw sync URL as pasted; the parsed parts land in remote.url (clean, +// userinfo stripped) and remote.token. An existing configuration shows the +// clean URL — the stored token is write-only, and a URL without userinfo +// keeps it. +const syncUrl = ref(props.dialog.data?.remote?.url || '') +const hadRemoteOnOpen = !!props.dialog.data?.remote + +function parseSyncUrl(value) { + const v = value.trim() + if (!v) return null + try { + const u = new URL(v.includes('://') ? v : 'https://' + v) + if (u.protocol !== 'http:' && u.protocol !== 'https:') return { error: true } + if (!u.hostname) return { error: true } + const token = decodeURIComponent(u.password || u.username || '') + const url = u.origin + (u.pathname === '/' ? '' : u.pathname.replace(/\/+$/, '')) + return { url, token, host: u.host } + } catch { + return { error: true } + } +} + +const parsedSyncUrl = computed(() => parseSyncUrl(syncUrl.value)) +const syncUrlInvalid = computed(() => !!(syncUrl.value.trim() && parsedSyncUrl.value?.error)) +// A token is required unless the domain already has one stored (write-only) +const syncUrlMissingToken = computed(() => { + const parsed = parsedSyncUrl.value + return !!(parsed && !parsed.error && !parsed.token && !hadRemoteOnOpen) }) -// Remote domains must mark an auth host (the server rejects the save) -const remoteMissingAuthHost = computed( - () => !!props.dialog.data?.remote && !props.dialog.data?.auth_host -) - -// --- Uplink verification --- -// Fetch the remote's /auth/api/settings (CORS-open), like the origin -// connectivity probes: proves reachability, that the peer is a paskia -// instance, and that it serves this very domain authoritatively — not as -// a satellite itself (nested satellites are not supported). Results are -// warnings, never submit blockers (the remote may not be deployed yet). -const uplinkCheck = ref(null) // null|validating|valid|unreachable|mismatch|nested -let uplinkTimer = null -let uplinkSeq = 0 +// Keep the submit model in sync with the pasted URL watch( - () => [props.dialog.data?.remote?.url, dialogRpId.value], - () => { - clearTimeout(uplinkTimer) - uplinkCheck.value = null - const url = props.dialog.data?.remote?.url?.trim().replace(/\/+$/, '') - if (!url || remoteUrlInvalid.value) return - uplinkTimer = setTimeout(() => checkUplink(url), 600) + parsedSyncUrl, + parsed => { + const d = props.dialog.data + if (!d?.remote) return + if (!parsed || parsed.error) { + d.remote.url = '' + return + } + d.remote.url = parsed.url + if (parsed.token) d.remote.token = parsed.token }, { immediate: true } ) -async function checkUplink(url) { +// --- Uplink verification --- +// Fetch the upstream's /auth/api/settings (CORS-open), like the origin +// connectivity probes: proves reachability, that the peer serves this very +// domain authoritatively — not as a satellite itself (nested satellites are +// not supported) — and that it has an auth host (where profile, admin and +// sign-in pages live for satellite users). A verified uplink is required to +// save a remote domain. +const uplinkCheck = ref(null) // null|validating|valid|unreachable|mismatch|nested|no-auth-host +let uplinkTimer = null +let uplinkSeq = 0 +watch( + () => [props.dialog.data?.remote ? parsedSyncUrl.value : null, dialogRpId.value], + ([parsed]) => { + clearTimeout(uplinkTimer) + uplinkCheck.value = null + if (!parsed || parsed.error) return + uplinkTimer = setTimeout(() => checkUplink(parsed.url, parsed.host), 600) + }, + { immediate: true } +) + +async function checkUplink(url, host) { const seq = ++uplinkSeq uplinkCheck.value = 'validating' try { @@ -76,8 +114,28 @@ async function checkUplink(url) { if (seq !== uplinkSeq) return // url/rp-id changed while fetching if (!response.ok) throw new Error('not ok') const data = await response.json() - uplinkCheck.value = - data.rp_id !== dialogRpId.value ? 'mismatch' : data.remote ? 'nested' : 'valid' + const d = props.dialog.data + // An existing (or already typed) rp-id must match; a fresh domain + // adopts the upstream's, along with its display name. + if (dialogRpId.value && data.rp_id !== dialogRpId.value) { + uplinkCheck.value = 'mismatch' + return + } + uplinkCheck.value = data.remote + ? 'nested' + : !data.auth_host + ? 'no-auth-host' + : 'valid' + if (uplinkCheck.value !== 'valid' || !d?.remote) return + if (d.isNew && !dialogRpId.value) { + d.rp_id = data.rp_id + if (!d.rp_name) d.rp_name = data.rp_name || '' + } + // The remote config takes over: origins are the whole rp-id domain + // plus the uplink's auth host, which hosts all account pages. + d.auth_host = host + d.origins = ['**.' + data.rp_id, host] + d.originValidation = [null, null] } catch { if (seq === uplinkSeq) uplinkCheck.value = 'unreachable' } @@ -85,19 +143,22 @@ async function checkUplink(url) { // Block submit on hard errors: malformed entries, an over-cap related // list (the server rejects the save), a save that would lock the admin -// out of the domain they are using, or validation still in flight. -// Connectivity and rp-id mismatch results are warnings only (entries may -// be hosted elsewhere, or a new domain whose DNS is not routed to this -// instance yet). +// out of the domain they are using, or validation still in flight. For +// remote domains the uplink must verify — it doubles as the source of +// the derived origins and auth host. const isValidationInvalid = computed(() => { const d = props.dialog.data + if (d?.remote) { + if (!parsedSyncUrl.value || parsedSyncUrl.value.error) return true + if (syncUrlMissingToken.value) return true + if (uplinkCheck.value !== 'valid') return true + return d.isNew && !isWellFormedDomain(d.rp_id || '') + } const bad = v => v === 'invalid' || v === 'validating' if (d.originValidation?.some(bad)) return true if (relatedEntries.value.length > 5) return true if (d.isNew && !isWellFormedDomain(d.rp_id || '')) return true if (lockoutWarning.value) return true - if (remoteUrlInvalid.value || remoteMissingAuthHost.value) return true - if (d.remote && !d.remote.url?.trim()) return true return false }) @@ -141,6 +202,68 @@ function copyText(value, label) { }) } +// --- Satellite access (this instance as the upstream) --- +// +// Tokens are scoped to this domain: a satellite receives only this +// domain's sessions (each with its user's identity context and effective +// permissions) and credentials — never the user directory — and can only +// write back session refreshes within scope. The raw token is shown +// once, embedded in the sync URL; revoking drops the satellite's live +// connections. Only an authoritative domain with an auth host can serve +// satellites (enforced by the server). +const syncTokens = ref(null) // null = not loaded yet +const mintedSyncUrl = ref('') +const satelliteBusy = ref(false) +const showSatelliteAccess = computed(() => { + const d = props.dialog.data + return !!(d && !d.isNew && !d.remote) +}) + +async function loadSyncTokens() { + const d = props.dialog.data + if (!showSatelliteAccess.value) return + try { + syncTokens.value = await apiJson( + `/auth/api/admin/domains/${d.rp_id}/sync-tokens`, { method: 'GET' } + ) + } catch { + syncTokens.value = [] + } +} +onMounted(loadSyncTokens) + +async function generateSyncToken() { + const d = props.dialog.data + if (!d || satelliteBusy.value) return + satelliteBusy.value = true + try { + const data = await apiJson( + `/auth/api/admin/domains/${d.rp_id}/sync-token`, { method: 'POST' } + ) + mintedSyncUrl.value = data.url + await loadSyncTokens() + } catch (e) { + authStore.showMessage(e.message || 'Failed to generate the sync URL', 'error') + } finally { + satelliteBusy.value = false + } +} + +async function revokeSyncToken(id) { + const d = props.dialog.data + if (!d || satelliteBusy.value) return + satelliteBusy.value = true + try { + await apiJson(`/auth/api/admin/domains/${d.rp_id}/sync-token/${id}`, { method: 'DELETE' }) + authStore.showMessage('Sync token revoked.', 'success', 2500) + await loadSyncTokens() + } catch (e) { + authStore.showMessage(e.message || 'Failed to revoke the sync token', 'error') + } finally { + satelliteBusy.value = false + } +} + // --- Lockout prevention (editing the domain in use) --- // When the admin edits the domain they are currently signed in on and no @@ -410,7 +533,8 @@ watch(() => relatedEntries.value.map(asHttpsOrigin).join('|'), testWellKnown, { // edits while it is still the untouched prefilled row; once the admin // edits it, it is left alone. Seeding waits for a complete-looking rp-id // (letters after the final dot) so mid-typing states like 'something.' -// don't prefill a broken '**.something'. +// don't prefill a broken '**.something'. Remote domains derive their +// origins from the uplink instead. function looksCompleteDomain(value) { const host = (value || '').trim().replace(/\.$/, '') return host === 'localhost' || /\.[a-z]{2,}$/i.test(host) @@ -419,7 +543,7 @@ function looksCompleteDomain(value) { let seededOrigin = null watch(dialogRpId, rp => { const d = props.dialog.data - if (!d?.isNew) return + if (!d?.isNew || d.remote) return if (!looksCompleteDomain(rp) || !isWellFormedDomain(rp)) return const seed = '**.' + rp.trim().replace(/\.$/, '') if (!d.origins.length) { @@ -554,35 +678,37 @@ function onRemoveOrigin(i) { -
- Allowed Origins - -
-
-
- - 🔑 - 🔗 -
- -
- - - +
+ + @@ -641,6 +798,12 @@ function onRemoveOrigin(i) { background: var(--color-error-bg, rgba(239, 68, 68, 0.05)); } -.remote-toggle { display: flex; align-items: center; gap: var(--space-xs); font-weight: 600; font-size: 0.95rem; } +/* The global label rule stacks children vertically — the toggle wants its + checkbox beside the text. */ +.remote-toggle { display: flex; flex-direction: row; align-items: center; gap: var(--space-xs); font-weight: 600; font-size: 0.95rem; } .remote-toggle input { width: auto; } + +.sync-token-list { display: flex; flex-direction: column; gap: 0.3rem; margin-bottom: var(--space-xs); } +.sync-token-row { display: flex; align-items: center; gap: var(--space-sm); } +.sync-token-id { font-family: var(--font-mono, monospace); } diff --git a/frontend/src/assets/style.css b/frontend/src/assets/style.css index e7f1fce..ac6867b 100644 --- a/frontend/src/assets/style.css +++ b/frontend/src/assets/style.css @@ -326,6 +326,9 @@ input[type="text"], input[type="search"], input[type="email"], input[type="tel"], +input[type="password"], +input[type="number"], +input[type="url"], textarea, select { font: inherit; @@ -343,6 +346,9 @@ input[type="text"]:focus, input[type="search"]:focus, input[type="email"]:focus, input[type="tel"]:focus, +input[type="password"]:focus, +input[type="number"]:focus, +input[type="url"]:focus, textarea:focus, select:focus { outline: none; diff --git a/paskia/db/__init__.py b/paskia/db/__init__.py index 55ba7ae..5a12bef 100644 --- a/paskia/db/__init__.py +++ b/paskia/db/__init__.py @@ -31,6 +31,7 @@ from paskia.db.operations import ( create_permission, create_reset_token, create_role, + create_sync_token, create_user, delete_credential, delete_domain, @@ -40,6 +41,7 @@ from paskia.db.operations import ( delete_role, delete_session, delete_sessions_for_user, + delete_sync_token, delete_user, is_username_taken, login, @@ -72,6 +74,7 @@ from paskia.db.structs import ( Role, Session, SessionContext, + SyncToken, User, ) @@ -96,6 +99,7 @@ __all__ = [ "Role", "Session", "SessionContext", + "SyncToken", "User", # Instance "data", @@ -111,6 +115,7 @@ __all__ = [ "create_domain", "create_reset_token", "create_role", + "create_sync_token", "create_user", "delete_credential", "delete_org", @@ -119,6 +124,7 @@ __all__ = [ "delete_role", "delete_session", "delete_sessions_for_user", + "delete_sync_token", "delete_user", "login", "oidc_login", diff --git a/paskia/db/operations.py b/paskia/db/operations.py index 58d66b7..df238ec 100644 --- a/paskia/db/operations.py +++ b/paskia/db/operations.py @@ -28,10 +28,12 @@ from paskia.db.structs import ( Role, Session, SessionContext, + SyncToken, User, ) from paskia.util.crypto import hash_secret from paskia.util.nameutil import slugify_name +from paskia.util.passphrase import generate as generate_passphrase _logger = logging.getLogger(__name__) @@ -563,6 +565,29 @@ def create_reset_token( return passphrase +def create_sync_token( + rp_id: str, *, ctx: SessionContext | None = None +) -> tuple[str, str]: + """Mint a satellite sync token scoped to an rp-id. + + Returns (key, token): the storage key (safe to echo for revocation) + and the raw token, which is shown once and never stored. + """ + token = generate_passphrase() + key = SyncToken.hash(token) + with _transaction("admin:create_sync_token", ctx): + _db.sync_tokens[key] = SyncToken(rp_id=rp_id, created=datetime.now(UTC)) + return key, token + + +def delete_sync_token(key: str, *, ctx: SessionContext | None = None) -> None: + """Revoke a satellite sync token by its storage key.""" + if key not in _db.sync_tokens: + raise ValueError("Sync token not found") + with _transaction("admin:delete_sync_token", ctx): + del _db.sync_tokens[key] + + # ------------------------------------------------------------------------- # Composite operations (used by app code) # ------------------------------------------------------------------------- diff --git a/paskia/db/structs.py b/paskia/db/structs.py index 0c340b2..f159244 100644 --- a/paskia/db/structs.py +++ b/paskia/db/structs.py @@ -428,15 +428,14 @@ class Session(msgspec.Struct, dict=True, omit_defaults=True): def store(self, last_seen: datetime) -> None: """Store this session in the database and record a visit. - Updates user.last_seen and user.visits. Must be called inside - a database transaction. + Updates user.last_seen and user.visits (instance-local stats, not + replicated). Must be called inside a database transaction. """ _data = db.data() _data.sessions[self.key] = self _data.users[self.user_uuid].last_seen = last_seen _data.users[self.user_uuid].visits += 1 syncfeed.emit("sessions", self.key, self) - syncfeed.emit("users", str(self.user_uuid), _data.users[self.user_uuid]) def delete(self) -> None: """Delete this session from the database. Must be called inside a transaction.""" @@ -566,6 +565,24 @@ class ResetToken(msgspec.Struct, dict=True): return token, passphrase +class SyncToken(msgspec.Struct, dict=True, omit_defaults=True): + """Satellite sync token: the bearer credential of a sync WebSocket. + + Stored keyed by ``hash_secret("sync", token)`` — the raw token is never + stored. Scoped to one rp-id: the feed carries only that domain's + sessions and credentials (each session with its identity context); the + user graph never replicates. Instance-local: never replicated, never + echoed back over the API. + """ + + rp_id: str + created: datetime + + @staticmethod + def hash(token: str) -> str: + return hash_secret("sync", token) + + # ------------------------------------------------------------------------- # OIDC Provider structures # ------------------------------------------------------------------------- @@ -647,8 +664,9 @@ class RemoteConfig(msgspec.Struct, omit_defaults=True): The satellite keeps a RAM-only read replica of the remote's tables and answers session-dependent reads locally; mutations are forwarded. The - token authenticates the sync channel (the remote reads accepted tokens - from its PASKIA_SYNC_TOKENS environment, never from its database). + token authenticates the sync channel: minted in the remote's admin UI, + stored there hashed (scoped to this domain's rp-id), and write-only in + the satellite's stored config view. """ url: str # e.g. "https://auth.example.com" @@ -710,6 +728,7 @@ class DB(msgspec.Struct, dict=True, omit_defaults=False): credentials: dict[UUID, Credential] = {} sessions: dict[str, Session] = {} reset_tokens: dict[str, ResetToken] = {} + sync_tokens: dict[str, SyncToken] = {} # OIDC provider data: one instance-global provider (single signing key # and client set); each request Host acts as an issuer alias. oidc: OIDC = msgspec.field(default_factory=OIDC) @@ -749,13 +768,19 @@ class DB(msgspec.Struct, dict=True, omit_defaults=False): Returns: SessionContext if valid, None if session not found, expired, or host mismatch """ - key = hash_secret("cookie", session_secret) try: s = self.sessions[key] except KeyError: return None + return self.session_ctx_for(s, host) + def session_ctx_for(self, s: Session, host: str | None) -> SessionContext | None: + """Full session context for a session object: host binding, then the + user/role/org/credential join with effective permissions. Sync + replicas override this to source the identity from the per-session + contexts the master pushes instead of local user tables. + """ # OIDC sessions (client_uuid set) are not valid for cookie-based auth if s.client_uuid is not None: return None diff --git a/paskia/domains.py b/paskia/domains.py index d284c13..8854f43 100644 --- a/paskia/domains.py +++ b/paskia/domains.py @@ -299,21 +299,6 @@ def validate_config( f"domain '{related_hosts[hn]}'" ) - # Several domains may share one remote instance (and its replica), but - # then its settings must be identical — a conflict would otherwise be - # resolved silently and arbitrarily. - remotes: dict[str, tuple[str, RemoteConfig]] = {} - for rp_id, domain in config.domains.items(): - if domain.remote is None: - continue - prior = remotes.setdefault(domain.remote.url, (rp_id, domain.remote)) - if prior[1] != domain.remote: - raise ValueError( - f"Domains '{prior[0]}' and '{rp_id}' share remote " - f"'{domain.remote.url}' with conflicting settings — a shared " - "remote must have one configuration" - ) - def sanitize_config( config: Config, *, related_origin_cap: int = DEFAULT_RELATED_ORIGIN_CAP @@ -332,7 +317,6 @@ def sanitize_config( warnings.append(msg) domains: dict[str, DomainConfig] = {} - remotes_seen: dict[str, RemoteConfig] = {} for rp_id, domain in config.domains.items(): try: hostutil.validate_rp_id(rp_id) @@ -400,14 +384,6 @@ def sanitize_config( remote = None else: remote = domain.remote - prior = remotes_seen.setdefault(remote.url, remote) - if prior != remote: - warn( - f"Domain '{rp_id}': remote '{remote.url}' settings " - "conflict with another domain's — aligned to the " - "first configuration" - ) - remote = prior if not auth_seen: warn( f"Domain '{rp_id}': remote domain without an auth host — " diff --git a/paskia/fastapi/admin/domains.py b/paskia/fastapi/admin/domains.py index 9b1fe32..1f907f4 100644 --- a/paskia/fastapi/admin/domains.py +++ b/paskia/fastapi/admin/domains.py @@ -7,13 +7,19 @@ origins on unrelated domains (WebAuthn Related Origin Requests). All changes are validated cross-domain before being persisted, and the runtime domain registry is rebuilt after each change so it takes effect immediately. + +A domain with an auth host may serve as a satellite upstream: sync tokens +are minted here (scoped to the rp-id), and the returned sync URL — +``https://@`` — is all a satellite's admin must paste. """ +from urllib.parse import unquote, urlsplit, urlunsplit + from fastapi import Body, FastAPI, Request from paskia import db, domains from paskia.db.structs import Config, DomainConfig, OriginEntry, RemoteConfig -from paskia.fastapi import authz +from paskia.fastapi import authz, sync from paskia.fastapi.admin.errors import install_error_handlers from paskia.fastapi.response import MsgspecResponse from paskia.fastapi.session import AUTH_COOKIE @@ -51,16 +57,25 @@ def _normalize_remote( ) -> RemoteConfig | None: """Parse a remote object from the admin UI (raises on malformed). - An absent/empty token keeps the previously stored one — the token is - write-only over the API. + The URL may carry the sync token as userinfo (the sync URL minted on + the upstream): ``https://@``. The token is split off + before storing. An absent token keeps the previously stored one — the + token is write-only over the API. """ if value is None: return None if not isinstance(value, dict) or not isinstance(value.get("url"), str): raise ValueError("remote must be an object with a url") - token = str(value.get("token") or "") or (existing.token if existing else "") + parts = urlsplit(value["url"].strip()) + userinfo_token = unquote(parts.password or parts.username or "") + netloc = parts.hostname or "" + if parts.port: + netloc = f"{netloc}:{parts.port}" + url = urlunsplit((parts.scheme, netloc, parts.path, "", "")).rstrip("/") + token = str(value.get("token") or "") or userinfo_token + token = token or (existing.token if existing else "") return RemoteConfig( - url=value["url"].rstrip("/"), + url=url, token=token, cache_ttl=max(1, int(value.get("cache_ttl") or 60)), refresh_interval=max(30, int(value.get("refresh_interval") or 300)), @@ -234,3 +249,71 @@ async def admin_delete_domain( db.delete_domain(rp_id, ctx=ctx) _rebuild_registry() return {"status": "ok"} + + +# ------------------------------------------------------------------------- +# Satellite sync tokens (this instance as the upstream) +# ------------------------------------------------------------------------- + + +def _sync_url(auth_host_url: str, token: str) -> str: + """The ready-to-paste satellite configuration URL: token in userinfo.""" + parts = urlsplit(auth_host_url) + return f"{parts.scheme}://{token}@{parts.netloc}" + + +@app.post("/{rp_id}/sync-token") +async def admin_create_sync_token(rp_id: str, request: Request, auth=AUTH_COOKIE): + """Mint a satellite sync token for this domain (sync URL shown once). + + Only an authoritative domain (not itself remote) with a marked auth + host can serve satellites — the auth host is where sign-in, profile + and admin pages live for satellite users. + """ + ctx = await authz.verify( + auth, ["auth:admin"], host=request.headers.get("host"), max_age="5m" + ) + domain = db.data().config.domains.get(rp_id) + if domain is None: + raise ValueError(f"Domain {rp_id} not found") + if domain.remote is not None: + raise ValueError( + f"Domain '{rp_id}' is itself served from a remote — " + "satellites cannot be nested" + ) + auth_host = domains.auth_host_url(domain) + if auth_host is None: + raise ValueError( + f"Domain '{rp_id}' must mark an auth host before it can serve satellites" + ) + key, token = db.create_sync_token(rp_id, ctx=ctx) + return {"id": key, "url": _sync_url(auth_host, token)} + + +@app.get("/{rp_id}/sync-tokens") +async def admin_list_sync_tokens(rp_id: str, request: Request, auth=AUTH_COOKIE): + """List this domain's sync tokens (storage ids, never the tokens).""" + await authz.verify(auth, ["auth:admin"], host=request.headers.get("host")) + return MsgspecResponse( + [ + {"id": key, "created": token.created} + for key, token in db.data().sync_tokens.items() + if token.rp_id == rp_id + ] + ) + + +@app.delete("/{rp_id}/sync-token/{key}") +async def admin_delete_sync_token( + rp_id: str, key: str, request: Request, auth=AUTH_COOKIE +): + """Revoke a satellite sync token; its live connections are dropped.""" + ctx = await authz.verify( + auth, ["auth:admin"], host=request.headers.get("host"), max_age="5m" + ) + stored = db.data().sync_tokens.get(key) + if stored is None or stored.rp_id != rp_id: + raise ValueError("Sync token not found") + db.delete_sync_token(key, ctx=ctx) + await sync.close_scope(rp_id) + return {"status": "ok"} diff --git a/paskia/fastapi/api.py b/paskia/fastapi/api.py index 85dc992..91987ce 100644 --- a/paskia/fastapi/api.py +++ b/paskia/fastapi/api.py @@ -163,6 +163,10 @@ async def check_user( No session cookie is read or written. Caller authentication is not required. """ + if (proxied := await satellite.forward_request(request)) is not None: + # User lookups without a session need the user graph, which only + # the remote holds. + return proxied host = hostutil.normalize_host(request.headers.get("host")) data = satellite.store_for_host(host) try: @@ -325,6 +329,10 @@ async def api_user_info( auth=AUTH_COOKIE, ): """Get full user profile including credentials and sessions.""" + if (proxied := await satellite.forward_request(request)) is not None: + # The replica holds per-session contexts only, not the user graph — + # profile data is answered by the remote. + return proxied if auth is None: raise authz.AuthException( status_code=401, diff --git a/paskia/fastapi/sync.py b/paskia/fastapi/sync.py index 0b017da..5f43574 100644 --- a/paskia/fastapi/sync.py +++ b/paskia/fastapi/sync.py @@ -1,12 +1,18 @@ """Sync WebSocket endpoint: serves snapshots and live events to satellites. -Token-gated via PASKIA_SYNC_TOKENS (env); closed when unset. All state is -RAM-only (syncfeed); the database schema is untouched. Protocol: snapshot -chunks per table, `ready`, then live upsert/delete events; the client sends -session_refresh write-backs. Reconnects always restart from a snapshot. +Token-gated by sync tokens minted in the admin UI (stored hashed in the +database, scoped to one rp-id). Beyond the token table all state is +RAM-only (syncfeed). A satellite receives only its own domain's sessions +and credentials; every session carries its ApiSessionContext (identity and +effective permissions for the session's host), so the user graph itself +never leaves this server. Protocol: snapshot chunks per table, `ready` +(carrying the token's rp-id scope), then live upsert/delete events; the +client sends session_refresh write-backs. Reconnects always restart from a +snapshot. """ import asyncio +import contextlib import logging from datetime import datetime @@ -14,23 +20,63 @@ import msgspec from fastapi import FastAPI, WebSocket, WebSocketDisconnect from paskia import db, syncfeed +from paskia.db.structs import SyncToken _logger = logging.getLogger(__name__) app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None) +# Live connections by subscriber queue, with their rp-id scope. +_connections: dict[asyncio.Queue, tuple[WebSocket, str]] = {} + + +async def close_scope(rp_id: str) -> None: + """Drop live sync connections for an rp-id (token revocation). + + Kicks every satellite of the domain; those whose token is still valid + reconnect immediately. + """ + for queue, (ws, scope) in list(_connections.items()): + if scope == rp_id: + with contextlib.suppress(Exception): + await ws.close(code=1008) + async def _send(ws: WebSocket, message: dict) -> None: await ws.send_bytes(syncfeed.encode(message)) -async def _apply_client_message(message: dict) -> None: - """Satellite write-behind: session refresh (validated/ip/user-agent).""" +def _token_scope(auth_header: str) -> str | None: + """The rp-id scope of a bearer sync token, or None when rejected.""" + token = auth_header.removeprefix("Bearer ").strip() + if not token: + return None + stored = db.data().sync_tokens.get(SyncToken.hash(token)) + return stored.rp_id if stored else None + + +def _snapshot_items(table: str, scope: str) -> list: + """Snapshot rows for a table, filtered to the token's rp-id scope; + sessions carry their ApiSessionContext as ``ctx``.""" + items = [] + for key, obj in getattr(db.data(), table).items(): + if obj.rp_id != scope: + continue + fields = msgspec.to_builtins(obj) + if table == "sessions": + fields["ctx"] = syncfeed.session_ctx_dict(obj) + items.append([str(key), fields]) + return items + + +async def _apply_client_message(message: dict, scope: str) -> None: + """Satellite write-behind: session refresh (validated/ip/user-agent), + confined to the token's rp-id scope.""" if message.get("type") != "session_refresh": return key = message.get("key") or "" session = db.data().sessions.get(key) - if session is None: + if session is None or session.rp_id != scope: return try: validated = msgspec.convert(message.get("validated"), datetime) @@ -46,35 +92,31 @@ async def _apply_client_message(message: dict) -> None: @app.websocket("/ws") async def sync_websocket(ws: WebSocket): - tokens = syncfeed.tokens_from_env() - auth = ws.headers.get("authorization", "") - if not tokens or auth.removeprefix("Bearer ").strip() not in tokens: + scope = _token_scope(ws.headers.get("authorization", "")) + if scope is None: await ws.close(code=1008) return await ws.accept() - queue = syncfeed.subscribe() + queue = syncfeed.subscribe(scope) + _connections[queue] = (ws, scope) try: - data = db.data() for table in syncfeed.TABLES: await _send( ws, { "type": "snapshot", "table": table, - "items": [ - [str(key), msgspec.to_builtins(obj)] - for key, obj in getattr(data, table).items() - ], + "items": _snapshot_items(table, scope), }, ) - await _send(ws, {"type": "ready"}) + await _send(ws, {"type": "ready", "rp_id": scope}) sender = asyncio.create_task(_pump(ws, queue)) try: while True: await _apply_client_message( - msgspec.json.decode(await ws.receive_bytes()) + msgspec.json.decode(await ws.receive_bytes()), scope ) finally: sender.cancel() @@ -83,6 +125,7 @@ async def sync_websocket(ws: WebSocket): except Exception: _logger.exception("Sync WebSocket failed") finally: + _connections.pop(queue, None) syncfeed.unsubscribe(queue) diff --git a/paskia/satellite.py b/paskia/satellite.py index 52a671e..2801fbd 100644 --- a/paskia/satellite.py +++ b/paskia/satellite.py @@ -5,13 +5,18 @@ paskia instance. This module owns the whole feature: it resolves which store serves a request host (local DB or the remote's read replica), dispatches session writes (refresh write-behind, logout eviction), and forwards requests the satellite cannot answer (exchange-code redemption, -OIDC, reset tokens) to the remote. +OIDC, reset tokens, session-less lookups like /check) to the remote. -A replica is a plain DB instance, never persisted, fed by a sync +A replica is a ReplicaDB instance, never persisted, fed by a sync WebSocket (snapshot on connect, then live events) and swept for -expired sessions locally. While the channel is down the replica stays trusted for the -domain's cache_ttl, then reads fail with RemoteUnavailable (fail-closed; -a large cache_ttl gives fail-open behavior bounded by session expiry). +expired sessions locally. The sync token is scoped to the domain's +rp-id: the replica holds only this domain's sessions and credentials, +each session carrying the ApiSessionContext the master computed for it +(identity display data and effective permissions for the session's host) +— the user graph itself never leaves the master. While the channel is +down the replica stays trusted for the domain's cache_ttl, then reads +fail closed (503; a large cache_ttl gives fail-open behavior bounded by +session expiry). """ import asyncio @@ -36,17 +41,15 @@ from paskia.db.structs import ( RemoteConfig, Role, Session, + SessionContext, User, ) +from paskia.util.apistructs import ApiSessionContext from paskia.util.crypto import hash_secret _logger = logging.getLogger(__name__) _TABLES = { - "permissions": (Permission, True), - "orgs": (Org, True), - "roles": (Role, True), - "users": (User, True), "credentials": (Credential, True), "sessions": (Session, False), } @@ -55,12 +58,58 @@ _RECONNECT_DELAY = 5 _SWEEP_INTERVAL = 60 -class RemoteReplica: - """One remote instance's replica, its sync client and write-behind queue.""" +class ReplicaDB(DB): + """A sync replica's store: the domain's sessions and credentials plus + the per-session contexts (identity and effective permissions) the + master pushes. Holds no user graph of its own; lookups join the + session with its pushed context.""" - def __init__(self, remote: RemoteConfig): + def __post_init__(self): + super().__post_init__() + self.session_contexts: dict[str, ApiSessionContext] = {} + + def session_ctx_for(self, s: Session, host: str | None) -> SessionContext | None: + if s.client_uuid is not None or s.host != host: + return None + stored = self.session_contexts.get(s.key) + if stored is None: + return None + try: + credential = self.credentials[s.credential_uuid] + except KeyError: + return None + user = User( + display_name=stored.user.display_name, + role_uuid=stored.role.uuid, + created_at=s.validated, + visits=0, + theme=stored.user.theme, + ) + user.uuid = stored.user.uuid + role = Role(org_uuid=stored.org.uuid, display_name=stored.role.display_name) + role.uuid = stored.role.uuid + org = Org(display_name=stored.org.display_name) + org.uuid = stored.org.uuid + return SessionContext( + session=s, + user=user, + org=org, + role=role, + credential=credential, + permissions=[ + Permission(scope=scope, display_name=scope) + for scope in stored.permissions + ], + ) + + +class RemoteReplica: + """One remote domain's replica, its sync client and write-behind queue.""" + + def __init__(self, remote: RemoteConfig, rp_id: str): self.remote = remote - self.db = DB() + self.rp_id = rp_id # the sync token must be scoped to this rp-id + self.db = ReplicaDB() self.last_contact = 0.0 # monotonic time the feed last went down self.connected = False self._pending_refresh: dict[str, dict] = {} @@ -116,6 +165,7 @@ class RemoteReplica: limit = datetime.now(UTC) - SESSION_LIFETIME for key in [k for k, s in self.db.sessions.items() if s.validated < limit]: del self.db.sessions[key] + self.db.session_contexts.pop(key, None) async def _run(self) -> None: while not self._stopped: @@ -143,7 +193,7 @@ class RemoteReplica: ping_timeout=5, ) as ws: sender = asyncio.create_task(self._send_loop(ws)) - staging: DB | None = None + staging: ReplicaDB | None = None ready_at = 0.0 try: while True: @@ -164,7 +214,7 @@ class RemoteReplica: message = msgspec.json.decode(await ws.recv()) mtype = message.get("type") if mtype == "snapshot": - staging = staging or DB() + staging = staging or ReplicaDB() for key, fields in message["items"]: _apply(staging, message["table"], key, fields) elif mtype == "event": @@ -177,6 +227,13 @@ class RemoteReplica: message.get("fields"), ) elif mtype == "ready": + if message.get("rp_id") != self.rp_id: + # The token is scoped to another domain — this + # replica would never see our data. + raise ValueError( + f"sync: token scope '{message.get('rp_id')}' " + f"does not match domain '{self.rp_id}'" + ) if staging is not None: self.db = staging staging = None @@ -196,24 +253,34 @@ class RemoteReplica: await self._refresh_signal.wait() -def _apply(replica: DB, table: str, key: str, fields: dict | None) -> None: - """Apply an upsert (fields given) or delete (fields None) to a replica.""" +def _apply(replica: ReplicaDB, table: str, key: str, fields: dict | None) -> None: + """Apply an upsert (fields given) or delete (fields None) to a replica. + Session upserts carry their master-computed context as ``ctx``.""" cls, uuid_key = _TABLES[table] store = getattr(replica, table) store_key = UUID(key) if uuid_key else key if fields is None: store.pop(store_key, None) + if table == "sessions": + replica.session_contexts.pop(key, None) return + fields = dict(fields) + ctx = fields.pop("ctx", None) if table == "sessions" else None obj = msgspec.convert(fields, cls) if uuid_key: obj.uuid = store_key else: obj.key = key store[store_key] = obj + if table == "sessions": + if ctx is not None: + replica.session_contexts[key] = msgspec.convert(ctx, ApiSessionContext) + else: + replica.session_contexts.pop(key, None) class SatelliteManager: - """Replicas keyed by remote URL; domains sharing a remote share one.""" + """Replicas keyed by rp-id; each remote domain syncs its own scope.""" def __init__(self): self.replicas: dict[str, RemoteReplica] = {} @@ -232,21 +299,19 @@ class SatelliteManager: async def reconcile(self, registry: domains.DomainRegistry) -> None: """Start/stop replicas to match the configured remote domains.""" - wanted = {} - for domain in registry.domains: - if domain.remote is not None: - wanted.setdefault(domain.remote.url, domain.remote) - for url in list(self.replicas): - if url not in wanted: - await self.replicas.pop(url).stop() - await _drop_client(url) - for url, remote in wanted.items(): - replica = self.replicas.get(url) + wanted = {d.rp_id: d.remote for d in registry.domains if d.remote is not None} + for rp_id in list(self.replicas): + if rp_id not in wanted: + replica = self.replicas.pop(rp_id) + await replica.stop() + await _drop_client(replica.remote.url) + for rp_id, remote in wanted.items(): + replica = self.replicas.get(rp_id) if replica is None or replica.remote != remote: if replica is not None: await replica.stop() - replica = RemoteReplica(remote) - self.replicas[url] = replica + replica = RemoteReplica(remote, rp_id) + self.replicas[rp_id] = replica await replica.start() @@ -263,7 +328,7 @@ def replica_for_host(host: str | None) -> RemoteReplica | None: domain = domains.registry().resolve(host) if domain is None or domain.remote is None: return None - return manager.replicas.get(domain.remote.url) + return manager.replicas.get(domain.rp_id) def store_for_host(host: str | None) -> DB: @@ -299,7 +364,9 @@ def evict_session(auth: str, host: str | None) -> None: """Drop a session from the replica (its remote deletion arrives via sync).""" replica = replica_for_host(host) if replica is not None: - replica.db.sessions.pop(hash_secret("cookie", auth), None) + key = hash_secret("cookie", auth) + replica.db.sessions.pop(key, None) + replica.db.session_contexts.pop(key, None) _TIMEOUT = httpx.Timeout(15.0, connect=5.0) diff --git a/paskia/syncfeed.py b/paskia/syncfeed.py index 8c3d3cd..2c80a8e 100644 --- a/paskia/syncfeed.py +++ b/paskia/syncfeed.py @@ -2,60 +2,127 @@ Nothing here touches the database file: committed mutations are pushed to connected satellites over the sync WebSocket (fastapi/sync.py). Satellites -authenticate with a token from the PASKIA_SYNC_TOKENS environment variable -(comma-separated); with the variable unset the sync endpoint stays closed. +authenticate with sync tokens minted in the admin UI (stored hashed in the +database, scoped to one rp-id); a scoped subscription receives only its own +rp-id's sessions and credentials — never the user graph. + +Data minimization is per session: every session upsert carries an embedded +``ctx`` (the ApiSessionContext also sent to users) holding the session's +user/org/role display data and effective permissions, computed for the +session's bound host. Mutations of the user graph (users, orgs, roles, +permissions) are not published; instead they trigger an eager re-push of +the affected sessions with freshly computed contexts, so revocations and +renames reach satellites within one RTT. There is deliberately no replay log: snapshots are small, so a reconnecting satellite simply takes a fresh one. """ import asyncio -import os +from uuid import UUID import msgspec -# Tables mirrored by satellites (reset tokens, OIDC data and domain config -# are instance-local and never replicated). -TABLES = ("permissions", "orgs", "roles", "users", "credentials", "sessions") +# Tables mirrored by satellites, all filtered to the subscriber's rp-id +# (reset tokens, sync tokens, OIDC data and domain config are +# instance-local and never replicated). +TABLES = ("credentials", "sessions") +SCOPED_TABLES = TABLES -_subscribers: set[asyncio.Queue] = set() +# The user graph is never published; upserts of these tables drive the +# eager re-push of affected session contexts instead. +_GRAPH_TABLES = ("permissions", "orgs", "roles", "users") + +_subscribers: dict[asyncio.Queue, str | None] = {} + + +def session_ctx_dict(session) -> dict | None: + """The session's ApiSessionContext as builtins (embedded in session + events and snapshots), computed for the session's bound host.""" + from paskia import db # noqa: PLC0415 (import cycle) + from paskia.util import userinfo # noqa: PLC0415 (import cycle) + + data = db.data() + if data is None: + return None + ctx = data.session_ctx_for(session, session.host) + if ctx is None: + return None + return msgspec.to_builtins(userinfo.build_session_context(ctx)) + + +def _repush_contexts(table: str, key: str) -> None: + """Re-emit the sessions whose context a user-graph upsert may have + changed: each carries a freshly computed ctx. Deletions are not needed + here — those cascade as session/credential delete events.""" + from paskia import db # noqa: PLC0415 (import cycle) + + data = db.data() + if data is None: + return + uuid = UUID(key) + if table == "users": + affected = {uuid} + elif table == "roles": + affected = {u.uuid for u in data.users.values() if u.role_uuid == uuid} + elif table == "orgs": + affected = { + u.uuid + for u in data.users.values() + if data.roles.get(u.role_uuid) is not None + and data.roles[u.role_uuid].org_uuid == uuid + } + else: # permissions: roles containing it, then their users + affected = { + u.uuid + for u in data.users.values() + if (role := data.roles.get(u.role_uuid)) is not None + and uuid in role.permissions + } + for skey, session in data.sessions.items(): + if session.user_uuid in affected: + emit("sessions", skey, session) def emit(table: str, key: str, obj) -> None: """Publish an upsert (obj given) or delete (obj None) to subscribers.""" if not _subscribers: return - event = { - "type": "event", - "table": table, - "key": key, - "fields": msgspec.to_builtins(obj) if obj is not None else None, - } - for queue in list(_subscribers): + if table in _GRAPH_TABLES: + # The user graph stays on the master; its changes reach satellites + # as re-pushed session contexts. + if obj is not None: + _repush_contexts(table, key) + return + fields = msgspec.to_builtins(obj) if obj is not None else None + if table == "sessions" and obj is not None: + fields["ctx"] = session_ctx_dict(obj) + event = {"type": "event", "table": table, "key": key, "fields": fields} + for queue, scope in list(_subscribers.items()): + if ( + scope is not None + and table in SCOPED_TABLES + and fields is not None + and fields.get("rp_id") != scope + ): + # Out of scope upsert; deletes pass (a replica ignores unknown + # keys, and the deleted object no longer tells its rp-id). + continue try: queue.put_nowait(event) except asyncio.QueueFull: # Slow consumer: drop it; the client reconnects and resyncs. - _subscribers.discard(queue) + _subscribers.pop(queue, None) -def subscribe() -> asyncio.Queue: +def subscribe(scope: str | None = None) -> asyncio.Queue: queue: asyncio.Queue = asyncio.Queue(maxsize=1000) - _subscribers.add(queue) + _subscribers[queue] = scope return queue def unsubscribe(queue: asyncio.Queue) -> None: - _subscribers.discard(queue) - - -def tokens_from_env() -> set[str]: - """Accepted sync tokens (PASKIA_SYNC_TOKENS, comma-separated).""" - return { - t.strip() - for t in os.environ.get("PASKIA_SYNC_TOKENS", "").split(",") - if t.strip() - } + _subscribers.pop(queue, None) def encode(message: dict) -> bytes: diff --git a/tests/test_remote.py b/tests/test_remote.py index 0d6bad5..bcd8b7e 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -15,27 +15,39 @@ from fastapi import Response import paskia.db.operations as ops_db from paskia import domains, satellite, syncfeed from paskia.db.structs import ( - DB, Config, Credential, DomainConfig, - Org, OriginEntry, - Permission, RemoteConfig, - Role, Session, User, ) from paskia.fastapi.mainapp import app from paskia.fastapi.session import AUTH_COOKIE_NAME +from paskia.util.apistructs import ( + ApiOrgContext, + ApiRoleContext, + ApiSessionContext, + ApiUserContext, +) from paskia.util.crypto import hash_secret -from .conftest import TEST_RP_ID +from .conftest import TEST_RP_ID, create_test_session REMOTE_URL = "http://remote.test" +def _api_ctx(display_name="Remote Admin", permissions=("auth:admin",)): + """A master-pushed per-session context, as embedded in session events.""" + return ApiSessionContext( + user=ApiUserContext(uuid=UUID(int=104), display_name=display_name), + org=ApiOrgContext(uuid=UUID(int=101), display_name="Org"), + role=ApiRoleContext(uuid=UUID(int=103), display_name="Admins"), + permissions=list(permissions), + ) + + def _remote_domain_config(**kw) -> Config: return Config( domains={ @@ -76,13 +88,20 @@ def test_sanitize_preserves_remote(): def test_apply_upsert_and_delete(): - replica = DB() - user = User.create(display_name="U", role=UUID(int=1)) - user.uuid = UUID(int=2) - satellite._apply(replica, "users", str(user.uuid), _builtins(user)) - assert replica.users[user.uuid].display_name == "U" - satellite._apply(replica, "users", str(user.uuid), None) - assert not replica.users + replica = satellite.ReplicaDB() + cred = Credential.create( + credential_id=b"cid", + user=UUID(int=1), + aaguid=UUID(int=0), + public_key=b"pk", + sign_count=0, + rp_id="example.com", + ) + cred.uuid = UUID(int=2) + satellite._apply(replica, "credentials", str(cred.uuid), _builtins(cred)) + assert replica.credentials[cred.uuid].public_key == b"pk" + satellite._apply(replica, "credentials", str(cred.uuid), None) + assert not replica.credentials def _builtins(obj): @@ -90,8 +109,9 @@ def _builtins(obj): def test_apply_session_roundtrip(): - """Sessions keep their string key and datetime/UUID fields.""" - replica = DB() + """Sessions keep their string key and datetime/UUID fields; the + embedded ctx lands in the replica's context map (and goes with it).""" + replica = satellite.ReplicaDB() session = Session.create( user=UUID(int=1), credential=UUID(int=2), @@ -102,16 +122,22 @@ def test_apply_session_roundtrip(): validated=datetime.now(UTC), rp_id="example.com", ) - satellite._apply(replica, "sessions", session.key, _builtins(session)) + fields = _builtins(session) + fields["ctx"] = _builtins(_api_ctx()) + satellite._apply(replica, "sessions", session.key, fields) stored = replica.sessions[session.key] assert stored.host == "app2.example.com" assert stored.validated == session.validated assert stored.user_uuid == UUID(int=1) + assert replica.session_contexts[session.key].user.display_name == "Remote Admin" + satellite._apply(replica, "sessions", session.key, None) + assert not replica.sessions + assert not replica.session_contexts def test_apply_credential_bytes_roundtrip(): """credential_id/public_key are bytes over the wire (base64 in JSON).""" - replica = DB() + replica = satellite.ReplicaDB() cred = Credential.create( credential_id=secrets.token_bytes(32), user=UUID(int=1), @@ -133,10 +159,17 @@ def test_apply_credential_bytes_roundtrip(): def test_feed_emit_to_subscribers(): queue = syncfeed.subscribe() try: - user = User.create(display_name="A", role=UUID(int=1)) - syncfeed.emit("users", "k1", user) - syncfeed.emit("users", "k1", None) - assert queue.get_nowait()["fields"]["display_name"] == "A" + cred = Credential.create( + credential_id=b"c", + user=UUID(int=1), + aaguid=UUID(int=0), + public_key=b"p", + sign_count=0, + rp_id="example.com", + ) + syncfeed.emit("credentials", "k1", cred) + syncfeed.emit("credentials", "k1", None) + assert queue.get_nowait()["fields"]["sign_count"] == 0 assert queue.get_nowait()["fields"] is None finally: syncfeed.unsubscribe(queue) @@ -146,32 +179,39 @@ def test_feed_drops_full_queue(): queue = syncfeed.subscribe() try: for i in range(1001): - syncfeed.emit("users", f"k{i}", None) + syncfeed.emit("credentials", f"k{i}", None) assert queue.qsize() == 1000 - syncfeed.emit("users", "k1001", None) # subscriber already dropped + syncfeed.emit("credentials", "k1001", None) # subscriber already dropped assert queue.qsize() == 1000 finally: syncfeed.unsubscribe(queue) @pytest.mark.asyncio -async def test_operations_emit_events(test_db): - """Writes through db.operations land on the sync feed.""" +async def test_operations_repush_session_contexts(test_db, test_user, test_credential): + """User-graph changes are never published; instead the affected + sessions are re-pushed with a freshly computed ctx.""" queue = syncfeed.subscribe() try: - user = next(iter(test_db.users.values())) - ops_db.update_user_display_name(user.uuid, "Renamed") - event = queue.get_nowait() - assert event["table"] == "users" - assert event["key"] == str(user.uuid) - assert event["fields"]["display_name"] == "Renamed" + key, _ = create_test_session( + user_uuid=test_user.uuid, credential_uuid=test_credential.uuid + ) + _drain(queue) # the session store itself emits (with ctx) + ops_db.update_user_display_name(test_user.uuid, "Renamed") + events = _drain(queue) + assert all(e["table"] != "users" for e in events) + repushed = [e for e in events if e["table"] == "sessions" and e["key"] == key] + assert repushed + assert repushed[-1]["fields"]["ctx"]["user"]["display_name"] == "Renamed" finally: syncfeed.unsubscribe(queue) @pytest.mark.asyncio async def test_replica_refresh_and_evict(): - replica = satellite.RemoteReplica(RemoteConfig(url=REMOTE_URL, token="t")) + replica = satellite.RemoteReplica( + RemoteConfig(url=REMOTE_URL, token="t"), "example.com" + ) token = secrets.token_urlsafe(12) session = Session.create( user=UUID(int=1), @@ -194,15 +234,15 @@ async def test_replica_refresh_and_evict(): # Host-keyed dispatch eviction (the replica's domain is resolved by host) domains.configure(listen=["localhost:4401"]) domains.init_registry(_remote_domain_config()) - satellite.manager.replicas[REMOTE_URL] = replica + satellite.manager.replicas["example.com"] = replica satellite.evict_session(token, "app2.example.com") assert not replica.db.sessions - satellite.manager.replicas.pop(REMOTE_URL) + satellite.manager.replicas.pop("example.com") def test_availability_gate(): replica = satellite.RemoteReplica( - RemoteConfig(url=REMOTE_URL, token="t", cache_ttl=60) + RemoteConfig(url=REMOTE_URL, token="t", cache_ttl=60), "example.com" ) assert not replica.available() # never synced replica.last_contact = time.monotonic() @@ -214,25 +254,12 @@ def test_availability_gate(): # ------------------------------------------------------------------------- -def _replica_db() -> tuple[DB, str]: - """A replica DB holding one org/role/perm/user/credential/session.""" - replica = DB() - org = Org.create(display_name="Org") - org.uuid = UUID(int=101) - replica.orgs[org.uuid] = org - perm = Permission.create(scope="auth:admin", display_name="Admin") - perm.uuid = UUID(int=102) - perm.orgs[org.uuid] = True - replica.permissions[perm.uuid] = perm - role = Role.create(org=org.uuid, display_name="Admins", permissions={perm.uuid}) - role.uuid = UUID(int=103) - replica.roles[role.uuid] = role - user = User.create(display_name="Remote Admin", role=role.uuid) - user.uuid = UUID(int=104) - replica.users[user.uuid] = user +def _replica_db() -> tuple[satellite.ReplicaDB, str]: + """A replica holding one credential/session and its pushed context.""" + replica = satellite.ReplicaDB() cred = Credential.create( credential_id=b"cid", - user=user.uuid, + user=UUID(int=104), aaguid=UUID(int=0), public_key=b"pk", sign_count=0, @@ -242,7 +269,7 @@ def _replica_db() -> tuple[DB, str]: replica.credentials[cred.uuid] = cred secret = secrets.token_urlsafe(12) session = Session.create( - user=user.uuid, + user=UUID(int=104), credential=cred.uuid, key=hash_secret("cookie", secret), host="app2.example.com", @@ -252,6 +279,7 @@ def _replica_db() -> tuple[DB, str]: rp_id="example.com", ) replica.sessions[session.key] = session + replica.session_contexts[session.key] = _api_ctx() return replica, secret @@ -262,17 +290,19 @@ async def remote_client(test_db): domains.configure(listen=["localhost:4401"]) domains.init_registry(config) replica_db, secret = _replica_db() - replica = satellite.RemoteReplica(RemoteConfig(url=REMOTE_URL, token="t")) + replica = satellite.RemoteReplica( + RemoteConfig(url=REMOTE_URL, token="t"), "example.com" + ) replica.db = replica_db replica.last_contact = time.monotonic() replica.connected = True - satellite.manager.replicas[REMOTE_URL] = replica + satellite.manager.replicas["example.com"] = replica transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient( transport=transport, base_url="http://localhost:4401" ) as client: yield client, secret, replica - satellite.manager.replicas.pop(REMOTE_URL, None) + satellite.manager.replicas.pop("example.com", None) @pytest.mark.asyncio @@ -339,6 +369,7 @@ async def test_logout_proxied_and_evicted(remote_client, monkeypatch): ) assert r.status_code == 200 assert not replica.db.sessions # evicted optimistically + assert not replica.db.session_contexts @pytest.mark.asyncio @@ -463,37 +494,92 @@ async def test_admin_remote_values_clamped(client, session_token, test_db): assert remote.refresh_interval == 30 -def _add_shared_remote(config: Config, token: str) -> None: - config.domains["sub.example.com"] = DomainConfig( - origins={ - "**.sub.example.com": True, - "auth.sub.example.com": OriginEntry(auth_host=True), +@pytest.mark.asyncio +async def test_admin_remote_url_userinfo_token(client, session_token, test_db): + """The sync URL minted on an upstream carries the token as userinfo; + it is split off before storing (and never echoed back).""" + r = await client.post( + "/auth/api/admin/domains/", + json={ + "rp_id": "example.com", + "origins": { + "**.example.com": True, + "auth.example.com": {"auth_host": True}, + }, + "remote": {"url": "https://alpha.bravo.charlie.delta.echo@remote.test"}, }, - remote=RemoteConfig(url=REMOTE_URL, token=token), + headers={ + "Host": "localhost:4401", + "Cookie": f"{AUTH_COOKIE_NAME}={session_token}", + }, + ) + assert r.status_code == 200, r.text + remote = test_db.config.domains["example.com"].remote + assert remote.url == "https://remote.test" + assert remote.token == "alpha.bravo.charlie.delta.echo" + + # Saving without userinfo keeps the stored token + r = await client.patch( + "/auth/api/admin/domains/example.com", + json={ + "rp_name": "Ex", + "origins": { + "**.example.com": True, + "auth.example.com": {"auth_host": True}, + }, + "remote": {"url": "https://remote.test"}, + }, + headers={ + "Host": "localhost:4401", + "Cookie": f"{AUTH_COOKIE_NAME}={session_token}", + }, + ) + assert r.status_code == 200, r.text + assert ( + test_db.config.domains["example.com"].remote.token + == "alpha.bravo.charlie.delta.echo" ) -def test_shared_remote_same_settings_ok(): - config = _remote_domain_config() - _add_shared_remote(config, token="t") - domains.validate_config(config) +def test_feed_scoped_subscription_filters(): + """A scoped subscriber gets only its rp-id's sessions/credentials + (upserts filtered, deletes passed); the user graph is never published.""" + scoped = syncfeed.subscribe("example.com") + try: + in_scope = Session.create( + user=UUID(int=1), + credential=UUID(int=2), + key="k1", + host="app2.example.com", + ip="1.1.1.1", + user_agent="ua", + validated=datetime.now(UTC), + rp_id="example.com", + ) + out_scope = Session.create( + user=UUID(int=1), + credential=UUID(int=3), + key="k2", + host="localhost", + ip="1.1.1.1", + user_agent="ua", + validated=datetime.now(UTC), + rp_id="localhost", + ) + syncfeed.emit("sessions", "k1", in_scope) + syncfeed.emit("sessions", "k2", out_scope) + syncfeed.emit("users", "u1", User.create(display_name="A", role=UUID(int=1))) + syncfeed.emit("sessions", "k2", None) # deletes pass regardless + assert [e["key"] for e in _drain(scoped)] == ["k1", "k2"] + finally: + syncfeed.unsubscribe(scoped) -def test_shared_remote_conflicting_settings_rejected(): - config = _remote_domain_config() - _add_shared_remote(config, token="different") - with pytest.raises(ValueError, match="conflicting"): - domains.validate_config(config) - - -def test_sanitize_aligns_conflicting_shared_remote(): - config = _remote_domain_config() - _add_shared_remote(config, token="different") - sanitized, warnings = domains.sanitize_config(config) - assert any("conflict" in w for w in warnings) - # Aligned to the first configuration, and the result validates - assert sanitized.domains["sub.example.com"].remote.token == "t" - domains.validate_config(sanitized) +def _drain(queue): + items = [] + while not queue.empty(): + items.append(queue.get_nowait()) + return items def test_rebuild_listener_skipped_without_running_loop(): @@ -569,3 +655,101 @@ async def test_user_api_proxied_on_remote_auth_host(remote_client, monkeypatch): ) assert r.status_code == 200 assert r.content == b"proxied" + + +# ------------------------------------------------------------------------- +# Sync tokens (this instance as the upstream) +# ------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_sync_token_lifecycle(client, session_token, test_db): + """Mint (auth host required), list, revoke — the raw token appears once.""" + headers = { + "Host": "localhost:4401", + "Cookie": f"{AUTH_COOKIE_NAME}={session_token}", + } + + r = await client.post( + "/auth/api/admin/domains/", + json={ + "rp_id": "example.com", + "rp_name": "Example", + "origins": {"**.example.com": True}, + }, + headers=headers, + ) + assert r.status_code == 200, r.text + + # No auth host yet — refused + r = await client.post( + "/auth/api/admin/domains/example.com/sync-token", headers=headers + ) + assert r.status_code == 400 + assert "auth host" in r.json()["detail"] + + # Mark an auth host, then mint + r = await client.patch( + "/auth/api/admin/domains/example.com", + json={ + "rp_name": "Example", + "origins": { + "**.example.com": True, + "auth.example.com": {"auth_host": True}, + }, + }, + headers=headers, + ) + assert r.status_code == 200, r.text + r = await client.post( + "/auth/api/admin/domains/example.com/sync-token", headers=headers + ) + assert r.status_code == 200, r.text + body = r.json() + assert body["url"].startswith("https://") + assert "@auth.example.com" in body["url"] + raw_token = body["url"].split("://", 1)[1].split("@", 1)[0] + assert raw_token.count(".") == 4 # five-word passphrase + + r = await client.get( + "/auth/api/admin/domains/example.com/sync-tokens", headers=headers + ) + assert [t["id"] for t in r.json()] == [body["id"]] + assert "alpha" not in r.text and raw_token not in r.text # hashes only + + r = await client.delete( + f"/auth/api/admin/domains/example.com/sync-token/{body['id']}", headers=headers + ) + assert r.status_code == 200 + assert not test_db.sync_tokens + r = await client.delete( + f"/auth/api/admin/domains/example.com/sync-token/{body['id']}", headers=headers + ) + assert r.status_code == 400 + + +@pytest.mark.asyncio +async def test_sync_token_refused_for_remote_domain(client, session_token): + """A satellite-served domain cannot be an upstream: no nesting.""" + headers = { + "Host": "localhost:4401", + "Cookie": f"{AUTH_COOKIE_NAME}={session_token}", + } + r = await client.post( + "/auth/api/admin/domains/", + json={ + "rp_id": "example.com", + "origins": { + "**.example.com": True, + "auth.example.com": {"auth_host": True}, + }, + "remote": {"url": "https://alpha.bravo.charlie.delta.echo@remote.test"}, + }, + headers=headers, + ) + assert r.status_code == 200, r.text + r = await client.post( + "/auth/api/admin/domains/example.com/sync-token", headers=headers + ) + assert r.status_code == 400 + assert "nested" in r.json()["detail"] diff --git a/tests/test_satellite_e2e.py b/tests/test_satellite_e2e.py index 43b3d71..c05a69f 100644 --- a/tests/test_satellite_e2e.py +++ b/tests/test_satellite_e2e.py @@ -14,10 +14,12 @@ are covered in test_remote.py with a mocked forward_request. """ import asyncio +import os import socket import time from datetime import UTC, datetime, timedelta from types import SimpleNamespace +from uuid import UUID import httpx import pytest @@ -26,7 +28,13 @@ import uvicorn import paskia.db.operations as ops_db from paskia import domains, satellite -from paskia.db.structs import Config, DomainConfig, OriginEntry, RemoteConfig +from paskia.db.structs import ( + Config, + Credential, + DomainConfig, + OriginEntry, + RemoteConfig, +) from paskia.fastapi.mainapp import app from paskia.fastapi.session import AUTH_COOKIE_NAME @@ -52,7 +60,7 @@ async def wait_for(desc: str, condition, timeout: float = 10) -> None: @pytest_asyncio.fixture -async def live_remote(test_db, monkeypatch): +async def live_remote(test_db): """A uvicorn-served 'remote' and the satellite manager wired against it. The manager is started for real: reconcile builds the replica from the @@ -60,7 +68,7 @@ async def live_remote(test_db, monkeypatch): client (ASGI, satellite side), replica, db (the remote's database), url/port of the remote, start_server (for restarts), servers. """ - monkeypatch.setenv("PASKIA_SYNC_TOKENS", "live-tok") + _, live_token = ops_db.create_sync_token(REMOTE_DOMAIN) servers: list[tuple[uvicorn.Server, asyncio.Task]] = [] async def start_server(port: int) -> uvicorn.Server: @@ -88,13 +96,13 @@ async def live_remote(test_db, monkeypatch): f"**.{REMOTE_DOMAIN}": True, f"auth.{REMOTE_DOMAIN}": OriginEntry(auth_host=True), }, - remote=RemoteConfig(url=url, token="live-tok"), + remote=RemoteConfig(url=url, token=live_token), ), } ) ) await satellite.manager.start() - replica = satellite.manager.replicas[url] + replica = satellite.manager.replicas[REMOTE_DOMAIN] await wait_for("initial snapshot", lambda: replica.available()) transport = httpx.ASGITransport(app=app) @@ -108,6 +116,7 @@ async def live_remote(test_db, monkeypatch): db=test_db, url=url, port=port, + token=live_token, start_server=start_server, servers=servers, ) @@ -122,27 +131,88 @@ def _app_headers(secret: str) -> dict[str, str]: return {"Host": APP_HOST, "Cookie": f"{AUTH_COOKIE_NAME}={secret}"} +@pytest_asyncio.fixture +async def remote_credential(test_db, test_user): + """A credential registered on the remote domain. The feed is rp-id + scoped: only the domain's own credentials replicate to its satellites.""" + credential = Credential.create( + credential_id=os.urandom(32), + user=test_user.uuid, + aaguid=UUID("00000000-0000-0000-0000-000000000000"), + public_key=os.urandom(64), + sign_count=0, + rp_id=REMOTE_DOMAIN, + ) + ops_db.create_credential(credential) + return credential + + @pytest.mark.asyncio -async def test_live_snapshot_and_user_events(live_remote, test_user): - """Bootstrap data arrives with the snapshot; later mutations as events.""" - assert test_user.uuid in live_remote.replica.db.users +async def test_live_snapshot_and_context_repush( + live_remote, test_user, remote_credential +): + """Sessions arrive with their master-computed context; user-graph + changes eagerly re-push the affected sessions' contexts.""" + live = live_remote + key, _ = create_test_session( + user_uuid=test_user.uuid, + credential_uuid=remote_credential.uuid, + host=APP_HOST, + rp_id=REMOTE_DOMAIN, + ) + await wait_for("session sync", lambda: key in live.replica.db.session_contexts) + assert live.replica.db.session_contexts[key].user.display_name == "Test Admin" + ops_db.update_user_display_name(test_user.uuid, "Live Rename") await wait_for( "rename to propagate", lambda: ( - live_remote.replica.db.users[test_user.uuid].display_name == "Live Rename" + live.replica.db.session_contexts[key].user.display_name == "Live Rename" ), ) @pytest.mark.asyncio -async def test_live_forward_and_writebehind(live_remote, test_user, test_credential): +async def test_live_permission_revocation_propagates( + live_remote, test_user, remote_credential +): + """Removing a permission from the user's role re-pushes the session + context: the satellite denies within one RTT.""" + live = live_remote + key, secret = create_test_session( + user_uuid=test_user.uuid, + credential_uuid=remote_credential.uuid, + host=APP_HOST, + rp_id=REMOTE_DOMAIN, + ) + await wait_for("session sync", lambda: key in live.replica.db.session_contexts) + r = await live.client.get( + "/auth/api/forward?perm=auth:admin", headers=_app_headers(secret) + ) + assert r.status_code == 204 + + perm_uuid = next( + p.uuid for p in live.db.permissions.values() if p.scope == "auth:admin" + ) + ops_db.remove_permission_from_role(test_user.role_uuid, perm_uuid) + await wait_for( + "revocation to propagate", + lambda: "auth:admin" not in live.replica.db.session_contexts[key].permissions, + ) + r = await live.client.get( + "/auth/api/forward?perm=auth:admin", headers=_app_headers(secret) + ) + assert r.status_code == 403 + + +@pytest.mark.asyncio +async def test_live_forward_and_writebehind(live_remote, test_user, remote_credential): """A session created on the remote authorizes locally; the /validate refresh travels back over the wire into the remote's database.""" live = live_remote key, secret = create_test_session( user_uuid=test_user.uuid, - credential_uuid=test_credential.uuid, + credential_uuid=remote_credential.uuid, host=APP_HOST, rp_id=REMOTE_DOMAIN, ) @@ -171,11 +241,13 @@ async def test_live_forward_and_writebehind(live_remote, test_user, test_credent @pytest.mark.asyncio -async def test_live_session_delete_propagates(live_remote, test_user, test_credential): +async def test_live_session_delete_propagates( + live_remote, test_user, remote_credential +): live = live_remote key, secret = create_test_session( user_uuid=test_user.uuid, - credential_uuid=test_credential.uuid, + credential_uuid=remote_credential.uuid, host=APP_HOST, rp_id=REMOTE_DOMAIN, ) @@ -191,7 +263,9 @@ async def test_live_session_delete_propagates(live_remote, test_user, test_crede @pytest.mark.asyncio async def test_live_wrong_token_rejected(live_remote): - bad = satellite.RemoteReplica(RemoteConfig(url=live_remote.url, token="wrong")) + bad = satellite.RemoteReplica( + RemoteConfig(url=live_remote.url, token="wrong"), REMOTE_DOMAIN + ) await bad.start() try: await asyncio.sleep(1.5) @@ -202,14 +276,14 @@ async def test_live_wrong_token_rejected(live_remote): @pytest.mark.asyncio async def test_live_disconnect_ttl_and_reconnect( - live_remote, test_user, test_credential + live_remote, test_user, remote_credential ): """Channel down: replica keeps serving within cache_ttl, then fails closed (503); reconnect starts from a snapshot and reconciles drift.""" live = live_remote key, secret = create_test_session( user_uuid=test_user.uuid, - credential_uuid=test_credential.uuid, + credential_uuid=remote_credential.uuid, host=APP_HOST, rp_id=REMOTE_DOMAIN, ) @@ -234,7 +308,7 @@ async def test_live_disconnect_ttl_and_reconnect( await live.start_server(live.port) await wait_for("reconnect", lambda: live.replica.available(), timeout=20) assert ( - live.replica.db.users[test_user.uuid].display_name == "While Down" + live.replica.db.session_contexts[key].user.display_name == "While Down" ) # snapshot reconciled the drift r = await live.client.get("/auth/api/forward", headers=_app_headers(secret)) assert r.status_code == 204 @@ -243,10 +317,84 @@ async def test_live_disconnect_ttl_and_reconnect( @pytest.mark.asyncio async def test_reconcile_stops_replica_when_remote_removed(live_remote): """A registry rebuild without the remote config stops the replica.""" - live = live_remote domains.init_registry( Config(domains={TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True})}) ) await wait_for( - "replica shutdown", lambda: live.url not in satellite.manager.replicas + "replica shutdown", lambda: REMOTE_DOMAIN not in satellite.manager.replicas ) + + +@pytest.mark.asyncio +async def test_live_out_of_scope_session_not_replicated( + live_remote, test_user, test_credential, remote_credential +): + """The feed is scoped to the token's rp-id: sessions of other domains + (here the satellite's own) never cross the wire.""" + live = live_remote + local_key, _ = create_test_session( + user_uuid=test_user.uuid, + credential_uuid=test_credential.uuid, + host="localhost:4401", + rp_id=TEST_RP_ID, + ) + # Positive control: an in-scope session created afterwards does arrive, + # so the out-of-scope one has already passed through the pipe. + remote_key, _ = create_test_session( + user_uuid=test_user.uuid, + credential_uuid=remote_credential.uuid, + host=APP_HOST, + rp_id=REMOTE_DOMAIN, + ) + await wait_for( + "in-scope session sync", lambda: remote_key in live.replica.db.sessions + ) + assert local_key not in live.replica.db.sessions + + +@pytest.mark.asyncio +async def test_live_out_of_scope_writebehind_ignored( + live_remote, test_user, test_credential, remote_credential +): + """A session_refresh for a session outside the token's scope is ignored + by the remote (fail-closed), while an in-scope one lands.""" + live = live_remote + stale = datetime.now(UTC) - timedelta(minutes=10) + local_key, _ = create_test_session( + user_uuid=test_user.uuid, + credential_uuid=test_credential.uuid, + host="localhost:4401", + rp_id=TEST_RP_ID, + ) + remote_key, _ = create_test_session( + user_uuid=test_user.uuid, + credential_uuid=remote_credential.uuid, + host=APP_HOST, + rp_id=REMOTE_DOMAIN, + ) + ops_db.update_session(local_key, validated=stale) + ops_db.update_session(remote_key, validated=stale) + + now = datetime.now(UTC) + live.replica.refresh_session(local_key, now, None, None) + live.replica.refresh_session(remote_key, now, None, None) + await wait_for( + "in-scope refresh to land", + lambda: live.db.sessions[remote_key].validated > stale, + ) + assert live.db.sessions[local_key].validated == stale + + +@pytest.mark.asyncio +async def test_live_scope_mismatch_rejected(live_remote): + """A valid token scoped to another domain is useless: the ready message + names its scope and the replica refuses to go live on the mismatch.""" + bad = satellite.RemoteReplica( + RemoteConfig(url=live_remote.url, token=live_remote.token), "other.org" + ) + await bad.start() + try: + await asyncio.sleep(1.5) + assert not bad.available() + finally: + await bad.stop()