paskia/tests/test_remote.py

"""Tests for remote (satellite) domains: config, replica application, feed."""

import collections
import secrets
import time
from datetime import UTC, datetime
from uuid import UUID

import httpx
import msgspec
import pytest
import pytest_asyncio
from fastapi import Response

import paskia.db.operations as ops_db
from paskia import domains, satellite, syncfeed
from paskia.db.structs import (
    DB,
    Config,
    Credential,
    DomainConfig,
    Org,
    OriginEntry,
    Permission,
    RemoteConfig,
    Role,
    Session,
    User,
)
from paskia.fastapi.mainapp import app
from paskia.fastapi.session import AUTH_COOKIE_NAME
from paskia.util.crypto import hash_secret

from .conftest import TEST_RP_ID

REMOTE_URL = "http://remote.test"


def _remote_domain_config(**kw) -> Config:
    return Config(
        domains={
            TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True}),
            "example.com": DomainConfig(
                origins={
                    "**.example.com": True,
                    "auth.example.com": OriginEntry(auth_host=True),
                },
                remote=RemoteConfig(url=REMOTE_URL, token="t", **kw),
            ),
        }
    )


def test_remote_domain_valid():
    domains.validate_config(_remote_domain_config())


def test_remote_domain_requires_auth_host():
    config = _remote_domain_config()
    config.domains["example.com"].origins = {"**.example.com": True}
    with pytest.raises(ValueError, match="auth host"):
        domains.validate_config(config)


def test_remote_domain_requires_http_url():
    config = _remote_domain_config()
    config.domains["example.com"].remote.url = "ftp://x"
    with pytest.raises(ValueError, match="http"):
        domains.validate_config(config)


def test_sanitize_preserves_remote():
    config, warnings = domains.sanitize_config(_remote_domain_config())
    assert not warnings
    assert config.domains["example.com"].remote.url == REMOTE_URL


def test_apply_upsert_and_delete():
    replica = DB()
    user = User.create(display_name="U", role=UUID(int=1))
    user.uuid = UUID(int=2)
    satellite._apply(
        replica, "users", str(user.uuid), "upsert", _builtins(user)
    )
    assert replica.users[user.uuid].display_name == "U"
    satellite._apply(replica, "users", str(user.uuid), "delete", None)
    assert not replica.users


def _builtins(obj):
    return msgspec.to_builtins(obj)


def test_apply_session_roundtrip():
    """Sessions keep their string key and datetime/UUID fields."""
    replica = DB()
    session = Session.create(
        user=UUID(int=1),
        credential=UUID(int=2),
        key=hash_secret("cookie", "sekret"),
        host="app2.example.com",
        ip="127.0.0.1",
        user_agent="ua",
        validated=datetime.now(UTC),
        rp_id="example.com",
    )
    satellite._apply(replica, "sessions", session.key, "upsert", _builtins(session))
    stored = replica.sessions[session.key]
    assert stored.host == "app2.example.com"
    assert stored.validated == session.validated
    assert stored.user_uuid == UUID(int=1)


def test_apply_credential_bytes_roundtrip():
    """credential_id/public_key are bytes over the wire (base64 in JSON)."""
    replica = DB()
    cred = Credential.create(
        credential_id=secrets.token_bytes(32),
        user=UUID(int=1),
        aaguid=UUID(int=0),
        public_key=secrets.token_bytes(64),
        sign_count=3,
        rp_id="example.com",
    )
    cred.uuid = UUID(int=9)
    # Simulate the full wire path: builtins -> JSON -> builtins
    wire = msgspec.json.decode(msgspec.json.encode(_builtins(cred)))
    satellite._apply(replica, "credentials", str(cred.uuid), "upsert", wire)
    stored = replica.credentials[cred.uuid]
    assert stored.credential_id == cred.credential_id
    assert stored.public_key == cred.public_key
    assert stored.sign_count == 3


def test_feed_emit_and_replay():
    feed = syncfeed.SyncFeed()
    user = User.create(display_name="A", role=UUID(int=1))
    feed.emit("users", "k1", user)
    feed.emit("users", "k1", None)
    assert feed.seq == 2
    assert feed.replay_since(0)[0]["op"] == "upsert"
    assert feed.replay_since(1)[0]["op"] == "delete"
    assert feed.replay_since(2) == []
    assert feed.replay_since(99) is None


