LeoVasanko/paskia
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Tests: align with instance-global OIDC and per-domain auth hosts

Browse Source 
- '*' origin is wildcard shorthand (https-only outside localhost)
- related origins may fall inside another domain's rp-id; the listing
  wins dispatch, an exact rp-id always wins
- shared auth hosts resolve best-suffix; no cross-domain fallback
- OIDC codes are host-independent; log censoring path is oidc.key
- legacy wildcards convert as-is; legacy OIDC carries over as-is
This commit is contained in:
LeoVasanko
2026-09-07 06:33:54 +00:00
parent af6d7e3a3f
commit 2039c47e46
9 changed files with 182 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
paskia/db/operations.py
+2 -6
View File
@@ -771,9 +771,7 @@ def delete_domain(rp_id: str, *, ctx: SessionContext | None = None) -> None:
# -------------------------------------------------------------------------
def create_oid_client(
client: Client, *, ctx: SessionContext | None = None
) -> None:
def create_oid_client(client: Client, *, ctx: SessionContext | None = None) -> None:
"""Create a new OIDC client."""
if client.uuid in _db.oidc.clients:
raise ValueError(f"OIDC client {client.uuid} already exists")
@@ -855,9 +853,7 @@ def reset_oid_client_secret(
_db.oidc.clients[client_uuid] = updated
def delete_oid_client(
client_uuid: UUID, *, ctx: SessionContext | None = None
) -> None:
def delete_oid_client(client_uuid: UUID, *, ctx: SessionContext | None = None) -> None:
"""Delete an OIDC client."""
if client_uuid not in _db.oidc.clients:
raise ValueError(f"OIDC client {client_uuid} not found")
paskia/domains.py
+1 -3
View File
@@ -436,9 +436,7 @@ def _derive_site(
if rp_id in domain.origins:
return origin_url(rp_id), "/auth/"
concrete = sorted(
k
for k in domain.origins
if k != "*" and not hostutil.is_wildcard_pattern(k)
k for k in domain.origins if k != "*" and not hostutil.is_wildcard_pattern(k)
)
if concrete:
return origin_url(concrete[0]), "/auth/"
paskia/fastapi/admin/oidc_clients.py
+1 -3
View File
@@ -201,9 +201,7 @@ async def admin_reset_oidc_client_secret(
raise ValueError("secret_hash must be a SHA-256 hash (32 bytes)")
try:
db.reset_oid_client_secret(
client_uuid, secret_hash, ctx=ctx
)
db.reset_oid_client_secret(client_uuid, secret_hash, ctx=ctx)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
paskia/fastapi/dispatch.py
+1 -3
View File
@@ -74,9 +74,7 @@ class DispatchMiddleware:
if origin_domain is not None and origin_domain is not host_domain:
# Cross-domain connection: only via the origin domain's own auth host.
own = origin_domain.own_auth_host
if not own or hostutil.normalize_host(
host
) != hostutil.normalize_host(own):
if not own or hostutil.normalize_host(host) != hostutil.normalize_host(own):
await send(
{"type": "websocket.close", "code": _WS_CLOSE_POLICY_VIOLATION}
)
paskia/fastapi/ws.py
+1 -3
View File
@@ -148,9 +148,7 @@ async def websocket_authenticate(
await ws.send_json({"status": 400, "detail": "Invalid redirect_uri"})
return
# Store as the only allowed redirect URI
db.update_oid_client(
client_uuid, redirect_uris=[redirect_uri]
)
db.update_oid_client(client_uuid, redirect_uris=[redirect_uri])
# Reload client to get updated redirect_uris
oidc_client = db.data().oidc.clients.get(client_uuid)
elif redirect_uri not in oidc_client.redirect_uris:
tests/test_admin.py
+12 -8
View File
@@ -1830,7 +1830,7 @@ class TestDomains:
assert domain["rp_id"] == "localhost"
assert domain["origins"] == {}
assert domain["related"] == {}
assert domain["effective_auth_host"] is None
assert domain["auth_host"] is None
assert domain["site_url"] == "http://localhost:4401"
@pytest.mark.asyncio
@@ -1945,9 +1945,6 @@ class TestDomains:
assert created["rp_name"] == "Example"
assert created["related"] == {"unrelated-site.com": True}
# OIDC provider seeded for the new domain
assert db.data().oidc_for("example.com") is not None
r = await client.delete("/auth/api/admin/domains/example.com", headers=headers)
assert r.status_code == 200, r.text
assert "example.com" not in db.data().config.domains
@@ -2102,14 +2099,15 @@ class TestDomains:
assert r.status_code == 200, r.text
@pytest.mark.asyncio
async def test_effective_auth_host_fallback(
async def test_no_cross_domain_auth_host_fallback(
self,
client: httpx.AsyncClient,
session_token: str,
test_user,
test_credential,
):
"""A domain without its own auth host uses the shared one in settings."""
"""A domain without its own auth host reports none — there is no
cross-domain fallback to another domain's auth host."""
headers = await self._set_auth_host(
client, session_token, test_user, test_credential
)
@@ -2118,9 +2116,15 @@ class TestDomains:
)
assert r.status_code == 200
# Settings on the example.com host report the shared effective auth host
# Settings on the example.com host report no auth host
r = await client.get("/auth/api/settings", headers={"Host": "example.com"})
assert r.status_code == 200
assert r.json()["rp_id"] == "example.com"
assert r.json()["auth_host"] == "auth.localhost"
assert r.json()["auth_host"] is None
assert r.json()["own_auth_host"] is None
# The localhost domain still reports its own auth host
r = await client.get("/auth/api/settings", headers={"Host": "auth.localhost"})
assert r.status_code == 200
assert r.json()["auth_host"] == "auth.localhost"
assert r.json()["own_auth_host"] == "auth.localhost"
tests/test_api.py
+1 -2
View File
@@ -731,10 +731,9 @@ class TestOidcUserInfoEndpoint:
if store is None:
raise RuntimeError("Test DB store is not initialized")
with store.transaction("create_test_oidc_client"):
test_db.oidc["localhost"].clients[oidc_client.uuid] = oidc_client
test_db.oidc.clients[oidc_client.uuid] = oidc_client
access_token = oidjwt.create_access_token(
"localhost",
issuer="http://localhost:4401",
subject=test_user.uuid,
audience=str(oidc_client.uuid),
tests/test_cli.py
+9 -4
View File
@@ -110,17 +110,22 @@ def test_init_adds_domains_to_existing_database(run_cli, tmp_path):
config = stored_config(tmp_path)
assert list(config.domains) == ["company.com", "app.com", "pro.com"]
assert config.domains["pro.com"].rp_name == "Pro Corp"
# OIDC providers seeded for the added domains
assert set(converted_oidc(tmp_path)) == {"company.com", "app.com", "pro.com"}
def converted_oidc(tmp_path):
def test_init_seeds_one_global_oidc_key(run_cli, tmp_path):
"""OIDC is instance-global: init seeds a single signing key."""
run_cli("init", "company.com")
run_cli("init", "app.com")
assert converted_oidc_key(tmp_path) is not None
def converted_oidc_key(tmp_path):
async def _read():
new_db = DB()
kanta = Kanta(str(tmp_path / "paskia.kantadb"), new_db)
await kanta.open(readonly=True)
try:
return set(kanta.data.oidc)
return kanta.data.oidc.key
finally:
await kanta.close()
tests/test_domains.py
+154 -40
View File
@@ -26,9 +26,18 @@ from paskia.db.legacy import (
)
from paskia.db.lifecycle import format_log_uuid
from paskia.db.operations import DB
from paskia.db.structs import Client, Config, Credential, DomainConfig, OriginEntry
from paskia.db.structs import (
OIDC,
Client,
Config,
Credential,
DomainConfig,
OriginEntry,
Session,
)
from paskia.fastapi.dispatch import DispatchMiddleware
from paskia.sansio import Passkey
from paskia.util.crypto import hash_secret
# -------------------------------------------------------------------------
# Registry construction helpers
@@ -143,16 +152,26 @@ class TestResolve:
assert reg.resolve("") is None
assert reg.resolve(None) is None
def test_effective_auth_host_fallback(self):
def test_auth_host_is_per_domain_no_fallback(self):
reg = build_registry(ROR_CONFIG.domains)
company = reg.get("company.com")
pro = reg.get("pro.com")
assert reg.effective_auth_host(company) == "auth.company.com"
# pro.com has no own auth host: falls back to the first configured one
assert reg.effective_auth_host(pro) == "auth.company.com"
# No auth hosts at all: None
reg2 = build_registry({"a.com": DomainConfig(), "b.com": DomainConfig()})
assert reg2.effective_auth_host(reg2.get("a.com")) is None
assert reg.get("company.com").own_auth_host == "auth.company.com"
# pro.com has no own auth host and there is no cross-domain fallback
assert reg.get("pro.com").own_auth_host is None
def test_shared_auth_host_resolves_best_suffix(self):
"""Domains may share an auth host (nested rp-ids); the longest
rp-id suffix match wins, first configured as tiebreak."""
reg = build_registry(
{
"com": DomainConfig(
origins={"auth.company.com": OriginEntry(auth_host=True)}
),
"company.com": DomainConfig(
origins={"auth.company.com": OriginEntry(auth_host=True)}
),
}
)
assert reg.resolve("auth.company.com").rp_id == "company.com"
# -------------------------------------------------------------------------
@@ -255,13 +274,39 @@ class TestValidateConfig:
)
)
def test_related_origin_inside_other_domain(self):
with pytest.raises(ValueError, match="falls inside domain"):
def test_related_origin_may_fall_inside_other_domain(self):
"""A related origin at/inside another domain's rp-id is allowed.
The related listing wins dispatch over suffix matching (a host that
*is* a configured rp-id always serves its own domain)."""
config = Config(
domains={
"a.com": DomainConfig(related={"app.b.com": True}),
"b.com": DomainConfig(),
}
)
domains.validate_config(config)
reg = build_registry(config.domains)
assert reg.resolve("app.b.com").rp_id == "a.com"
assert reg.resolve("b.com").rp_id == "b.com"
def test_related_origin_shared_when_covered_by_rp_id(self):
"""Two domains may list the same related host when it falls inside
a configured rp-id; otherwise the collision is rejected."""
domains.validate_config(
Config(
domains={
"a.com": DomainConfig(related={"app.b.com": True}),
"c.com": DomainConfig(related={"app.b.com": True}),
"b.com": DomainConfig(),
}
)
)
with pytest.raises(ValueError, match="configured for both"):
domains.validate_config(
Config(
domains={
"a.com": DomainConfig(related={"app.b.com": True}),
"b.com": DomainConfig(),
"a.com": DomainConfig(related={"shared.com": True}),
"c.com": DomainConfig(related={"shared.com": True}),
}
)
)
@@ -369,7 +414,8 @@ class TestSanitizeConfig:
assert config.domains["a.com"].origins == {"auth.a.com": True}
assert any("collides" in w for w in warnings)
def test_related_colliding_with_other_domain_dropped(self):
def test_related_inside_other_domain_kept(self):
"""A related origin falling inside another domain's rp-id is kept."""
config, _ = domains.sanitize_config(
Config(
domains={
@@ -378,7 +424,21 @@ class TestSanitizeConfig:
}
)
)
assert config.domains["a.com"].related == {}
assert config.domains["a.com"].related == {"app.b.com": True}
def test_related_claimed_twice_first_wins(self):
"""Two non-owner domains claiming one related host: first wins."""
config, warnings = domains.sanitize_config(
Config(
domains={
"a.com": DomainConfig(related={"shared.com": True}),
"b.com": DomainConfig(related={"shared.com": True}),
}
)
)
assert config.domains["a.com"].related == {"shared.com": True}
assert config.domains["b.com"].related == {}
assert any("first domain wins" in w for w in warnings)
def test_no_domains_is_fatal(self):
with pytest.raises(ValueError, match="No servable domain"):
@@ -440,14 +500,23 @@ class TestOriginValidation:
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("http://app.example.com:8080")
def test_star_entry_matches_any_scheme_and_port(self):
"""The bare '*' entry allows anything within the rp-id domain."""
def test_star_entry_is_wildcard_shorthand(self):
"""The bare '*' entry is shorthand for '*.{rp-id}' — https only."""
p = Passkey(rp_id="example.com", origins=["*"])
assert p.validate_origin("https://example.com")
assert p.validate_origin("http://app.example.com:8080")
assert p.validate_origin("https://app.example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("http://app.example.com:8080")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://other.com")
def test_localhost_wildcard_matches_any_scheme_and_port(self):
"""Under localhost, wildcards match any scheme and any port."""
p = Passkey(rp_id="localhost", origins=["*"])
assert p.validate_origin("http://localhost:8080")
assert p.validate_origin("http://app.localhost:3000")
assert p.validate_origin("https://localhost")
def test_exact_entry_matches_scheme_and_port(self):
p = Passkey(rp_id="localhost", origins=["http://localhost:4403"])
assert p.validate_origin("http://localhost:4403")
@@ -555,15 +624,16 @@ class TestDispatchMiddleware:
assert stub.scope["state"]["domain"].rp_id == "company.com"
@pytest.mark.asyncio
async def test_ws_cross_domain_requires_effective_auth_host(self):
async def test_ws_cross_domain_requires_origin_own_auth_host(self):
build_registry(ROR_CONFIG.domains)
# pro.com page connecting to the shared auth host: allowed, pro domain
# pro.com has no own auth host: its page may only connect to pro.com
# hosts — the company.com auth host does not serve foreign domains
stub, sent = await drive_ws(
DispatchMiddleware(StubApp()),
[(b"host", b"auth.company.com"), (b"origin", b"https://pro.com")],
)
assert sent == []
assert stub.scope["state"]["domain"].rp_id == "pro.com"
assert stub.scope is None
assert sent == [{"type": "websocket.close", "code": 1008}]
# pro.com page connecting to some other host: closed pre-accept
stub, sent = await drive_ws(
@@ -573,6 +643,35 @@ class TestDispatchMiddleware:
assert stub.scope is None
assert sent == [{"type": "websocket.close", "code": 1008}]
@pytest.mark.asyncio
async def test_ws_cross_domain_via_own_auth_host(self):
"""On a shared auth host (nested rp-ids), the WS Origin selects the
domain: plain HTTP resolves to the longest-suffix claimant, but a
WebSocket from another claimant's page is dispatched by Origin."""
build_registry(
{
"com": DomainConfig(
origins={"auth.company.com": OriginEntry(auth_host=True)}
),
"company.com": DomainConfig(
origins={"auth.company.com": OriginEntry(auth_host=True)}
),
}
)
# Host alone resolves to company.com (longest suffix)
stub, sent = await drive_ws(
DispatchMiddleware(StubApp()), [(b"host", b"auth.company.com")]
)
assert stub.scope["state"]["domain"].rp_id == "company.com"
# A page on com (the other claimant) is accepted: the Host is its
# own auth host, and the Origin selects its domain
stub, sent = await drive_ws(
DispatchMiddleware(StubApp()),
[(b"host", b"auth.company.com"), (b"origin", b"https://com")],
)
assert sent == []
assert stub.scope["state"]["domain"].rp_id == "com"
@pytest.mark.asyncio
async def test_ws_unknown_origin_uses_host_domain(self):
build_registry(ROR_CONFIG.domains)
@@ -614,25 +713,40 @@ class TestAuthCodeDomainBinding:
assert response.status_code == 401
@pytest.mark.asyncio
async def test_oidc_code_rejected_on_other_domain(
self, client: httpx.AsyncClient, test_db: DB
async def test_oidc_code_is_host_independent(
self, client: httpx.AsyncClient, test_db: DB, test_user, test_credential
):
"""OIDC codes carry no domain binding: the provider is
instance-global, so a code is redeemable at any host."""
oidc_client, secret = Client.create(
name="Test Client",
redirect_uris=["https://client.example/callback"],
client_secret="topsecret",
)
token = "doesnotmatter1234"
session = Session.create(
user=test_user.uuid,
credential=test_credential.uuid,
key=hash_secret("oidc", token),
host="other.com",
ip="127.0.0.1",
user_agent="pytest",
validated=datetime.now(UTC),
client=oidc_client.uuid,
rp_id="other.com",
issuer="https://other.com",
)
store = test_db._store
with store.transaction("create_test_oidc_client"):
test_db.oidc["localhost"].clients[oidc_client.uuid] = oidc_client
with store.transaction("seed_oidc_session"):
test_db.oidc.clients[oidc_client.uuid] = oidc_client
test_db.sessions[session.key] = session
code = authcode.store_oidc(
authcode.OIDCCode(
session_key="doesnotmatter1234",
session_key=token,
created=datetime.now(UTC),
redirect_uri="https://client.example/callback",
scope="openid",
rp_id="other.com",
)
)
response = await client.post(
@@ -646,9 +760,8 @@ class TestAuthCodeDomainBinding:
},
headers={"Host": "localhost:4401"},
)
assert response.status_code == 400
assert response.json()["error"] == "invalid_grant"
assert "domain" in response.json()["error_description"]
assert response.status_code == 200
assert response.json()["access_token"]
# -------------------------------------------------------------------------
@@ -700,6 +813,7 @@ class TestLegacyConversion:
user_agent="pytest",
validated=datetime.now(UTC),
)
kanta.data.oidc = OIDC(key=b"legacy-signing-key")
await kanta.close()
asyncio.run(_write())
@@ -707,13 +821,14 @@ class TestLegacyConversion:
config = convert_legacy_database(src_file, tmp_path / "paskia.kantadb")
domain = config.domains["example.com"]
assert domain.rp_name == "Example"
# A legacy wildcard over the rp-id itself becomes the bare '*'
assert domain.origins == {"app.example.com": True, "*": True}
# Legacy wildcard origins convert as-is (https-only outside localhost)
assert domain.origins == {"app.example.com": True, "*.example.com": True}
converted = _read_db(tmp_path / "paskia.kantadb")
assert converted.credentials[cred_uuid].rp_id == "example.com"
assert converted.sessions["session-key"].rp_id == "example.com"
assert set(converted.oidc.keys()) == {"example.com"}
# The legacy OIDC provider carries over as the instance-global one
assert converted.oidc.key == b"legacy-signing-key"
# -------------------------------------------------------------------------
@@ -723,15 +838,14 @@ class TestLegacyConversion:
class TestLogCensoring:
def test_oidc_key_values_hidden(self):
assert format_log_uuid(b"raw-key-material", "oidc.localhost.key") == "<hidden>"
assert format_log_uuid("secret", "oidc.example.com.key") == "<hidden>"
assert format_log_uuid(b"raw-key-material", "oidc.key") == "<hidden>"
def test_oidc_key_path_component_visible(self):
# The path component itself must stay visible ("oidc.<rp-id>.key = <hidden>")
assert format_log_uuid("key", "oidc.localhost.key") is None
# The path component itself must stay visible ("oidc.key = <hidden>")
assert format_log_uuid("key", "oidc.key") is None
def test_other_paths_unaffected(self):
assert format_log_uuid("not-a-uuid", "oidc.localhost.clients") is None
assert format_log_uuid("not-a-uuid", "oidc.clients") is None
assert format_log_uuid("not-a-uuid", "config.domains") is None