paskia/tests/test_satellite_e2e.py

"""Live end-to-end tests for satellite domains over a real sync channel.

A real uvicorn server plays the remote (its sync WebSocket fed from the
process-global database), while the satellite side runs through the ASGI
app with a real RemoteReplica started by the real SatelliteManager
reconcile. Unlike test_remote.py (which injects a warm replica by hand),
nothing here is stubbed: the snapshot, live events, write-behind, dead-peer
TTL handling and reconnect reconciliation all cross a TCP socket.

One in-process distortion remains: both sides share the module-level
database and registry, so endpoints the satellite *forwards* to the remote
(logout, set-session, OIDC) would re-forward and loop here — those paths
are covered in test_remote.py with a mocked forward_request.
"""

import asyncio
import os
import socket
import time
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from uuid import UUID

import httpx
import pytest
import pytest_asyncio
import uvicorn

import paskia.db.operations as ops_db
from paskia import domains, satellite
from paskia.db.structs import (
    Config,
    Credential,
    DomainConfig,
    OriginEntry,
    RemoteConfig,
)
from paskia.fastapi.mainapp import app
from paskia.fastapi.session import AUTH_COOKIE_NAME

from .conftest import TEST_RP_ID, create_test_session

REMOTE_DOMAIN = "example.com"
APP_HOST = "app2.example.com"


def _free_port() -> int:
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


async def wait_for(desc: str, condition, timeout: float = 10) -> None:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if condition():
            return
        await asyncio.sleep(0.02)
    raise AssertionError(f"timed out waiting for {desc}")


@pytest_asyncio.fixture
async def live_remote(test_db):
    """A uvicorn-served 'remote' and the satellite manager wired against it.

    The manager is started for real: reconcile builds the replica from the
    domain config and its sync client connects over TCP. Yielded namespace:
    client (ASGI, satellite side), replica, db (the remote's database),
    url/port of the remote, start_server (for restarts), servers.
    """
    _, live_token = ops_db.create_sync_token(REMOTE_DOMAIN)
    servers: list[tuple[uvicorn.Server, asyncio.Task]] = []

    async def start_server(port: int) -> uvicorn.Server:
        config = uvicorn.Config(
            app, host="127.0.0.1", port=port, log_level="error", lifespan="off"
        )
        server = uvicorn.Server(config)
        task = asyncio.create_task(server.serve())
        servers.append((server, task))
        while not server.started:
            await asyncio.sleep(0.02)
        return server

    port = _free_port()
    url = f"http://127.0.0.1:{port}"
    await start_server(port)

    domains.configure(listen=["localhost:4401"])
    domains.init_registry(
        Config(
            domains={
                TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True}),
                REMOTE_DOMAIN: DomainConfig(
                    origins={
                        f"**.{REMOTE_DOMAIN}": True,
                        f"auth.{REMOTE_DOMAIN}": OriginEntry(auth_host=True),
                    },
                    remote=RemoteConfig(url=url, token=live_token),
                ),
            }
        )
    )
    await satellite.manager.start()
    replica = satellite.manager.replicas[REMOTE_DOMAIN]
    await wait_for("initial snapshot", lambda: replica.available())

    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(
        transport=transport, base_url="http://localhost:4401"
    ) as client:
        try:
            yield SimpleNamespace(
                client=client,
                replica=replica,
                db=test_db,
                url=url,
                port=port,
                token=live_token,
                start_server=start_server,
                servers=servers,
            )
        finally:
            await satellite.manager.stop()
            for server, task in servers:
                server.should_exit = True
                await task


def _app_headers(secret: str) -> dict[str, str]:
    return {"Host": APP_HOST, "Cookie": f"{AUTH_COOKIE_NAME}={secret}"}


@pytest_asyncio.fixture
async def remote_credential(test_db, test_user):
    """A credential registered on the remote domain. The feed is rp-id
    scoped: only the domain's own credentials replicate to its satellites."""
    credential = Credential.create(
        credential_id=os.urandom(32),
        user=test_user.uuid,
        aaguid=UUID("00000000-0000-0000-0000-000000000000"),
        public_key=os.urandom(64),
        sign_count=0,
        rp_id=REMOTE_DOMAIN,
    )
    ops_db.create_credential(credential)
    return credential


