LeoVasanko
52f3b26630
'**.example.com' covers the apex and subdomains at any depth;
'*.example.com' covers exactly one subdomain level (neither apex nor
deeper) — analogous to permission scope wildcards, and sidestepping the
DNS/TLS/nginx ambiguity around '*.'. This also allows excluding the apex
where wanted. The seeded/default entry becomes '**.{rp-id}' (init,
add-domain, legacy empty-origins conversion, branch-era '*' sanitize
rewrite).
331 lines
9.9 KiB
Python
331 lines
9.9 KiB
Python
"""
|
|
Pytest configuration and fixtures for Paskia API tests.
|
|
|
|
FastAPI provides excellent testing support through httpx.ASGITransport,
|
|
which allows us to make async requests directly to the ASGI app without
|
|
running a server.
|
|
|
|
Since we can't emulate WebAuthn passkeys, we create sessions directly
|
|
in the database to test authenticated endpoints.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
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
|
|
import pytest
|
|
import pytest_asyncio
|
|
from kanta import Kanta
|
|
|
|
import paskia.db.operations as ops_db
|
|
from paskia import domains
|
|
from paskia.authsession import reset_expires
|
|
from paskia.config import SESSION_LIFETIME
|
|
from paskia.db import (
|
|
Credential,
|
|
Org,
|
|
Permission,
|
|
Role,
|
|
User,
|
|
create_credential,
|
|
create_reset_token,
|
|
create_role,
|
|
create_user,
|
|
)
|
|
from paskia.db.bootstrap import bootstrap
|
|
from paskia.db.operations import DB
|
|
from paskia.db.structs import Config, DomainConfig, Session
|
|
from paskia.fastapi.mainapp import app
|
|
from paskia.fastapi.session import AUTH_COOKIE_NAME
|
|
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():
|
|
"""Create an event loop for the test session."""
|
|
loop = asyncio.get_event_loop_policy().new_event_loop()
|
|
yield 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.
|
|
|
|
Uses a kanta bootstrap callback to properly initialize the database with:
|
|
- 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)
|
|
|
|
# Register bootstrap callback so kanta seeds the empty DB during open()
|
|
@kanta.bootstrap(action="bootstrap")
|
|
def bootstrap_test_db(data: DB) -> None:
|
|
bootstrap(
|
|
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()
|
|
ops_db._db = db
|
|
ops_db._db._store = kanta
|
|
yield ops_db._db
|
|
await kanta.close()
|
|
ops_db._db = None
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
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")
|
|
async def admin_permission(test_db: DB) -> Permission:
|
|
"""Get the auth:admin permission created by bootstrap."""
|
|
return next(p for p in test_db.permissions.values() if p.scope == "auth:admin")
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def org_admin_permission(test_db: DB) -> Permission:
|
|
"""Get the auth:org:admin permission created by bootstrap."""
|
|
return next(p for p in test_db.permissions.values() if p.scope == "auth:org:admin")
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def test_org(test_db: DB) -> Org:
|
|
"""Get the test organization created by bootstrap."""
|
|
# Bootstrap creates exactly one org
|
|
return next(iter(test_db.orgs.values()))
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def test_role(test_db: DB) -> Role:
|
|
"""Get the Administration role created by bootstrap."""
|
|
# Bootstrap creates exactly one role (Administration)
|
|
return next(iter(test_db.roles.values()))
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def user_role(test_db: DB, test_org: Org) -> Role:
|
|
"""Create a test role without admin permission (regular user)."""
|
|
role = Role.create(
|
|
org=test_org.uuid,
|
|
display_name="User Role",
|
|
)
|
|
create_role(role)
|
|
return role
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def test_user(test_db: DB) -> User:
|
|
"""Get the admin user created by bootstrap."""
|
|
# Bootstrap creates exactly one user (admin)
|
|
return next(iter(test_db.users.values()))
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def regular_user(test_db: DB, user_role: Role) -> User:
|
|
"""Create a regular test user without admin permissions."""
|
|
user = User.create(
|
|
display_name="Regular User",
|
|
role=user_role.uuid,
|
|
)
|
|
create_user(user)
|
|
return user
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def test_credential(test_db: DB, test_user: User) -> Credential:
|
|
"""Create a test credential for the admin user."""
|
|
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=TEST_RP_ID,
|
|
)
|
|
create_credential(credential)
|
|
return credential
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def regular_credential(test_db: DB, regular_user: User) -> Credential:
|
|
"""Create a test credential for the regular user."""
|
|
credential = Credential.create(
|
|
credential_id=os.urandom(32),
|
|
user=regular_user.uuid,
|
|
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
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def session_token(
|
|
test_db: DB, test_user: User, test_credential: Credential
|
|
) -> str:
|
|
"""Create a session for the admin user and return the token."""
|
|
_db_key, secret = create_test_session(
|
|
user_uuid=test_user.uuid,
|
|
credential_uuid=test_credential.uuid,
|
|
)
|
|
return secret
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def regular_session_token(
|
|
test_db: DB, regular_user: User, regular_credential: Credential
|
|
) -> str:
|
|
"""Create a session for a regular user and return the token."""
|
|
_db_key, secret = create_test_session(
|
|
user_uuid=regular_user.uuid,
|
|
credential_uuid=regular_credential.uuid,
|
|
)
|
|
return secret
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def reset_token(test_db: DB, test_user: User, test_credential: Credential) -> str:
|
|
"""Create a reset token for the test user."""
|
|
return create_reset_token(
|
|
user_uuid=test_user.uuid,
|
|
expiry=reset_expires(),
|
|
token_type="reset",
|
|
)
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def client(
|
|
test_db: DB, domain_registry: domains.DomainRegistry
|
|
) -> AsyncGenerator[httpx.AsyncClient]:
|
|
"""Create an async test client for the FastAPI app."""
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(
|
|
transport=transport,
|
|
base_url="http://localhost:4401",
|
|
) as client:
|
|
yield client
|
|
|
|
|
|
def auth_headers(token: str) -> dict[str, str]:
|
|
"""Return headers with auth cookie set."""
|
|
return {"Cookie": f"{AUTH_COOKIE_NAME}={token}"}
|
|
|
|
|
|
def auth_cookie(token: str) -> httpx.Cookies:
|
|
"""Return cookies dict with auth cookie."""
|
|
cookies = httpx.Cookies()
|
|
cookies.set(AUTH_COOKIE_NAME, token, domain="localhost")
|
|
return cookies
|
|
|
|
|
|
def create_test_session(
|
|
user_uuid: UUID,
|
|
credential_uuid: UUID,
|
|
host: str = "localhost",
|
|
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.
|
|
|
|
- key: str used for session lookup (base64url encoded)
|
|
- token: stored in cookie/sent to client
|
|
"""
|
|
if duration is None:
|
|
duration = SESSION_LIFETIME
|
|
if user_uuid not in ops_db._db.users:
|
|
raise ValueError(f"User {user_uuid} not found")
|
|
if credential_uuid not in ops_db._db.credentials:
|
|
raise ValueError(f"Credential {credential_uuid} not found")
|
|
now = datetime.now(UTC)
|
|
|
|
# Generate token and derive key
|
|
token = secrets.token_urlsafe(12)
|
|
key = hash_secret("cookie", token)
|
|
|
|
session = Session.create(
|
|
user=user_uuid,
|
|
credential=credential_uuid,
|
|
key=key,
|
|
host=host,
|
|
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")
|
|
store = ops_db._db._store
|
|
if store is None:
|
|
raise RuntimeError("Test DB store is not initialized")
|
|
with store.transaction("create_test_session"):
|
|
session.store(now)
|
|
return session.key, token
|
|
|
|
|
|
def create_test_image_bytes(
|
|
*,
|
|
image_format: str = "WEBP",
|
|
) -> bytes:
|
|
"""Return deterministic test upload bytes without image-library dependencies."""
|
|
fixtures = {
|
|
"WEBP": (
|
|
b"RIFF\x1a\x00\x00\x00WEBPVP8 "
|
|
b"\x0e\x00\x00\x000\x01\x00\x9d\x01*\x01\x00\x01\x00\x01\x00"
|
|
),
|
|
"PNG": (
|
|
b"\x89PNG\r\n\x1a\n"
|
|
b"\x00\x00\x00\rIHDR"
|
|
b"\x00\x00\x00\x01\x00\x00\x00\x01\x08\x02\x00\x00\x00"
|
|
b"\x90wS\xde"
|
|
),
|
|
}
|
|
|
|
try:
|
|
return fixtures[image_format.upper()]
|
|
except KeyError as exc:
|
|
raise ValueError(f"Unsupported test image format: {image_format}") from exc
|