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

MultiSite: one instance serves authentication across many domains (#4)

Browse Source 
- Serve multiple domains (RP IDs) from one instance: host-based dispatch,
  per-domain credentials and sessions, domains managed at runtime in the
  admin UI — previously one RP per instance
- Cross-domain sign-in via Related Origin Requests: per-domain related-origins
  list with a served .well-known/webauthn document
- Explicit per-domain origin lists with shell-glob wildcards (**. for apex +
  any subdomain depth, *. for one level), editable in the admin UI with
  validation and self-lockout guards
- Per-domain auth hosts: the account/admin UI can live on a different host
  per domain, no longer confined to subdomains of a single RP
- CLI: 'paskia init <rp-id [rp-name]' initializes or adds a domain to an
  existing database; 'paskia migrate' converts legacy databases

BREAKING CHANGES (v2.0):
- Database schema: config is now per-domain and credentials/sessions carry
  an rp_id — existing databases must be converted with 'paskia migrate'
- Origins are now explicit: main implicitly allowed every subdomain of the
  RP; configure '**.' origins to reproduce that behavior
- CLI: the flat '--rp-id/--rp-name/--origin/--auth/--save' flags are
  replaced by the 'init' and 'migrate' subcommandsReviewed-on: #4
This commit is contained in:
LeoVasanko
2026-09-07 22:14:42 +00:00
parent 383c9f472e
commit 84985501f5
78 changed files with 5291 additions and 1484 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+37 -51
View File
@@ -12,12 +12,12 @@ in the database to test authenticated endpoints.
from __future__ import annotations
import asyncio
import json
import os
import secrets
import tempfile
from collections.abc import AsyncGenerator
from datetime import UTC, datetime, timedelta
from pathlib import Path
from uuid import UUID
import httpx
@@ -25,22 +25,8 @@ import pytest
import pytest_asyncio
from kanta import Kanta
# Keep runtime initialization invariant aligned with production:
# db.lifecycle requires PASKIA_CONFIG at import time.
os.environ.setdefault(
"PASKIA_CONFIG",
json.dumps(
{
"config": {"rp_id": "localhost", "rp_name": "localhost"},
"site_url": "http://localhost:4401",
"site_path": "/auth/",
"save": False,
}
),
)
import paskia.db.operations as ops_db
from paskia import globals as paskia_globals
from paskia import domains
from paskia.authsession import reset_expires
from paskia.config import SESSION_LIFETIME
from paskia.db import (
@@ -56,12 +42,15 @@ from paskia.db import (
)
from paskia.db.bootstrap import bootstrap
from paskia.db.operations import DB
from paskia.db.structs import Session
from paskia.db.structs import Config, DomainConfig, Session
from paskia.fastapi.mainapp import app
from paskia.fastapi.session import AUTH_COOKIE_NAME
from paskia.sansio import Passkey
from paskia.util import avatar
from paskia.util.crypto import hash_secret
TEST_RP_ID = "localhost"
TEST_LISTEN = ["localhost:4401"]
@pytest.fixture(scope="session")
def event_loop():
@@ -71,6 +60,19 @@ def event_loop():
loop.close()
@pytest.fixture(autouse=True)
def _avatar_tmp_root(tmp_path, monkeypatch):
"""Redirect avatar storage to a per-test temporary directory."""
root = tmp_path / "users"
def users_root(create_root: bool = False) -> Path:
if create_root:
root.mkdir(parents=True, exist_ok=True)
return root
monkeypatch.setattr(avatar, "users_root_path", users_root)
@pytest_asyncio.fixture(scope="function")
async def test_db() -> AsyncGenerator[DB]:
"""Create a temporary JSONL database for testing using kanta.
@@ -79,15 +81,11 @@ async def test_db() -> AsyncGenerator[DB]:
- auth:admin and auth:org:admin permissions
- A default organization with Administration role
- An admin user with the Administration role
- The localhost domain configuration (with its OIDC provider)
"""
with tempfile.NamedTemporaryFile(suffix=".jsonl", delete=True) as f:
db = DB()
kanta = Kanta(
f.name,
db,
migrations="paskia.db.migrations",
)
kanta.ctx.rp_id = "test.example.com"
kanta = Kanta(f.name, db)
# Register bootstrap callback so kanta seeds the empty DB during open()
@kanta.bootstrap(action="bootstrap")
@@ -96,6 +94,11 @@ async def test_db() -> AsyncGenerator[DB]:
data,
org_name="Test Organization",
admin_name="Test Admin",
config=Config(
domains={
TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True})
}
),
)
await kanta.open()
@@ -107,25 +110,10 @@ async def test_db() -> AsyncGenerator[DB]:
@pytest_asyncio.fixture(scope="function")
async def passkey_instance() -> Passkey:
"""Override the module-level passkey instance for testing."""
pk = Passkey(
rp_id="localhost",
rp_name="Test RP",
origins=["http://localhost:4401"],
)
original = {
"rp_id": paskia_globals.passkey.rp_id,
"rp_name": paskia_globals.passkey.rp_name,
"allowed_origins": paskia_globals.passkey.allowed_origins,
}
paskia_globals.passkey.rp_id = pk.rp_id
paskia_globals.passkey.rp_name = pk.rp_name
paskia_globals.passkey.allowed_origins = pk.allowed_origins
yield pk
paskia_globals.passkey.rp_id = original["rp_id"]
paskia_globals.passkey.rp_name = original["rp_name"]
paskia_globals.passkey.allowed_origins = original["allowed_origins"]
async def domain_registry(test_db: DB) -> domains.DomainRegistry:
"""Install the domain registry built from the test database config."""
domains.configure(listen=TEST_LISTEN)
return domains.init_registry(test_db.config)
@pytest_asyncio.fixture(scope="function")
@@ -192,6 +180,7 @@ async def test_credential(test_db: DB, test_user: User) -> Credential:
aaguid=UUID("00000000-0000-0000-0000-000000000000"),
public_key=os.urandom(64),
sign_count=0,
rp_id=TEST_RP_ID,
)
create_credential(credential)
return credential
@@ -206,6 +195,7 @@ async def regular_credential(test_db: DB, regular_user: User) -> Credential:
aaguid=UUID("00000000-0000-0000-0000-000000000000"),
public_key=os.urandom(64),
sign_count=0,
rp_id=TEST_RP_ID,
)
create_credential(credential)
return credential
@@ -247,15 +237,9 @@ async def reset_token(test_db: DB, test_user: User, test_credential: Credential)
@pytest_asyncio.fixture(scope="function")
async def client(
test_db: DB, passkey_instance: Passkey
test_db: DB, domain_registry: domains.DomainRegistry
) -> AsyncGenerator[httpx.AsyncClient]:
"""Create an async test client for the FastAPI app.
Note: We import the app inside the fixture to ensure globals are
initialized first.
"""
# Import app after globals are set
"""Create an async test client for the FastAPI app."""
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport,
@@ -283,6 +267,7 @@ def create_test_session(
ip: str = "127.0.0.1",
user_agent: str = "pytest",
duration: timedelta | None = None,
rp_id: str = TEST_RP_ID,
) -> tuple[str, str]:
"""Create a test session. Returns (key, token) tuple.
@@ -309,6 +294,7 @@ def create_test_session(
ip=ip,
user_agent=user_agent,
validated=now,
rp_id=rp_id,
)
if session.key in ops_db._db.sessions:
raise ValueError("Session already exists")
tests/test_admin.py
+303 -53
View File
@@ -22,7 +22,7 @@ import pytest
import pytest_asyncio
import uuid7
from paskia import db
from paskia import db, domains
from paskia.db import (
Credential,
Org,
@@ -37,10 +37,7 @@ from paskia.db import (
create_user,
)
from paskia.db.operations import DB
from paskia.util import hostutil
from paskia.util.crypto import hash_secret
from paskia.util.runtime import clear_config_cache
from paskia.util.runtime import config as runtime_config
from tests.conftest import auth_headers, create_test_image_bytes, create_test_session
# -------------------- Additional Fixtures --------------------
@@ -91,6 +88,7 @@ async def second_org_credential(test_db: DB, second_org_user: User) -> Credentia
aaguid=UUID("00000000-0000-0000-0000-000000000000"),
public_key=os.urandom(64),
sign_count=0,
rp_id="localhost",
)
create_credential(credential)
return credential
@@ -145,6 +143,7 @@ async def org_admin_credential(test_db: DB, org_admin_user: User) -> Credential:
aaguid=UUID("00000000-0000-0000-0000-000000000000"),
public_key=os.urandom(64),
sign_count=0,
rp_id="localhost",
)
create_credential(credential)
return credential
@@ -253,8 +252,6 @@ class TestAdminOrganizations:
monkeypatch,
):
"""Admin org payload should include canonical avatar URLs for listed users."""
monkeypatch.setenv("PASKIA_DB", str(tmp_path / "test-avatar-db.paskiadb"))
upload = await client.put(
f"/auth/api/user/{test_user.uuid}/profile.webp",
files={"file": ("avatar.webp", create_test_image_bytes(), "image/webp")},
@@ -948,8 +945,6 @@ class TestAdminUsersInOrg:
monkeypatch,
):
"""Admin should be able to upload avatar for a managed user."""
monkeypatch.setenv("PASKIA_DB", str(tmp_path / "test-admin-avatar-db.paskiadb"))
response = await client.put(
f"/auth/api/user/{test_user.uuid}/profile.webp",
files={"file": ("avatar.webp", create_test_image_bytes(), "image/webp")},
@@ -1794,38 +1789,60 @@ class TestOrgAdminAuthExceptions:
assert response.status_code == 403
class TestServerConfig:
"""Tests for GET/PATCH /auth/api/admin/server-config/ runtime updates."""
@pytest.fixture(scope="function")
def restore_runtime_config(self):
"""Restore PASKIA_CONFIG env and cache after a test mutates runtime."""
original = os.environ["PASKIA_CONFIG"]
yield
os.environ["PASKIA_CONFIG"] = original
clear_config_cache()
class TestDomains:
"""Tests for the domain management API (/auth/api/admin/domains/)."""
async def _set_auth_host(self, client, session_token, test_user, test_credential):
"""Configure an auth host via PATCH, as the admin UI would."""
"""Configure an auth host on the localhost domain, as the admin UI would."""
r = await client.patch(
"/auth/api/admin/server-config/",
"/auth/api/admin/domains/localhost",
json={
"rp_name": "",
"auth_host": "auth.localhost",
"origins": ["auth.localhost", "localhost"],
"origins": {
"auth.localhost": {"auth_host": True},
"localhost": True,
},
},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert r.status_code == 200, r.text
assert db.data().config.auth_host == "https://auth.localhost"
assert hostutil.dedicated_auth_host() == "auth.localhost"
assert hostutil.auth_site_url() == "https://auth.localhost/"
domain_cfg = db.data().config.domains["localhost"]
assert domains.auth_host_url(domain_cfg) == "https://auth.localhost"
domain = domains.registry().get("localhost")
assert domain.own_auth_host == "auth.localhost"
assert domain.auth_site_url == "https://auth.localhost/"
# Session for requests coming from the auth host (sessions are host-bound)
_, token = create_test_session(
test_user.uuid, test_credential.uuid, host="auth.localhost"
)
return {**auth_headers(token), "Host": "auth.localhost"}
@pytest.mark.asyncio
async def test_list_domains(self, client: httpx.AsyncClient, session_token: str):
r = await client.get(
"/auth/api/admin/domains/",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert r.status_code == 200, r.text
data = r.json()
assert len(data) == 1
domain = data[0]
assert domain["rp_id"] == "localhost"
assert domain["origins"] == {"**.localhost": True}
assert "related" not in domain
assert domain["auth_host"] is None
assert domain["site_url"] == "http://localhost:4401"
@pytest.mark.asyncio
async def test_domains_require_master_admin(
self, client: httpx.AsyncClient, regular_session_token: str
):
r = await client.get(
"/auth/api/admin/domains/",
headers={**auth_headers(regular_session_token), "Host": "localhost:4401"},
)
assert r.status_code in (401, 403)
@pytest.mark.asyncio
async def test_remove_auth_host_updates_runtime(
self,
@@ -1833,41 +1850,41 @@ class TestServerConfig:
session_token: str,
test_user,
test_credential,
restore_runtime_config,
):
"""Removing auth_host must clear it from runtime config and URLs."""
"""Removing the auth host mark must clear it from runtime config and URLs."""
headers = await self._set_auth_host(
client, session_token, test_user, test_credential
)
# The dialog still lists the old auth host among origins, so it is sent back
r = await client.patch(
"/auth/api/admin/server-config/",
"/auth/api/admin/domains/localhost",
json={
"rp_name": "",
"auth_host": "",
"origins": ["auth.localhost", "localhost"],
"origins": {"auth.localhost": True, "localhost": True},
},
headers=headers,
)
assert r.status_code == 200, r.text
assert db.data().config.auth_host is None
domain_cfg = db.data().config.domains["localhost"]
assert domains.auth_host_url(domain_cfg) is None
rt = runtime_config()
assert rt.config.auth_host is None
assert rt.site_path == "/auth/"
assert "auth.localhost" not in rt.site_url
assert hostutil.dedicated_auth_host() is None
assert "auth.localhost" not in hostutil.auth_site_url()
domain = domains.registry().get("localhost")
assert domain.own_auth_host is None
assert domain.ui_base_path == "/auth/"
# Site URL derivation is stateless: with the auth host mark removed,
# the exact rp-id origin becomes the site URL.
assert domain.auth_site_url == "https://localhost/auth/"
# GET and settings reflect the cleared state
r = await client.get(
"/auth/api/admin/server-config/",
"/auth/api/admin/domains/",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert r.json()["auth_host"] == ""
assert r.json()[0]["origins"] == {"auth.localhost": True, "localhost": True}
r = await client.get("/auth/api/settings")
assert r.json()["auth_host"] is None
assert r.json()["own_auth_host"] is None
assert r.json()["ui_base_path"] == "/auth/"
# Middleware no longer redirects to the removed auth host
@@ -1879,28 +1896,261 @@ class TestServerConfig:
assert "auth.localhost" not in r.headers.get("location", "")
@pytest.mark.asyncio
async def test_remove_auth_host_without_origins_falls_back_to_rp_id(
async def test_remove_auth_host_without_origins_falls_back(
self,
client: httpx.AsyncClient,
session_token: str,
):
"""Emptying a domain's origins table must not keep the removed auth
host in derived URLs. Only possible on a domain other than the one
in use — the lockout guard refuses it there."""
headers = {**auth_headers(session_token), "Host": "localhost:4401"}
r = await client.post(
"/auth/api/admin/domains/",
json={
"rp_id": "example.com",
"origins": {
"auth.example.com": {"auth_host": True},
"app.example.com": True,
},
},
headers=headers,
)
assert r.status_code == 200, r.text
domain = domains.registry().get("example.com")
assert domain.own_auth_host == "auth.example.com"
assert "auth.example.com" in domain.site_url
r = await client.patch(
"/auth/api/admin/domains/example.com",
json={"rp_name": "", "origins": {}},
headers=headers,
)
assert r.status_code == 200, r.text
domain = domains.registry().get("example.com")
assert domain.own_auth_host is None
assert domain.ui_base_path == "/auth/"
assert "auth.example.com" not in domain.site_url
assert "auth.example.com" not in domain.auth_site_url
@pytest.mark.asyncio
async def test_create_and_delete_domain(
self, client: httpx.AsyncClient, session_token: str
):
headers = {**auth_headers(session_token), "Host": "localhost:4401"}
r = await client.post(
"/auth/api/admin/domains/",
json={
"rp_id": "example.com",
"rp_name": "Example",
"origins": {"app.example.com": True, "unrelated-site.com": True},
},
headers=headers,
)
assert r.status_code == 200, r.text
r = await client.get("/auth/api/admin/domains/", headers=headers)
domains_list = {domain["rp_id"]: domain for domain in r.json()}
assert set(domains_list) == {"localhost", "example.com"}
created = domains_list["example.com"]
assert created["rp_name"] == "Example"
# In-domain and related origins live in one table; classification
# is derived from the rp-id
assert created["origins"] == {
"app.example.com": True,
"unrelated-site.com": True,
}
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
assert domains.registry().get("example.com") is None
@pytest.mark.asyncio
async def test_create_domain_validation(
self, client: httpx.AsyncClient, session_token: str
):
headers = {**auth_headers(session_token), "Host": "localhost:4401"}
# rp_id is required
r = await client.post("/auth/api/admin/domains/", json={}, headers=headers)
assert r.status_code == 400
# Duplicate rp-id
r = await client.post(
"/auth/api/admin/domains/", json={"rp_id": "localhost"}, headers=headers
)
assert r.status_code == 400
# Invalid rp-id
r = await client.post(
"/auth/api/admin/domains/", json={"rp_id": "not a domain!"}, headers=headers
)
assert r.status_code == 400
# An auth host must be within the rp-id domain
r = await client.post(
"/auth/api/admin/domains/",
json={
"rp_id": "example.com",
"origins": {"auth.other.com": {"auth_host": True}},
},
headers=headers,
)
assert r.status_code == 400
# Related origin host may not collide across domains
r = await client.post(
"/auth/api/admin/domains/",
json={"rp_id": "example.com", "origins": {"shared-app.com": True}},
headers=headers,
)
assert r.status_code == 200
r = await client.post(
"/auth/api/admin/domains/",
json={"rp_id": "other.com", "origins": {"shared-app.com": True}},
headers=headers,
)
assert r.status_code == 400
# Cross-domain entries are related origins — accepted in the same table
r = await client.post(
"/auth/api/admin/domains/",
json={"rp_id": "another.com", "origins": {"elsewhere.com": True}},
headers=headers,
)
assert r.status_code == 200
# Plain '*' is rejected — wildcards must be explicit ('**.another.com')
r = await client.post(
"/auth/api/admin/domains/",
json={"rp_id": "star.com", "origins": {"*": True}},
headers=headers,
)
assert r.status_code == 400
@pytest.mark.asyncio
async def test_delete_domain_guards(
self, client: httpx.AsyncClient, session_token: str, test_credential
):
headers = {**auth_headers(session_token), "Host": "localhost:4401"}
# Cannot delete the last domain
r = await client.delete("/auth/api/admin/domains/localhost", headers=headers)
assert r.status_code == 400
# Unknown domain
r = await client.delete("/auth/api/admin/domains/nope.com", headers=headers)
assert r.status_code == 400
# A domain with credentials still registered under it cannot be deleted
r = await client.post(
"/auth/api/admin/domains/", json={"rp_id": "example.com"}, headers=headers
)
assert r.status_code == 200
cred = Credential.create(
credential_id=secrets.token_bytes(32),
user=test_credential.user_uuid,
aaguid=UUID("00000000-0000-0000-0000-000000000000"),
public_key=secrets.token_bytes(64),
sign_count=0,
rp_id="example.com",
)
create_credential(cred)
r = await client.delete("/auth/api/admin/domains/example.com", headers=headers)
assert r.status_code == 400
@pytest.mark.asyncio
async def test_update_domain_refuses_self_lockout(
self, client: httpx.AsyncClient, session_token: str
):
"""An allow-list excluding the admin's current host is refused."""
headers = {**auth_headers(session_token), "Host": "localhost:4401"}
# Allow-list without the current host and no auth host → lockout
r = await client.patch(
"/auth/api/admin/domains/localhost",
json={"rp_name": "", "origins": {"auth.localhost": True}},
headers=headers,
)
assert r.status_code == 400
assert "lock you out" in r.text
# Emptying the origins table entirely is likewise a lockout
r = await client.patch(
"/auth/api/admin/domains/localhost",
json={"rp_name": "", "origins": {}},
headers=headers,
)
assert r.status_code == 400
assert "lock you out" in r.text
# Allow-list including the current host is fine
r = await client.patch(
"/auth/api/admin/domains/localhost",
json={"rp_name": "", "origins": {"localhost:4401": True}},
headers=headers,
)
assert r.status_code == 200, r.text
# An allow-list without the current host is also fine when an auth
# host is set: ceremonies move there (and it is always allowed).
# Done last: with an auth host set, the API here routes differently.
r = await client.patch(
"/auth/api/admin/domains/localhost",
json={
"rp_name": "",
"origins": {"auth.localhost": {"auth_host": True}},
},
headers=headers,
)
assert r.status_code == 200, r.text
@pytest.mark.asyncio
async def test_delete_current_domain_refused(
self, client: httpx.AsyncClient, session_token: str
):
headers = {**auth_headers(session_token), "Host": "localhost:4401"}
r = await client.post(
"/auth/api/admin/domains/", json={"rp_id": "example.com"}, headers=headers
)
assert r.status_code == 200
# Deleting the domain in use is refused even if it has no credentials
r = await client.delete("/auth/api/admin/domains/localhost", headers=headers)
assert r.status_code == 400
assert "currently using" in r.text
# Deleting another domain while authenticated here is fine
r = await client.delete("/auth/api/admin/domains/example.com", headers=headers)
assert r.status_code == 200, r.text
@pytest.mark.asyncio
async def test_no_cross_domain_auth_host_fallback(
self,
client: httpx.AsyncClient,
session_token: str,
test_user,
test_credential,
restore_runtime_config,
):
"""With no origins left, site_url must not keep the removed auth host."""
"""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
)
r = await client.patch(
"/auth/api/admin/server-config/",
json={"rp_name": "", "auth_host": "", "origins": []},
headers=headers,
r = await client.post(
"/auth/api/admin/domains/", json={"rp_id": "example.com"}, headers=headers
)
assert r.status_code == 200, r.text
assert r.status_code == 200
rt = runtime_config()
assert rt.config.auth_host is None
assert rt.site_path == "/auth/"
assert "auth.localhost" not in rt.site_url
assert "auth.localhost" not in hostutil.auth_site_url()
# 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"] 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
+17 -18
View File
@@ -18,12 +18,12 @@ from uuid import UUID
import httpx
import pytest
from paskia import authcode, db
from paskia import authcode, db, domains
from paskia.authsession import EXPIRES
from paskia.db import delete_session
from paskia.db.structs import Client
from paskia.db.structs import Client, Config, DomainConfig, OriginEntry
from paskia.fastapi.api import _REFRESH_INTERVAL
from paskia.util import avatar, hostutil, oidjwt, permutil
from paskia.util import avatar, oidjwt, permutil
from paskia.util.crypto import hash_secret
from paskia.util.passphrase import generate
from tests.conftest import auth_headers, create_test_image_bytes, create_test_session
@@ -42,7 +42,7 @@ class TestSettingsEndpoint:
assert "rp_name" in data
assert "session_cookie" in data
assert data["rp_id"] == "localhost"
assert data["rp_name"] == "Test RP"
assert data["rp_name"] == "localhost"
assert data["session_cookie"] == "__Host-paskia"
@pytest.mark.asyncio
@@ -69,16 +69,20 @@ class TestAvatarUrls:
self, tmp_path, monkeypatch
):
"""Absolute avatar URLs should preserve /auth/api even with an auth host."""
db_root = tmp_path / "test-avatar-db.paskiadb"
monkeypatch.setenv("PASKIA_DB", str(db_root))
monkeypatch.setattr(
hostutil,
"api_url",
lambda path="": f"https://auth.zi.fi/auth/api/{path.lstrip('/')}",
domains.configure(listen=None)
domains.init_registry(
Config(
domains={
"zi.fi": DomainConfig(
origins={"auth.zi.fi": OriginEntry(auth_host=True)}
)
}
)
)
user_uuid = test_uuid = UUID("019c6831-84cf-7b88-b66c-c8165890b7c5")
path = db_root / "users" / str(test_uuid) / "profile.webp"
# The autouse avatar fixture redirects storage to tmp_path / "users"
user_uuid = UUID("019c6831-84cf-7b88-b66c-c8165890b7c5")
path = tmp_path / "users" / str(user_uuid) / "profile.webp"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(b"RIFF1234WEBP")
@@ -646,8 +650,6 @@ class TestUserInfoEndpoint:
monkeypatch,
):
"""User info should include the canonical avatar URL when present."""
monkeypatch.setenv("PASKIA_DB", str(tmp_path / "test-avatar-db.paskiadb"))
upload = await client.put(
f"/auth/api/user/{test_user.uuid}/profile.webp",
files={"file": ("avatar.webp", create_test_image_bytes(), "image/webp")},
@@ -676,8 +678,6 @@ class TestUserInfoEndpoint:
monkeypatch,
):
"""Avatar route should honor If-None-Match for unchanged avatars."""
monkeypatch.setenv("PASKIA_DB", str(tmp_path / "test-avatar-db.paskiadb"))
upload = await client.put(
f"/auth/api/user/{test_user.uuid}/profile.webp",
files={"file": ("avatar.webp", create_test_image_bytes(), "image/webp")},
@@ -714,8 +714,6 @@ class TestOidcUserInfoEndpoint:
monkeypatch,
):
"""OIDC userinfo should expose picture when profile scope is granted."""
monkeypatch.setenv("PASKIA_DB", str(tmp_path / "test-avatar-db.paskiadb"))
upload = await client.put(
f"/auth/api/user/{test_user.uuid}/profile.webp",
files={"file": ("avatar.webp", create_test_image_bytes(), "image/webp")},
@@ -776,6 +774,7 @@ class TestSetSessionEndpoint:
authcode.CookieCode(
session_key=session_token,
created=datetime.now(UTC),
rp_id="localhost",
)
)
response = await client.post(
tests/test_cli.py
+234 -102
View File
@@ -1,4 +1,10 @@
"""Tests for the CLI entry point in paskia/__main__.py."""
"""Tests for the CLI entry point in paskia/__main__.py.
The CLI is split into ``paskia init`` (create the combined paskia.kantadb
with the initial domain(s)), ``paskia migrate`` (convert a legacy
``<rp-id>.paskiadb`` database), and bare ``paskia`` (serve the stored
domains; never migrates).
"""
from __future__ import annotations
@@ -6,141 +12,245 @@ import asyncio
import os
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any
import msgspec
import pytest
from kanta import Kanta
from paskia.__main__ import main
from paskia.__main__ import _load_stored_config, main
from paskia.db import legacy
from paskia.db.structs import DB, Config
from paskia.util.runtime import clear_config_cache
from paskia.util.runtime import config as runtime_config
from paskia.util.runtime import ServeConfig, clear_cache
@pytest.fixture
def cli_run(monkeypatch):
"""Run the CLI main() with the given args and return the RuntimeConfig."""
def run_cli(monkeypatch, tmp_path):
"""Run the CLI main() in a temporary working directory.
def _run(*args: str, db_root: str | None = None) -> Any:
env = os.environ.copy()
if db_root is not None:
env["PASKIA_DB"] = db_root
monkeypatch.setattr(os, "environ", env)
Returns a callable; server.run and the startup box are stubbed out.
The returned dict records the server.run invocation (if any).
"""
monkeypatch.chdir(tmp_path)
calls: dict = {}
monkeypatch.setattr(
"fastapi_vue.server.run",
lambda app, **kw: calls.update({"app": app, **kw}),
)
monkeypatch.setattr(
"paskia.util.startupbox.print_startup_config", lambda *a, **kw: None
)
monkeypatch.setattr("logging.basicConfig", lambda **kw: None)
# Isolate environment mutations (PASKIA_CONFIG) from other tests
env = os.environ.copy()
env.pop("PASKIA_CONFIG", None)
env.pop("PASKIA_VITE_URL", None)
monkeypatch.setattr(os, "environ", env)
def _run(*args: str) -> dict:
monkeypatch.setattr(sys, "argv", ["paskia", *args])
monkeypatch.setattr("fastapi_vue.server.run", lambda *_args, **_kw: None)
monkeypatch.setattr(
"paskia.util.startupbox.print_startup_config", lambda _rt: None
)
monkeypatch.setattr("logging.basicConfig", lambda **_kw: None)
clear_config_cache()
main()
runtime = runtime_config()
clear_config_cache()
return runtime
clear_cache()
try:
main()
finally:
clear_cache()
return calls
return _run
async def _write_config(db_path: Path, config: Config) -> None:
"""Write a Config into a JSONL database file using Kanta.
The initial root uses a different rp_id so the stored diff includes the
target rp_id (required because Config omits defaults when diffing).
"""
kanta = Kanta(
str(db_path),
DB(config=Config(rp_id="uninitialized.invalid")),
migrations="paskia.db.migrations",
)
kanta.ctx.rp_id = config.rp_id
await kanta.open()
with kanta.transaction("test:write_config"):
kanta.data.config = config
await kanta.close()
def stored_config(tmp_path: Path) -> Config:
"""Read back the stored combined configuration."""
return _load_stored_config(tmp_path / "paskia.kantadb")
def write_config(db_path: Path, config: Config) -> None:
"""Synchronous wrapper for _write_config."""
asyncio.run(_write_config(db_path, config))
def write_legacy_db(root: Path, config: legacy.LegacyConfig) -> Path:
"""Create a legacy-format database directory <rp-id>.paskiadb/main.db."""
src_dir = root / f"{config.rp_id}.paskiadb"
src_dir.mkdir()
db_file = src_dir / "main.db"
async def _write() -> None:
kanta = Kanta(str(db_file), legacy.LegacyDB())
await kanta.open()
with kanta.transaction("test:seed"):
kanta.data.config = config
await kanta.close()
asyncio.run(_write())
return src_dir
def test_cli_defaults(cli_run):
with tempfile.TemporaryDirectory() as tmp:
runtime = cli_run("--rp-id", "localhost", db_root=tmp)
def test_init_defaults(run_cli, tmp_path):
run_cli("init")
assert runtime.config.rp_id == "localhost"
assert runtime.config.rp_name is None
assert runtime.config.auth_host is None
assert runtime.config.origins is None
assert runtime.site_url == "http://localhost:4401"
assert runtime.site_path == "/auth/"
assert runtime.save is False
config = stored_config(tmp_path)
assert list(config.domains) == ["localhost"]
assert config.domains["localhost"].rp_name is None
assert config.domains["localhost"].origins == {"**.localhost": True}
assert config.listen is None
def test_cli_explicit_options(cli_run):
runtime = cli_run(
"--rp-id",
"example.com",
"--rp-name",
"Example Corp",
"--auth-host",
"auth.example.com",
"--origin",
"https://app.example.com",
)
def test_init_full_options(run_cli, tmp_path):
run_cli("init", "example.com", "Example Corp", "--listen", "4402")
assert runtime.config.rp_id == "example.com"
assert runtime.config.rp_name == "Example Corp"
assert runtime.config.auth_host == "https://auth.example.com"
assert runtime.config.origins == [
"https://auth.example.com",
"https://app.example.com",
]
assert runtime.site_url == "https://auth.example.com"
assert runtime.site_path == "/"
config = stored_config(tmp_path)
domain = config.domains["example.com"]
assert domain.rp_name == "Example Corp"
assert domain.origins == {"**.example.com": True}
assert config.listen == ["4402"]
def test_cli_loads_stored_config(cli_run):
with tempfile.TemporaryDirectory() as tmp:
db_path = Path(tmp) / "main.db"
write_config(
db_path,
Config(
rp_id="example.com",
rp_name="Stored Name",
origins=["https://stored.example.com"],
),
)
runtime = cli_run("--rp-id", "example.com", db_root=tmp)
def test_init_adds_domains_to_existing_database(run_cli, tmp_path):
"""Further rp-ids are added by repeating init; no comma separation."""
run_cli("init", "company.com")
run_cli("init", "app.com")
run_cli("init", "pro.com", "Pro Corp")
assert runtime.config.rp_name == "Stored Name"
assert runtime.config.origins == ["https://stored.example.com"]
assert runtime.site_url == "https://stored.example.com"
config = stored_config(tmp_path)
assert list(config.domains) == ["company.com", "app.com", "pro.com"]
assert config.domains["pro.com"].rp_name == "Pro Corp"
def test_cli_overrides_stored_config(cli_run):
with tempfile.TemporaryDirectory() as tmp:
db_path = Path(tmp) / "main.db"
write_config(db_path, Config(rp_id="example.com", rp_name="Stored Name"))
runtime = cli_run(
"--rp-id", "example.com", "--rp-name", "Overridden", db_root=tmp
)
assert runtime.config.rp_name == "Overridden"
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 test_cli_save_flag(cli_run):
runtime = cli_run("--save")
assert runtime.save is True
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 kanta.data.oidc.key
finally:
await kanta.close()
return asyncio.run(_read())
def test_cli_invalid_auth_host(cli_run):
def test_init_updates_rp_name_of_existing_domain(run_cli, tmp_path):
run_cli("init", "example.com", "Old Name")
run_cli("init", "example.com", "New Name")
assert stored_config(tmp_path).domains["example.com"].rp_name == "New Name"
def test_init_noop_on_existing_domain(run_cli):
run_cli("init")
with pytest.raises(SystemExit, match="already configured"):
run_cli("init")
def test_init_refuses_legacy_database(run_cli, tmp_path):
write_legacy_db(tmp_path, legacy.LegacyConfig(rp_id="example.com"))
with pytest.raises(SystemExit):
cli_run("--rp-id", "example.com", "--auth-host", "notsub.example.org")
run_cli("init")
def test_init_rejects_removed_options(run_cli):
"""Origins and auth hosts are admin-interface configuration, not init's."""
with pytest.raises(SystemExit):
run_cli("init", "example.com", "--auth-host", "auth.example.com")
with pytest.raises(SystemExit):
run_cli("init", "--origin", "https://app.example.com")
def test_serve_requires_database(run_cli):
with pytest.raises(SystemExit, match="paskia init"):
run_cli()
def test_serve_uses_stored_config(run_cli, tmp_path):
run_cli("init", "example.com", "Stored Name")
calls = run_cli()
assert calls["app"] == "paskia.fastapi.mainapp:app"
assert calls["listen"] is None # stored listen (None) used
serve = msgspec.json.decode(os.environ["PASKIA_CONFIG"].encode(), type=ServeConfig)
assert serve.listen is None
def test_serve_listen_override_not_persisted(run_cli, tmp_path):
run_cli("init", "--listen", "4402")
calls = run_cli("--listen", "4403")
assert calls["listen"] == ["4403"]
serve = msgspec.json.decode(os.environ["PASKIA_CONFIG"].encode(), type=ServeConfig)
assert serve.listen == ["4403"]
# Stored config keeps the original listen value
assert stored_config(tmp_path).listen == ["4402"]
def test_serve_suggests_migrate_when_legacy_present(run_cli, tmp_path):
write_legacy_db(tmp_path, legacy.LegacyConfig(rp_id="example.com"))
with pytest.raises(SystemExit, match="paskia migrate"):
run_cli()
def test_migrate_converts_legacy_database(run_cli, tmp_path):
src_dir = write_legacy_db(
tmp_path, legacy.LegacyConfig(rp_id="example.com", rp_name="Legacy Name")
)
# Persisted user files move to the new data root
avatar = src_dir / "users" / "019c6831-84cf-7b88-b66c-c8165890b7c5"
avatar.mkdir(parents=True)
(avatar / "profile.webp").write_bytes(b"RIFF1234WEBP")
run_cli("migrate")
config = stored_config(tmp_path)
assert list(config.domains) == ["example.com"]
assert config.domains["example.com"].rp_name == "Legacy Name"
# Legacy directory renamed aside, user files moved over
assert not src_dir.exists()
assert (tmp_path / "example.com.paskiadb.converted-bak").is_dir()
assert (
tmp_path
/ "paskia.data"
/ "users"
/ "019c6831-84cf-7b88-b66c-c8165890b7c5"
/ "profile.webp"
).read_bytes() == b"RIFF1234WEBP"
def test_migrate_multiple_legacy_databases_require_rp_id(run_cli, tmp_path):
write_legacy_db(tmp_path, legacy.LegacyConfig(rp_id="one.com"))
write_legacy_db(tmp_path, legacy.LegacyConfig(rp_id="two.com"))
with pytest.raises(SystemExit, match="paskia migrate"):
run_cli("migrate")
def test_migrate_explicit_rp_id_selects_candidate(run_cli, tmp_path):
write_legacy_db(tmp_path, legacy.LegacyConfig(rp_id="one.com"))
write_legacy_db(tmp_path, legacy.LegacyConfig(rp_id="two.com"))
run_cli("migrate", "two.com")
config = stored_config(tmp_path)
assert list(config.domains) == ["two.com"]
# The other candidate is left in place
assert (tmp_path / "one.com.paskiadb").is_dir()
assert (tmp_path / "two.com.paskiadb.converted-bak").is_dir()
def test_migrate_unknown_rp_id(run_cli, tmp_path):
write_legacy_db(tmp_path, legacy.LegacyConfig(rp_id="one.com"))
with pytest.raises(SystemExit, match="nope.com.paskiadb"):
run_cli("migrate", "nope.com")
def test_migrate_refuses_existing_database(run_cli):
run_cli("init")
with pytest.raises(SystemExit, match="already exists"):
run_cli("migrate")
def test_migrate_without_legacy_database(run_cli):
with pytest.raises(SystemExit, match="No legacy"):
run_cli("migrate")
def test_cli_help():
@@ -152,3 +262,25 @@ def test_cli_help():
)
assert result.returncode == 0
assert "Paskia authentication server" in result.stdout
def test_cli_init_help():
result = subprocess.run(
[sys.executable, "-m", "paskia", "init", "--help"],
capture_output=True,
text=True,
check=False,
)
assert result.returncode == 0
assert "Bootstrap" in result.stdout
def test_cli_migrate_help():
result = subprocess.run(
[sys.executable, "-m", "paskia", "migrate", "--help"],
capture_output=True,
text=True,
check=False,
)
assert result.returncode == 0
assert "Convert" in result.stdout
tests/test_domains.py
+995
View File
@@ -0,0 +1,995 @@
"""Tests for the multi-domain machinery: registry resolution, config
validation, ASGI dispatch, domain binding of auth codes, legacy database
conversion, log censoring and bootstrap caveats.
"""
from __future__ import annotations
import asyncio
import os
from datetime import UTC, datetime
from uuid import UUID
import httpx
import pytest
from kanta import Kanta
from paskia import authcode, domains
from paskia.bootstrap import check_admin_credentials
from paskia.db import create_credential
from paskia.db.legacy import (
LegacyConfig,
LegacyCredential,
LegacyDB,
LegacySession,
convert_legacy_database,
)
from paskia.db.lifecycle import format_log_uuid
from paskia.db.operations import DB
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
# -------------------------------------------------------------------------
def build_registry(configs: dict[str, DomainConfig]) -> domains.DomainRegistry:
"""Build and install a registry from domain configs (listen unset)."""
domains.configure(listen=None)
return domains.init_registry(Config(domains=configs))
ROR_CONFIG = Config(
domains={
"company.com": DomainConfig(
rp_name="Company",
origins={
"auth.company.com": OriginEntry(auth_host=True),
"app.com": True, # related origin (outside the rp-id domain)
},
),
"pro.com": DomainConfig(rp_name="Pro", origins={"**.pro.com": True}),
}
)
class StubApp:
"""ASGI app recording the scope it was called with."""
def __init__(self):
self.scope = None
async def __call__(self, scope, receive, send):
self.scope = scope
async def drive_ws(middleware: DispatchMiddleware, headers: list[tuple[bytes, bytes]]):
"""Run a websocket scope through the middleware, capturing sent messages."""
async def receive():
return {"type": "websocket.connect"}
sent = []
async def send(message):
sent.append(message)
stub = middleware.app
await middleware(
{"type": "websocket", "headers": headers, "path": "/"}, receive, send
)
return stub, sent
async def drive_http(
middleware: DispatchMiddleware, headers: list[tuple[bytes, bytes]]
):
"""Run an http scope through the middleware, capturing sent messages."""
async def receive():
return {"type": "http.request", "body": b""}
sent = []
async def send(message):
sent.append(message)
stub = middleware.app
await middleware(
{
"type": "http",
"headers": headers,
"method": "GET",
"path": "/",
"query_string": b"",
},
receive,
send,
)
return stub, sent
# -------------------------------------------------------------------------
# Host resolution
# -------------------------------------------------------------------------
class TestResolve:
def test_exact_rp_id(self):
reg = build_registry(ROR_CONFIG.domains)
assert reg.resolve("pro.com").rp_id == "pro.com"
assert reg.resolve("company.com").rp_id == "company.com"
def test_auth_host_and_related_origin(self):
reg = build_registry(ROR_CONFIG.domains)
assert reg.resolve("auth.company.com").rp_id == "company.com"
assert reg.resolve("app.com").rp_id == "company.com"
def test_subdomain_suffix_longest_match(self):
reg = build_registry(
{"example.com": DomainConfig(), "sub.example.com": DomainConfig()}
)
assert reg.resolve("www.example.com").rp_id == "example.com"
assert reg.resolve("api.sub.example.com").rp_id == "sub.example.com"
def test_port_and_trailing_dot_normalized(self):
reg = build_registry(ROR_CONFIG.domains)
assert reg.resolve("pro.com:8443").rp_id == "pro.com"
assert reg.resolve("app.com.").rp_id == "company.com"
def test_unknown_host(self):
reg = build_registry(ROR_CONFIG.domains)
assert reg.resolve("evil.com") is None
assert reg.resolve("") is None
assert reg.resolve(None) is None
def test_auth_host_is_per_domain_no_fallback(self):
reg = build_registry(ROR_CONFIG.domains)
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"
# -------------------------------------------------------------------------
# Cross-domain configuration validation
# -------------------------------------------------------------------------
class TestValidateConfig:
def test_valid(self):
domains.validate_config(ROR_CONFIG)
def test_empty_origins_table_is_valid(self):
"""No origins at all: nothing of the domain is allowed, but the
configuration itself is legal (e.g. a related-only domain)."""
domains.validate_config(Config(domains={"a.com": DomainConfig()}))
domains.validate_config(
Config(domains={"a.com": DomainConfig(origins={"b.com": True})})
)
def test_related_origin_cap(self):
domains.validate_config(
Config(
domains={
"company.com": DomainConfig(
origins={f"app{i}.com": True for i in range(5)}
)
}
)
)
with pytest.raises(ValueError, match="related origins"):
domains.validate_config(
Config(
domains={
"company.com": DomainConfig(
origins={f"app{i}.com": True for i in range(6)}
)
}
)
)
def test_star_origin_rejected(self):
"""Plain '*' suggests 'anything goes' — the wildcard must be
explicit and under the rp-id."""
with pytest.raises(ValueError, match="not allowed"):
domains.validate_config(
Config(domains={"a.com": DomainConfig(origins={"*": True})})
)
def test_malformed_origin_hostname_rejected(self):
"""No empty hostname labels — leading, trailing and double dots
are invalid, in concrete entries and wildcard bases alike."""
for key in (".a.com", "a..com", "a.com.", "http://.a.com:8080"):
with pytest.raises(ValueError, match="Invalid origin"):
domains.validate_config(
Config(domains={"a.com": DomainConfig(origins={key: True})})
)
for key in ("*..a.com", "**..a.com"):
with pytest.raises(ValueError, match="Invalid wildcard origin"):
domains.validate_config(
Config(domains={"a.com": DomainConfig(origins={key: True})})
)
def test_subdomain_entry_is_in_domain(self):
"""An entry within the rp-id domain is an ordinary in-domain
sign-in site, never a related origin."""
config = Config(domains={"a.com": DomainConfig(origins={"app.a.com": True})})
domains.validate_config(config)
reg = build_registry(config.domains)
assert reg.get("a.com").related_origins == []
def test_wildcard_outside_rp_id_rejected(self):
"""Related origins are individual hosts; wildcards must stay within
the rp-id domain. Both wildcard forms are accepted in-domain."""
domains.validate_config(
Config(domains={"a.com": DomainConfig(origins={"*.a.com": True})})
)
domains.validate_config(
Config(domains={"a.com": DomainConfig(origins={"**.a.com": True})})
)
with pytest.raises(ValueError, match="wildcard outside the rp-id"):
domains.validate_config(
Config(domains={"a.com": DomainConfig(origins={"**.b.com": True})})
)
def test_wildcard_auth_host_rejected(self):
with pytest.raises(ValueError, match="cannot be the auth host"):
domains.validate_config(
Config(
domains={
"a.com": DomainConfig(
origins={"*.a.com": OriginEntry(auth_host=True)}
)
}
)
)
def test_related_auth_host_rejected(self):
"""The auth host is always in-domain; a related origin cannot
carry the mark."""
with pytest.raises(ValueError, match="cannot be the auth host"):
domains.validate_config(
Config(
domains={
"a.com": DomainConfig(
origins={"auth.b.com": OriginEntry(auth_host=True)}
)
}
)
)
def test_several_auth_hosts_rejected(self):
with pytest.raises(ValueError, match="several origins as the auth host"):
domains.validate_config(
Config(
domains={
"a.com": DomainConfig(
origins={
"auth.a.com": OriginEntry(auth_host=True),
"login.a.com": OriginEntry(auth_host=True),
}
)
}
)
)
def test_auth_host_collision(self):
with pytest.raises(ValueError, match="collides with a related origin"):
domains.validate_config(
Config(
domains={
"a.com": DomainConfig(
origins={"auth.a.com": OriginEntry(auth_host=True)}
),
"b.com": DomainConfig(origins={"auth.a.com": True}),
}
)
)
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(origins={"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(origins={"app.b.com": True}),
"c.com": DomainConfig(origins={"app.b.com": True}),
"b.com": DomainConfig(),
}
)
)
with pytest.raises(ValueError, match="configured for both"):
domains.validate_config(
Config(
domains={
"a.com": DomainConfig(origins={"shared.com": True}),
"c.com": DomainConfig(origins={"shared.com": True}),
}
)
)
def test_auth_host_must_not_collide_with_rp_id(self):
with pytest.raises(ValueError, match="collides with an rp-id"):
domains.validate_config(
Config(
domains={
"a.com": DomainConfig(
origins={"b.a.com": OriginEntry(auth_host=True)}
),
"b.a.com": DomainConfig(),
}
)
)
# -------------------------------------------------------------------------
# Best-effort serving: stored config sanitization
# -------------------------------------------------------------------------
class TestSanitizeConfig:
"""Serving never fails on stored config problems; it degrades + warns."""
def test_cross_domain_origin_stays_as_related(self):
"""An out-of-domain entry simply IS a related origin — no repair
needed, no warning."""
config, warnings = domains.sanitize_config(
Config(domains={"localhost": DomainConfig(origins={"example.com": True})})
)
assert config.domains["localhost"].origins == {"example.com": True}
assert not warnings
domains.validate_config(config) # sanitized config is strict-clean
def test_star_origin_rewritten_explicit(self):
"""Branch-era '*' shorthand is rewritten to '**.{rp-id}'; an auth
mark on it is cleared."""
config, warnings = domains.sanitize_config(
Config(domains={"a.com": DomainConfig(origins={"*": True})})
)
assert config.domains["a.com"].origins == {"**.a.com": True}
assert any("**." in w for w in warnings)
domains.validate_config(config)
config, warnings = domains.sanitize_config(
Config(
domains={
"a.com": DomainConfig(origins={"*": OriginEntry(auth_host=True)})
}
)
)
assert config.domains["a.com"].origins == {"**.a.com": True}
assert any("mark cleared" in w for w in warnings)
domains.validate_config(config)
def test_malformed_origin_dropped(self):
config, warnings = domains.sanitize_config(
Config(domains={"a.com": DomainConfig(origins={"https://": True})})
)
assert config.domains["a.com"].origins == {}
assert warnings
def test_malformed_hostname_dropped(self):
"""Empty hostname labels (leading/trailing/double dots) are dropped,
from concrete entries and wildcard bases alike."""
config, warnings = domains.sanitize_config(
Config(
domains={
"a.com": DomainConfig(
origins={
".a.com": True,
"a.com.": True,
"**.a..com": True,
"ok.a.com": True,
}
)
}
)
)
assert list(config.domains["a.com"].origins) == ["ok.a.com"]
assert len(warnings) == 3
domains.validate_config(config) # sanitized config is strict-clean
def test_invalid_rp_id_domain_dropped(self):
config, warnings = domains.sanitize_config(
Config(domains={"not a domain!": DomainConfig(), "ok.com": DomainConfig()})
)
assert list(config.domains) == ["ok.com"]
assert any("dropped" in w for w in warnings)
def test_wildcard_outside_rp_id_dropped(self):
config, warnings = domains.sanitize_config(
Config(domains={"a.com": DomainConfig(origins={"*.b.com": True})})
)
assert config.domains["a.com"].origins == {}
assert any("wildcard" in w for w in warnings)
domains.validate_config(config) # sanitized config is strict-clean
def test_cap_exceeded_truncated(self):
config, warnings = domains.sanitize_config(
Config(
domains={
"a.com": DomainConfig(
origins={f"app{i}.com": True for i in range(6)}
)
}
)
)
assert len(config.domains["a.com"].origins) == 5
assert any("maximum" in w for w in warnings)
domains.validate_config(config)
def test_related_auth_host_mark_cleared(self):
"""An auth mark on a related (out-of-domain) origin is cleared."""
config, warnings = domains.sanitize_config(
Config(
domains={
"a.com": DomainConfig(
origins={"auth.b.com": OriginEntry(auth_host=True)}
)
}
)
)
assert config.domains["a.com"].origins == {"auth.b.com": True}
assert any("cannot be the auth host" in w for w in warnings)
domains.validate_config(config)
def test_auth_host_colliding_with_rp_id_cleared(self):
config, warnings = domains.sanitize_config(
Config(
domains={
"a.com": DomainConfig(
origins={"auth.a.com": OriginEntry(auth_host=True)}
),
"auth.a.com": DomainConfig(),
}
)
)
assert config.domains["a.com"].origins == {"auth.a.com": True}
assert any("collides" in w for w in warnings)
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={
"a.com": DomainConfig(origins={"app.b.com": True}),
"b.com": DomainConfig(),
}
)
)
assert config.domains["a.com"].origins == {"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(origins={"shared.com": True}),
"b.com": DomainConfig(origins={"shared.com": True}),
}
)
)
assert config.domains["a.com"].origins == {"shared.com": True}
assert config.domains["b.com"].origins == {}
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"):
domains.sanitize_config(Config(domains={}))
with pytest.raises(ValueError, match="No servable domain"):
domains.sanitize_config(Config(domains={"not a domain!": DomainConfig()}))
def test_build_serves_related_origin(self):
reg = build_registry({"localhost": DomainConfig(origins={"example.com": True})})
domain = reg.get("localhost")
assert domain.related_origins == ["https://example.com"]
domain.passkey.validate_origin("https://example.com")
# -------------------------------------------------------------------------
# Origin validation semantics (Passkey)
# -------------------------------------------------------------------------
class TestOriginValidation:
"""The allow-list is explicit: empty allows nothing, wildcards cover
subtrees, related origins are additive exact matches."""
def test_empty_allow_list_denies_all(self):
p = Passkey(rp_id="example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://app.example.com")
def test_allow_list_restricts_subtree(self):
p = Passkey(rp_id="example.com", origins=["https://app.example.com"])
assert p.validate_origin("https://app.example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://www.example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://example.com")
def test_related_origins_are_additive(self):
p = Passkey(rp_id="example.com", related_origins=["https://app2.com"])
assert p.validate_origin("https://app2.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://app.example.com") # nothing in-domain listed
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://other.com")
def test_double_star_matches_apex_and_any_depth(self):
"""'**.example.com' covers the apex and subdomains at any depth
(the shell-glob convention)."""
p = Passkey(rp_id="example.com", origins=["**.example.com"])
assert p.validate_origin("https://example.com") # apex
assert p.validate_origin("https://app.example.com") # one level
assert p.validate_origin("https://a.b.c.example.com") # any depth
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://anotherexample.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://other.com")
def test_single_star_matches_exactly_one_level(self):
"""'*.example.com' covers exactly one subdomain level — neither the
apex nor deeper levels."""
p = Passkey(rp_id="example.com", origins=["*.example.com"])
assert p.validate_origin("https://app.example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://example.com") # apex excluded
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://a.b.example.com") # too deep
def test_wildcard_is_https_only(self):
"""A '**.example.com' entry does not fall back to other schemes."""
p = Passkey(rp_id="example.com", origins=["**.example.com"])
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("http://example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("http://app.example.com:8080")
def test_star_entry_rejected(self):
with pytest.raises(ValueError, match="Invalid origin"):
Passkey(rp_id="example.com", origins=["*"])
def test_malformed_hostname_rejected(self):
"""Leading/trailing/double dots are invalid in any entry form."""
with pytest.raises(ValueError, match="malformed hostname"):
Passkey(rp_id="example.com", origins=["https://.example.com"])
with pytest.raises(ValueError, match="malformed hostname"):
Passkey(rp_id="example.com", origins=["**.a..example.com"])
with pytest.raises(ValueError, match="malformed hostname"):
Passkey(rp_id="example.com", related_origins=["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=["**.localhost"])
assert p.validate_origin("http://localhost:8080")
assert p.validate_origin("http://app.localhost:3000")
assert p.validate_origin("http://a.b.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")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://localhost:4403")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("http://localhost:4404")
def test_sub_wildcard_matches_only_its_subtree(self):
p = Passkey(rp_id="example.com", origins=["**.app.example.com"])
assert p.validate_origin("https://app.example.com")
assert p.validate_origin("https://www.app.example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://example.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://other.example.com")
def test_wildcard_related_origin_rejected(self):
with pytest.raises(ValueError, match="wildcard"):
Passkey(rp_id="example.com", related_origins=["*.other.com"])
def test_related_origins_combined_with_allow_list(self):
p = Passkey(
rp_id="example.com",
origins=["https://app.example.com"],
related_origins=["https://app2.com"],
)
assert p.validate_origin("https://app.example.com")
assert p.validate_origin("https://app2.com")
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://www.example.com")
def test_constructor_rejects_mixed_up_fields(self):
with pytest.raises(ValueError, match="related origin"):
Passkey(rp_id="example.com", origins=["https://app2.com"])
with pytest.raises(ValueError, match="within the rp-id domain"):
Passkey(rp_id="example.com", related_origins=["https://app.example.com"])
def test_domain_wires_both_lists(self):
reg = build_registry(ROR_CONFIG.domains)
p = reg.get("company.com").passkey
assert p.validate_origin("https://app.com") # related origin
assert p.validate_origin("https://auth.company.com") # allow-listed
with pytest.raises(ValueError, match="not allowed"):
p.validate_origin("https://www.company.com") # not allow-listed
# -------------------------------------------------------------------------
# ASGI dispatch
# -------------------------------------------------------------------------
class TestDispatchMiddleware:
@pytest.mark.asyncio
async def test_http_unknown_host_421(self):
build_registry(ROR_CONFIG.domains)
stub, sent = await drive_http(
DispatchMiddleware(StubApp()), [(b"host", b"evil.com")]
)
assert stub.scope is None # Inner app not called
assert sent[0]["type"] == "http.response.start"
assert sent[0]["status"] == 421
@pytest.mark.asyncio
async def test_http_dispatches_domain(self):
build_registry(ROR_CONFIG.domains)
stub, _sent = await drive_http(
DispatchMiddleware(StubApp()), [(b"host", b"app.com.")]
)
assert stub.scope is not None
assert stub.scope["state"]["domain"].rp_id == "company.com"
@pytest.mark.asyncio
async def test_http_current_domain_set_inside_request(self):
build_registry(ROR_CONFIG.domains)
seen = {}
async def app(scope, receive, send):
seen["domain"] = domains.current_domain()
await drive_http(DispatchMiddleware(app), [(b"host", b"pro.com")])
assert seen["domain"].rp_id == "pro.com"
# Contextvar is reset after the request; with several domains there
# is no implicit current domain outside a request context.
with pytest.raises(RuntimeError, match="request context"):
domains.current_domain()
@pytest.mark.asyncio
async def test_ws_unknown_host_closed(self):
build_registry(ROR_CONFIG.domains)
stub, sent = await drive_ws(
DispatchMiddleware(StubApp()), [(b"host", b"evil.com")]
)
assert stub.scope is None
assert sent == [{"type": "websocket.close", "code": 1008}]
@pytest.mark.asyncio
async def test_ws_same_domain_origin(self):
build_registry(ROR_CONFIG.domains)
stub, sent = await drive_ws(
DispatchMiddleware(StubApp()),
[(b"host", b"auth.company.com"), (b"origin", b"https://app.com")],
)
assert sent == []
assert stub.scope["state"]["domain"].rp_id == "company.com"
@pytest.mark.asyncio
async def test_ws_cross_domain_requires_origin_own_auth_host(self):
build_registry(ROR_CONFIG.domains)
# 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 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(
DispatchMiddleware(StubApp()),
[(b"host", b"company.com"), (b"origin", b"https://pro.com")],
)
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)
# Missing origin
stub, _ = await drive_ws(DispatchMiddleware(StubApp()), [(b"host", b"pro.com")])
assert stub.scope["state"]["domain"].rp_id == "pro.com"
# Unknown origin: host domain applies (endpoint-side validation decides)
stub, _ = await drive_ws(
DispatchMiddleware(StubApp()),
[(b"host", b"pro.com"), (b"origin", b"https://evil.com")],
)
assert stub.scope["state"]["domain"].rp_id == "pro.com"
# -------------------------------------------------------------------------
# Domain binding of auth codes
# -------------------------------------------------------------------------
class TestAuthCodeDomainBinding:
@pytest.mark.asyncio
async def test_cookie_code_rejected_on_other_domain(
self, client: httpx.AsyncClient, session_token: str
):
code = authcode.store_cookie(
authcode.CookieCode(
session_key=session_token,
created=datetime.now(UTC),
rp_id="other.com",
)
)
response = await client.post(
"/auth/api/set-session",
headers={
"Authorization": f"Bearer {code}",
"Host": "localhost:4401",
},
)
assert response.status_code == 401
@pytest.mark.asyncio
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("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=token,
created=datetime.now(UTC),
redirect_uri="https://client.example/callback",
scope="openid",
)
)
response = await client.post(
"/auth/oidc/token",
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": "https://client.example/callback",
"client_id": str(oidc_client.uuid),
"client_secret": secret,
},
headers={"Host": "localhost:4401"},
)
assert response.status_code == 200
assert response.json()["access_token"]
# -------------------------------------------------------------------------
# Legacy database conversion
# -------------------------------------------------------------------------
def _read_db(path) -> DB:
async def _read() -> DB:
new_db = DB()
kanta = Kanta(str(path), new_db)
await kanta.open(readonly=True)
return kanta.data
return asyncio.run(_read())
async def _write_legacy(src_file, config: LegacyConfig) -> None:
kanta = Kanta(str(src_file), LegacyDB())
await kanta.open()
with kanta.transaction("test:seed"):
kanta.data.config = config
await kanta.close()
class TestLegacyConversion:
def test_convert_stamps_domain_everywhere(self, tmp_path):
src = tmp_path / "example.com.paskiadb"
src.mkdir()
src_file = src / "main.db"
cred_uuid = UUID("019c6831-84cf-7b88-b66c-c8165890b7c5")
user_uuid = UUID("019c6831-84cf-7b88-b66c-c8165890b7c6")
async def _write() -> None:
kanta = Kanta(str(src_file), LegacyDB())
await kanta.open()
with kanta.transaction("test:seed"):
kanta.data.config = LegacyConfig(
rp_id="example.com",
rp_name="Example",
origins=["https://app.example.com", "*.example.com"],
)
kanta.data.credentials[cred_uuid] = LegacyCredential(
credential_id=b"credential-id",
user_uuid=user_uuid,
aaguid=UUID(int=0),
public_key=b"public-key",
sign_count=3,
created_at=datetime.now(UTC),
)
kanta.data.sessions["session-key"] = LegacySession(
user_uuid=user_uuid,
credential_uuid=cred_uuid,
host="example.com",
ip="127.0.0.1",
user_agent="pytest",
validated=datetime.now(UTC),
)
kanta.data.oidc = OIDC(key=b"legacy-signing-key")
await kanta.close()
asyncio.run(_write())
config = convert_legacy_database(src_file, tmp_path / "paskia.kantadb")
domain = config.domains["example.com"]
assert domain.rp_name == "Example"
# 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"
# The legacy OIDC provider carries over as the instance-global one
assert converted.oidc.key == b"legacy-signing-key"
def test_convert_empty_origins_seeds_wildcard(self, tmp_path):
"""Legacy 'no origins' meant the whole rp-id domain; the new format
makes that explicit as '**.{rp-id}'."""
src_file = tmp_path / "main.db"
asyncio.run(
_write_legacy(src_file, LegacyConfig(rp_id="example.com", rp_name="Ex"))
)
config = convert_legacy_database(src_file, tmp_path / "paskia.kantadb")
assert config.domains["example.com"].origins == {"**.example.com": True}
# -------------------------------------------------------------------------
# Transaction log censoring
# -------------------------------------------------------------------------
class TestLogCensoring:
def test_oidc_key_values_hidden(self):
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.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.clients") is None
assert format_log_uuid("not-a-uuid", "config.domains") is None
# -------------------------------------------------------------------------
# Bootstrap caveat: admin credential is checked on the configured domains
# -------------------------------------------------------------------------
class TestBootstrapCaveat:
@pytest.mark.asyncio
async def test_admin_without_credentials_gets_link(
self, test_db: DB, domain_registry
):
assert await check_admin_credentials() is True
@pytest.mark.asyncio
async def test_admin_with_domain_credential_ok(
self, test_db: DB, domain_registry, test_user, test_credential
):
assert await check_admin_credentials() is False
@pytest.mark.asyncio
async def test_admin_with_only_unconfigured_domain_credential_gets_link(
self, test_db: DB, domain_registry, test_user
):
"""A passkey under an rp-id outside the config does not satisfy the check."""
cred = Credential.create(
credential_id=os.urandom(32),
user=test_user.uuid,
aaguid=UUID(int=0),
public_key=os.urandom(64),
sign_count=0,
rp_id="example.com",
)
create_credential(cred)
assert await check_admin_credentials() is True
tests/test_user.py
-31
View File
@@ -15,7 +15,6 @@ from urllib.parse import urlsplit
import httpx
import pytest
from paskia.db.paths import db_file_path, users_root_path
from tests.conftest import auth_headers, create_test_image_bytes
@@ -93,8 +92,6 @@ class TestUserAvatar:
monkeypatch,
):
"""Uploading a WebP avatar should store and expose the canonical URL."""
monkeypatch.setenv("PASKIA_DB", str(tmp_path / "test-avatar-db.paskiadb"))
upload_bytes = create_test_image_bytes()
response = await client.put(
@@ -138,8 +135,6 @@ class TestUserAvatar:
monkeypatch,
):
"""Avatar uploads must already be browser-prepared WebP."""
monkeypatch.setenv("PASKIA_DB", str(tmp_path / "test-avatar-db.paskiadb"))
response = await client.put(
f"/auth/api/user/{test_user.uuid}/profile.webp",
files={
@@ -165,8 +160,6 @@ class TestUserAvatar:
monkeypatch,
):
"""Deleting avatar should clear the user avatar URL."""
monkeypatch.setenv("PASKIA_DB", str(tmp_path / "test-avatar-db.paskiadb"))
await client.put(
f"/auth/api/user/{test_user.uuid}/profile.webp",
files={"file": ("avatar.webp", create_test_image_bytes(), "image/webp")},
@@ -203,30 +196,6 @@ class TestUserAvatar:
assert response.status_code == 403
def test_paskia_db_legacy_file_is_migrated_to_root_dir(tmp_path, monkeypatch):
legacy_path = tmp_path / "legacy.paskiadb"
legacy_bytes = b'{"v":0}\n'
legacy_path.write_bytes(legacy_bytes)
monkeypatch.setenv("PASKIA_DB", str(legacy_path))
db_path = db_file_path(create_root=True)
assert legacy_path.is_dir()
assert db_path == legacy_path / "main.db"
assert db_path.read_bytes() == legacy_bytes
def test_paskia_db_root_uses_users_directory(tmp_path, monkeypatch):
root_path = tmp_path / "instance-root"
monkeypatch.setenv("PASKIA_DB", str(root_path))
users_path = users_root_path(create_root=True)
assert users_path == root_path / "users"
assert users_path.parent == root_path
class TestUserLogoutAll:
"""Tests for POST /auth/api/user/logout-all"""