@pytest.mark.asyncio
async def test_live_snapshot_and_context_repush(
    live_remote, test_user, remote_credential
):
    """Sessions arrive with their master-computed context; user-graph
    changes eagerly re-push the affected sessions' contexts."""
    live = live_remote
    key, _ = create_test_session(
        user_uuid=test_user.uuid,
        credential_uuid=remote_credential.uuid,
        host=APP_HOST,
        rp_id=REMOTE_DOMAIN,
    )
    await wait_for("session sync", lambda: key in live.replica.db.session_contexts)
    assert live.replica.db.session_contexts[key].user.display_name == "Test Admin"

    ops_db.update_user_display_name(test_user.uuid, "Live Rename")
    await wait_for(
        "rename to propagate",
        lambda: (
            live.replica.db.session_contexts[key].user.display_name == "Live Rename"
        ),
    )


@pytest.mark.asyncio
async def test_live_permission_revocation_propagates(
    live_remote, test_user, remote_credential
):
    """Removing a permission from the user's role re-pushes the session
    context: the satellite denies within one RTT."""
    live = live_remote
    key, secret = create_test_session(
        user_uuid=test_user.uuid,
        credential_uuid=remote_credential.uuid,
        host=APP_HOST,
        rp_id=REMOTE_DOMAIN,
    )
    await wait_for("session sync", lambda: key in live.replica.db.session_contexts)
    r = await live.client.get(
        "/auth/api/forward?perm=auth:admin", headers=_app_headers(secret)
    )
    assert r.status_code == 204

    perm_uuid = next(
        p.uuid for p in live.db.permissions.values() if p.scope == "auth:admin"
    )
    ops_db.remove_permission_from_role(test_user.role_uuid, perm_uuid)
    await wait_for(
        "revocation to propagate",
        lambda: "auth:admin" not in live.replica.db.session_contexts[key].permissions,
    )
    r = await live.client.get(
        "/auth/api/forward?perm=auth:admin", headers=_app_headers(secret)
    )
    assert r.status_code == 403


@pytest.mark.asyncio
async def test_live_forward_and_writebehind(live_remote, test_user, remote_credential):
    """A session created on the remote authorizes locally; the /validate
    refresh travels back over the wire into the remote's database."""
    live = live_remote
    key, secret = create_test_session(
        user_uuid=test_user.uuid,
        credential_uuid=remote_credential.uuid,
        host=APP_HOST,
        rp_id=REMOTE_DOMAIN,
    )
    await wait_for("session sync", lambda: key in live.replica.db.sessions)

    r = await live.client.get(
        "/auth/api/forward?perm=auth:admin", headers=_app_headers(secret)
    )
    assert r.status_code == 204
    assert r.headers["remote-name"] == "Test Admin"

    # Age the session on the remote (propagates), then validate on the satellite
    stale = datetime.now(UTC) - timedelta(minutes=10)
    ops_db.update_session(key, validated=stale)
    await wait_for(
        "staleness to propagate",
        lambda: live.replica.db.sessions[key].validated == stale,
    )
    r = await live.client.post("/auth/api/validate", headers=_app_headers(secret))
    assert r.status_code == 200
    assert r.json()["renewed"] is True
    await wait_for(
        "write-behind to land on the remote",
        lambda: live.db.sessions[key].validated > stale,
    )


@pytest.mark.asyncio
async def test_live_session_delete_propagates(
    live_remote, test_user, remote_credential
):
    live = live_remote
    key, secret = create_test_session(
        user_uuid=test_user.uuid,
        credential_uuid=remote_credential.uuid,
        host=APP_HOST,
        rp_id=REMOTE_DOMAIN,
    )
    await wait_for("session sync", lambda: key in live.replica.db.sessions)

    ops_db.delete_session(key)
    await wait_for(
        "session delete to propagate", lambda: key not in live.replica.db.sessions
    )
    r = await live.client.get("/auth/api/forward", headers=_app_headers(secret))
    assert r.status_code == 401


@pytest.mark.asyncio
async def test_live_wrong_token_rejected(live_remote):
    bad = satellite.RemoteReplica(
        RemoteConfig(url=live_remote.url, token="wrong"), REMOTE_DOMAIN
    )
    await bad.start()
    try:
        await asyncio.sleep(1.5)
        assert not bad.available()
    finally:
        await bad.stop()