def test_feed_ring_overflow_replay_none():
    feed = syncfeed.SyncFeed()
    feed.events = collections.deque(maxlen=3)
    for i in range(5):
        feed.emit("users", f"k{i}", None)
    assert feed.replay_since(0) is None  # fell off the ring
    assert [e["seq"] for e in feed.replay_since(4)] == [5]
    assert feed.replay_since(5) == []


@pytest.mark.asyncio
async def test_operations_emit_events(test_db):
    """Writes through db.operations land on the sync feed."""
    syncfeed.feed.events.clear()
    syncfeed.feed.seq = 0
    user = next(iter(test_db.users.values()))
    ops_db.update_user_display_name(user.uuid, "Renamed")
    tables = {e["table"] for e in syncfeed.feed.events}
    assert "users" in tables
    key = syncfeed.feed.events[-1]["key"]
    assert syncfeed.feed.events[-1]["fields"]["display_name"] == "Renamed"
    assert key == str(user.uuid)


@pytest.mark.asyncio
async def test_replica_refresh_and_evict():
    replica = satellite.RemoteReplica(RemoteConfig(url=REMOTE_URL, token="t"))
    token = secrets.token_urlsafe(12)
    session = Session.create(
        user=UUID(int=1),
        credential=UUID(int=2),
        key=hash_secret("cookie", token),
        host="app2.example.com",
        ip="1.1.1.1",
        user_agent="ua",
        validated=datetime(2020, 1, 1, tzinfo=UTC),
    )
    replica.db.sessions[session.key] = session

    now = datetime.now(UTC)
    replica.refresh_session(session.key, now, "2.2.2.2", "new-ua")
    assert replica.db.sessions[session.key].validated == now
    queued = replica._pending_refresh[session.key]
    assert queued["type"] == "session_refresh"
    assert queued["ip"] == "2.2.2.2"

    replica.evict_session(token)
    assert not replica.db.sessions


def test_availability_gate():
    replica = satellite.RemoteReplica(
        RemoteConfig(url=REMOTE_URL, token="t", cache_ttl=60)
    )
    assert not replica.available()  # never synced
    replica.last_contact = time.monotonic()
    assert replica.available()


# -------------------------------------------------------------------------
# API-level: endpoints served from an injected replica
# -------------------------------------------------------------------------


def _replica_db() -> tuple[DB, str]:
    """A replica DB holding one org/role/perm/user/credential/session."""
    replica = DB()
    org = Org.create(display_name="Org")
    org.uuid = UUID(int=101)
    replica.orgs[org.uuid] = org
    perm = Permission.create(scope="auth:admin", display_name="Admin")
    perm.uuid = UUID(int=102)
    perm.orgs[org.uuid] = True
    replica.permissions[perm.uuid] = perm
    role = Role.create(org=org.uuid, display_name="Admins", permissions={perm.uuid})
    role.uuid = UUID(int=103)
    replica.roles[role.uuid] = role
    user = User.create(display_name="Remote Admin", role=role.uuid)
    user.uuid = UUID(int=104)
    replica.users[user.uuid] = user
    cred = Credential.create(
        credential_id=b"cid",
        user=user.uuid,
        aaguid=UUID(int=0),
        public_key=b"pk",
        sign_count=0,
        rp_id="example.com",
    )
    cred.uuid = UUID(int=105)
    replica.credentials[cred.uuid] = cred
    secret = secrets.token_urlsafe(12)
    session = Session.create(
        user=user.uuid,
        credential=cred.uuid,
        key=hash_secret("cookie", secret),
        host="app2.example.com",
        ip="127.0.0.1",
        user_agent="pytest",
        validated=datetime.now(UTC),
        rp_id="example.com",
    )
    replica.sessions[session.key] = session
    return replica, secret


