Satellite v2: upstream-minted scoped sync tokens, per-session contexts

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://<token>@<auth-host>. 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 (**.<rp-id> + upstream auth host) and auth host. New
domains adopt the upstream's rp-id and name.
This commit is contained in:
LeoVasanko committed 2026-10-01 05:40:32 +00:00
1 parent da0b3e0e8b
commit 2209e2395c
14 files changed
+1159 -330

No files matched your search

+67 -39
View File
@@ -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://<token>@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
`**.<rp-id>` 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
+243 -80
View File
@@ -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) {
<input v-model="dialog.data.rp_name" :placeholder="dialog.data.rp_id" />
</label>
<div class="origin-label">
Allowed Origins
<button type="button" class="icon-btn origin-add-btn" @click="addOrigin()" aria-label="Add origin" title="Add origin">➕</button>
</div>
<div v-if="dialog.data.origins.length" class="origin-list">
<div v-for="(_, i) in dialog.data.origins" :key="i" class="origin-row">
<input
ref="originInputs"
:value="dialog.data.origins[i]"
@input="e => onOriginInput(i, e)"
@blur="onOriginBlur(i)"
class="origin-input"
:class="{ 'input-error': dialog.data.originValidation[i] === 'invalid' }"
/>
<span v-if="isAuthHostEntry(dialog.data.origins[i])" class="key-badge" title="Authentication site — the account and admin interface live here">🔑</span>
<span v-else-if="isRelatedEntry(dialog.data.origins[i])" class="key-badge" title="Related origin (WebAuthn ROR) — shares this domain's passkeys">🔗</span>
<div class="row-menu">
<button type="button" class="icon-btn" @click.stop="openMenu = openMenu === i ? null : i" aria-label="Origin actions" title="Actions">⋮</button>
<div v-if="openMenu === i" class="row-menu-popup">
<button v-if="isAuthHostEntry(dialog.data.origins[i])" type="button" @click="clearAuthHost()"><span class="menu-icon">🔑</span>Remove auth host</button>
<button v-else-if="!isRelatedEntry(dialog.data.origins[i]) && originHostname(dialog.data.origins[i])" type="button" @click="setAuthHost(i)"><span class="menu-icon">🔑</span>Set as auth host</button>
<button type="button" @click="onRemoveOrigin(i)"><span class="menu-icon delete-menu-icon">❌</span>Delete</button>
<template v-if="!dialog.data.remote">
<div class="origin-label">
Allowed Origins
<button type="button" class="icon-btn origin-add-btn" @click="addOrigin()" aria-label="Add origin" title="Add origin">➕</button>
</div>
<div v-if="dialog.data.origins.length" class="origin-list">
<div v-for="(_, i) in dialog.data.origins" :key="i" class="origin-row">
<input
ref="originInputs"
:value="dialog.data.origins[i]"
@input="e => onOriginInput(i, e)"
@blur="onOriginBlur(i)"
class="origin-input"
:class="{ 'input-error': dialog.data.originValidation[i] === 'invalid' }"
/>
<span v-if="isAuthHostEntry(dialog.data.origins[i])" class="key-badge" title="Authentication site — the account and admin interface live here">🔑</span>
<span v-else-if="isRelatedEntry(dialog.data.origins[i])" class="key-badge" title="Related origin (WebAuthn ROR) — shares this domain's passkeys">🔗</span>
<div class="row-menu">
<button type="button" class="icon-btn" @click.stop="openMenu = openMenu === i ? null : i" aria-label="Origin actions" title="Actions">⋮</button>
<div v-if="openMenu === i" class="row-menu-popup">
<button v-if="isAuthHostEntry(dialog.data.origins[i])" type="button" @click="clearAuthHost()"><span class="menu-icon">🔑</span>Remove auth host</button>
<button v-else-if="!isRelatedEntry(dialog.data.origins[i]) && originHostname(dialog.data.origins[i])" type="button" @click="setAuthHost(i)"><span class="menu-icon">🔑</span>Set as auth host</button>
<button type="button" @click="onRemoveOrigin(i)"><span class="menu-icon delete-menu-icon">❌</span>Delete</button>
</div>
</div>
</div>
</div>
</div>
<p class="small muted">
Only the listed sites may sign in with {{ dialog.data.rp_id }} passkeys. Wildcards may be used: <strong>**.{{ dialog.data.rp_id }}</strong> allows the whole domain, <strong>*.{{ dialog.data.rp_id }}</strong> only a single subdomain level.<template v-if="relatedEntries.length"> 🔗 means related host requiring WebAuthn ROR setup.</template><template v-if="dialog.data.auth_host"> 🔑 is the dedicated Paskia host for all account management.</template>
</p>
<p class="small muted">
Only the listed sites may sign in with {{ dialog.data.rp_id }} passkeys. Wildcards may be used: <strong>**.{{ dialog.data.rp_id }}</strong> allows the whole domain, <strong>*.{{ dialog.data.rp_id }}</strong> only a single subdomain level.<template v-if="relatedEntries.length"> 🔗 means related host requiring WebAuthn ROR setup.</template><template v-if="dialog.data.auth_host"> 🔑 is the dedicated Paskia host for all account management.</template>
</p>
</template>
<div class="origin-label">
<label class="remote-toggle">
@@ -591,20 +717,27 @@ function onRemoveOrigin(i) {
</label>
</div>
<template v-if="dialog.data.remote">
<label>Remote URL
<input v-model="dialog.data.remote.url" placeholder="https://auth.example.com" data-form-type="other" :class="{ 'input-error': remoteUrlInvalid }" />
<label>Sync URL
<input
v-model="syncUrl"
placeholder="https://alpha.bravo.charlie.delta.echo@auth.example.com"
autocomplete="off"
data-lpignore="true"
data-1p-ignore="true"
data-form-type="other"
spellcheck="false"
:class="{ 'input-error': syncUrlInvalid || syncUrlMissingToken }"
/>
</label>
<p v-if="remoteUrlInvalid" class="small error">Must be an http(s) URL.</p>
<p class="small muted">Minted on the authoritative instance — open the domain there and use <em>Satellite access → Generate sync URL</em>.<template v-if="hadRemoteOnOpen"> Paste a new one to replace the stored token.</template></p>
<p v-if="syncUrlInvalid" class="small error">Must be an http(s) URL.</p>
<p v-else-if="syncUrlMissingToken" class="small error">The URL carries no token — paste the complete sync URL from the upstream.</p>
<p v-else-if="uplinkCheck === 'validating'" class="small muted">Checking the uplink…</p>
<p v-else-if="uplinkCheck === 'valid'" class="small">✓ Uplink verified — the remote serves this domain authoritatively.</p>
<p v-else-if="uplinkCheck === 'unreachable'" class="small">The remote is unreachable — the uplink cannot be verified. Save anyway only if it is not deployed yet.</p>
<p v-else-if="uplinkCheck === 'mismatch'" class="small error">The remote does not serve {{ dialogRpId }} — check the URL. Connect directly to the authoritative instance.</p>
<p v-else-if="uplinkCheck === 'nested'" class="small error">The remote is itself a satellite for this domain — nested satellites are not supported. Point directly at the authoritative instance.</p>
<p v-if="remoteMissingAuthHost" class="small error">A remote domain must mark an auth host above — profile, admin and sign-in pages live there (typically the remote's own site).</p>
<label>Sync token
<input v-model="dialog.data.remote.token" type="password" placeholder="Token in the remote's PASKIA_SYNC_TOKENS" autocomplete="off" data-form-type="other" />
</label>
<p class="small muted">Accepted by the remote via its PASKIA_SYNC_TOKENS environment variable.<template v-if="!dialog.data.isNew"> Leave empty to keep the stored token.</template></p>
<p v-else-if="uplinkCheck === 'valid'" class="small">✓ Uplink verified — {{ parsedSyncUrl?.host }} serves this domain authoritatively. Origins and the auth host are taken from it.</p>
<p v-else-if="uplinkCheck === 'unreachable'" class="small error">The upstream is unreachable — check the URL.</p>
<p v-else-if="uplinkCheck === 'mismatch'" class="small error">The upstream does not serve {{ dialogRpId }} — its sync URL must be minted for this domain.</p>
<p v-else-if="uplinkCheck === 'nested'" class="small error">The upstream is itself a satellite for this domain — nested satellites are not supported. Point directly at the authoritative instance.</p>
<p v-else-if="uplinkCheck === 'no-auth-host'" class="small error">The upstream has no auth host for this domain — only an instance with an auth host can serve satellites.</p>
<label>Staleness limit (cache TTL, seconds)
<input v-model.number="dialog.data.remote.cache_ttl" type="number" min="1" />
</label>
@@ -612,9 +745,33 @@ function onRemoveOrigin(i) {
<input v-model.number="dialog.data.remote.refresh_interval" type="number" min="30" />
</label>
<p class="small muted">
Session checks run locally against a RAM replica of the remote (sub-millisecond). If the connection is down longer than the staleness limit, checks fail closed (503).
Session checks run locally against a RAM replica of the upstream (sub-millisecond). If the connection is down longer than the staleness limit, checks fail closed (503).
</p>
</template>
<template v-if="showSatelliteAccess">
<div class="origin-label">Satellite access</div>
<template v-if="dialog.data.auth_host">
<p class="small muted">
Other paskia instances can serve this domain as satellites — local RAM replicas kept in sync over a feed scoped to {{ dialog.data.rp_id }}. Generate one sync URL per satellite and paste it into that instance's configuration for the domain.
</p>
<div v-if="syncTokens?.length" class="sync-token-list">
<div v-for="t in syncTokens" :key="t.id" class="sync-token-row">
<span class="small sync-token-id" title="Sync token id">…{{ t.id.slice(-8) }}</span>
<span class="small muted">{{ formatDate(t.created) }}</span>
<button type="button" class="icon-btn" :disabled="satelliteBusy" @click="revokeSyncToken(t.id)" aria-label="Revoke sync token" title="Revoke — drops the satellite's connection">❌</button>
</div>
</div>
<div>
<button type="button" class="btn-secondary" :disabled="satelliteBusy" @click="generateSyncToken">Generate sync URL</button>
</div>
<template v-if="mintedSyncUrl">
<pre class="wellknown-doc" title="Click to copy" tabindex="0" @click="copyText(mintedSyncUrl, 'Sync URL')" @keydown.enter.prevent="copyText(mintedSyncUrl, 'Sync URL')">{{ mintedSyncUrl }}</pre>
<p class="small muted">Shown once — the token is stored hashed and cannot be recovered later.</p>
</template>
</template>
<p v-else class="small muted">Mark an auth host above and save to let satellites connect — sign-in, profile and admin pages for satellite users live there.</p>
</template>
</AdminDialog>
</template>
@@ -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); }
</style>
+6
View File
@@ -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;
+6
View File
@@ -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",
+25
View File
@@ -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)
# -------------------------------------------------------------------------
+31 -6
View File
@@ -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
-24
View File
@@ -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 — "
+88 -5
View File
@@ -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://<token>@<auth-host>`` — 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://<token>@<auth-host>``. 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"}
+8
View File
@@ -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,
+61 -18
View File
@@ -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)
+99 -32
View File
@@ -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)
+94 -27
View File
@@ -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:
+264 -80
View File
@@ -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"]
+167 -19
View File
@@ -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()