@pytest.mark.asyncio
async def test_live_disconnect_ttl_and_reconnect(
    live_remote, test_user, remote_credential
):
    """Channel down: replica keeps serving within cache_ttl, then fails
    closed (503); reconnect starts from a snapshot and reconciles drift."""
    live = live_remote
    key, secret = create_test_session(
        user_uuid=test_user.uuid,
        credential_uuid=remote_credential.uuid,
        host=APP_HOST,
        rp_id=REMOTE_DOMAIN,
    )
    await wait_for("session sync", lambda: key in live.replica.db.sessions)

    server, task = live.servers[0]
    server.should_exit = True
    await task
    await wait_for(
        "dead peer detection", lambda: not live.replica.connected, timeout=15
    )

    r = await live.client.get("/auth/api/forward", headers=_app_headers(secret))
    assert r.status_code == 204  # still authoritative within cache_ttl

    live.replica.remote.cache_ttl = 0  # expire the trust window
    r = await live.client.get("/auth/api/forward", headers=_app_headers(secret))
    assert r.status_code == 503

    # Drift while the channel is down, then let the client auto-reconnect
    ops_db.update_user_display_name(test_user.uuid, "While Down")
    await live.start_server(live.port)
    await wait_for("reconnect", lambda: live.replica.available(), timeout=20)
    assert (
        live.replica.db.session_contexts[key].user.display_name == "While Down"
    )  # snapshot reconciled the drift
    r = await live.client.get("/auth/api/forward", headers=_app_headers(secret))
    assert r.status_code == 204


@pytest.mark.asyncio
async def test_reconcile_stops_replica_when_remote_removed(live_remote):
    """A registry rebuild without the remote config stops the replica."""
    domains.init_registry(
        Config(domains={TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True})})
    )
    await wait_for(
        "replica shutdown", lambda: REMOTE_DOMAIN not in satellite.manager.replicas
    )


@pytest.mark.asyncio
async def test_live_out_of_scope_session_not_replicated(
    live_remote, test_user, test_credential, remote_credential
):
    """The feed is scoped to the token's rp-id: sessions of other domains
    (here the satellite's own) never cross the wire."""
    live = live_remote
    local_key, _ = create_test_session(
        user_uuid=test_user.uuid,
        credential_uuid=test_credential.uuid,
        host="localhost:4401",
        rp_id=TEST_RP_ID,
    )
    # Positive control: an in-scope session created afterwards does arrive,
    # so the out-of-scope one has already passed through the pipe.
    remote_key, _ = create_test_session(
        user_uuid=test_user.uuid,
        credential_uuid=remote_credential.uuid,
        host=APP_HOST,
        rp_id=REMOTE_DOMAIN,
    )
    await wait_for(
        "in-scope session sync", lambda: remote_key in live.replica.db.sessions
    )
    assert local_key not in live.replica.db.sessions


@pytest.mark.asyncio
async def test_live_out_of_scope_writebehind_ignored(
    live_remote, test_user, test_credential, remote_credential
):
    """A session_refresh for a session outside the token's scope is ignored
    by the remote (fail-closed), while an in-scope one lands."""
    live = live_remote
    stale = datetime.now(UTC) - timedelta(minutes=10)
    local_key, _ = create_test_session(
        user_uuid=test_user.uuid,
        credential_uuid=test_credential.uuid,
        host="localhost:4401",
        rp_id=TEST_RP_ID,
    )
    remote_key, _ = create_test_session(
        user_uuid=test_user.uuid,
        credential_uuid=remote_credential.uuid,
        host=APP_HOST,
        rp_id=REMOTE_DOMAIN,
    )
    ops_db.update_session(local_key, validated=stale)
    ops_db.update_session(remote_key, validated=stale)

    now = datetime.now(UTC)
    live.replica.refresh_session(local_key, now, None, None)
    live.replica.refresh_session(remote_key, now, None, None)
    await wait_for(
        "in-scope refresh to land",
        lambda: live.db.sessions[remote_key].validated > stale,
    )
    assert live.db.sessions[local_key].validated == stale


@pytest.mark.asyncio
async def test_live_scope_mismatch_rejected(live_remote):
    """A valid token scoped to another domain is useless: the ready message
    names its scope and the replica refuses to go live on the mismatch."""
    bad = satellite.RemoteReplica(
        RemoteConfig(url=live_remote.url, token=live_remote.token), "other.org"
    )
    await bad.start()
    try:
        await asyncio.sleep(1.5)
        assert not bad.available()
    finally:
        await bad.stop()