@pytest_asyncio.fixture
async def remote_client(test_db):
    """ASGI client with example.com as a remote domain on a warm replica."""
    config = _remote_domain_config()
    domains.configure(listen=["localhost:4401"])
    domains.init_registry(config)
    replica_db, secret = _replica_db()
    replica = satellite.RemoteReplica(RemoteConfig(url=REMOTE_URL, token="t"))
    replica.db = replica_db
    replica.last_contact = time.monotonic()
    replica.connected = True
    satellite.manager.replicas[REMOTE_URL] = replica
    satellite.attach_stores()
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(
        transport=transport, base_url="http://localhost:4401"
    ) as client:
        yield client, secret, replica
    satellite.manager.replicas.pop(REMOTE_URL, None)


@pytest.mark.asyncio
async def test_forward_served_from_replica(remote_client):
    client, secret, _ = remote_client
    r = await client.get(
        "/auth/api/forward?perm=auth:admin",
        headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"},
    )
    assert r.status_code == 204
    assert r.headers["remote-name"] == "Remote Admin"
    assert r.headers["remote-groups"] == "auth:admin"


@pytest.mark.asyncio
async def test_forward_replica_denies_missing_perm(remote_client):
    client, secret, _ = remote_client
    r = await client.get(
        "/auth/api/forward?perm=other:scope",
        headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"},
    )
    assert r.status_code == 403


@pytest.mark.asyncio
async def test_validate_renews_locally_and_queues_writebehind(remote_client):
    client, secret, replica = remote_client
    session = next(iter(replica.db.sessions.values()))
    session.validated = datetime(2020, 1, 1, tzinfo=UTC)  # force refresh threshold
    r = await client.post(
        "/auth/api/validate",
        headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"},
    )
    assert r.status_code == 200
    assert r.json()["renewed"] is True
    assert session.validated.year > 2020  # applied to the replica
    queued = replica._pending_refresh[session.key]
    assert queued["type"] == "session_refresh"


@pytest.mark.asyncio
async def test_remote_domain_503_when_replica_stale(remote_client):
    client, secret, replica = remote_client
    replica.connected = False
    replica.last_contact = 0
    r = await client.get(
        "/auth/api/forward",
        headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"},
    )
    assert r.status_code == 503


@pytest.mark.asyncio
async def test_logout_proxied_and_evicted(remote_client, monkeypatch):
    client, secret, replica = remote_client

    async def fake_proxy(request, remote):
        return Response(status_code=200, content=b'{"message": "Logged out"}')

    monkeypatch.setattr("paskia.fastapi.proxy.proxy_to_remote", fake_proxy)
    r = await client.post(
        "/auth/api/logout",
        headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"},
    )
    assert r.status_code == 200
    assert not replica.db.sessions  # evicted optimistically


@pytest.mark.asyncio
async def test_admin_configures_remote_domain(client, session_token, test_db):
    """The admin domains API stores remote config and masks the token."""
    r = await client.post(
        "/auth/api/admin/domains/",
        json={
            "rp_id": "example.com",
            "rp_name": "Example",
            "origins": {"**.example.com": True, "auth.example.com": {"auth_host": True}},
            "remote": {"url": "http://remote.test", "token": "sekret", "cache_ttl": 30},
        },
        headers={"Host": "localhost:4401", "Cookie": f"{AUTH_COOKIE_NAME}={session_token}"},
    )
    assert r.status_code == 200, r.text
    stored = test_db.config.domains["example.com"]
    assert stored.remote.url == "http://remote.test"
    assert stored.remote.token == "sekret"

    r = await client.get(
        "/auth/api/admin/domains/",
        headers={"Host": "localhost:4401", "Cookie": f"{AUTH_COOKIE_NAME}={session_token}"},
    )
    entry = next(d for d in r.json() if d["rp_id"] == "example.com")
    assert entry["remote"]["url"] == "http://remote.test"
    assert "token" not in entry["remote"]  # write-only


@pytest.mark.asyncio
async def test_admin_remote_domain_requires_auth_host(client, session_token):
    r = await client.post(
        "/auth/api/admin/domains/",
        json={
            "rp_id": "example.com",
            "origins": {"**.example.com": True},
            "remote": {"url": "http://remote.test"},
        },
        headers={"Host": "localhost:4401", "Cookie": f"{AUTH_COOKIE_NAME}={session_token}"},
    )
    assert r.status_code == 400
    assert "auth host" in r.json()["detail"]