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 upstream (its sync WebSocket fed from the
process-global database), while the satellite side runs through the ASGI
app with a real UplinkReplica started by the real SatelliteManager. Unlike
test_remote.py (which injects a warm replica by hand), nothing here is
stubbed: the 3-way handshake, pushed domains, snapshot, live events,
write-behind, dead-peer TTL handling, reconnect reconciliation, live scope
changes and revocation all cross a TCP socket.

One in-process distortion remains: both sides share the module-level
database and registry, so the satellite's pushed-domain application is
arranged to be a no-op (the link's domains are absent from the shared
config — the satellite registry is synthetic instead), and endpoints the
satellite *forwards* to the upstream (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 msgspec.structs
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,
    Uplink,
)
from paskia.fastapi import sync
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"
LINK_ID = "live-link"
LINK_KEY = b"e2e-link-key"


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_server(test_db):
    """A uvicorn-served upstream over a real TCP socket.

    Yields a namespace: db, url, port, start_server (for restarts),
    servers.
    """
    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}"
    domains.configure(listen=["localhost:4401"])
    domains.init_registry(test_db.config)
    await start_server(port)
    try:
        yield SimpleNamespace(
            db=test_db, url=url, port=port, start_server=start_server, servers=servers
        )
    finally:
        for server, task in servers:
            server.should_exit = True
            await task


@pytest_asyncio.fixture
async def live_remote(live_server):
    """An established link against the live upstream, and the satellite
    manager started for real: the replica connects over TCP and runs the
    established-link handshake (hello {link, key} → welcome → ready).

    Both sides' link rows are stored directly (registration itself is
    covered by the dedicated tests); the satellite's remote domain lives
    only in the registry, not the shared config, so pushed-domain
    application stays a no-op.
    """
    live = live_server
    ops_db.create_satellite_link(LINK_ID, [REMOTE_DOMAIN], LINK_KEY, label="e2e")
    uplink = Uplink(url=live.url, created=datetime.now(UTC), key=LINK_KEY)
    ops_db.create_uplink(LINK_ID, uplink, {})

    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(link=LINK_ID),
                ),
            }
        )
    )
    await satellite.manager.start()
    replica = satellite.manager.replicas[LINK_ID]
    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:
            live.client = client
            live.replica = replica
            yield live
        finally:
            await satellite.manager.stop()


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 link's own domains' credentials replicate."""
    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


# -------------------------------------------------------------------------
# Registration (the real 3-way handshake, crossing TCP)
# -------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_live_registration(live_server):
    """A RAM-only offer is spent by the handshake; both sides store the
    link with identical key material, and the established-link handshake
    works right after."""
    live = live_server
    offer_url = sync.create_offer([], live.url)
    assert offer_url.startswith("ws://")

    result = await satellite.register_uplink(offer_url)
    link_id = result["link"]
    assert result["domains"] == []  # empty scope: register first, scope later

    uplink = live.db.uplinks[link_id]
    assert uplink.url == live.url
    link = live.db.satellite_links[link_id]
    assert uplink.key == link.key and len(link.key) == 64  # both parts
    assert link.label  # the satellite's self-reported name

    # Re-registering to the same upstream is refused (the offer itself is
    # single-use — covered in test_remote.py's offer test)
    with pytest.raises(ValueError, match="Already linked"):
        await satellite.register_uplink(offer_url)

    # The established-link handshake: the replica connects and syncs
    await satellite.manager.start()
    try:
        replica = satellite.manager.replicas[link_id]
        await wait_for("established link sync", lambda: replica.available())
    finally:
        await satellite.manager.stop()


@pytest.mark.asyncio
async def test_live_registration_rejections(live_server):
    """Malformed URLs, unknown keys and unreachable hosts all fail the
    registration without storing anything."""
    with pytest.raises(ValueError, match="wss"):
        await satellite.register_uplink("https://key@auth.example.com")
    with pytest.raises(ValueError, match="key"):
        await satellite.register_uplink("wss://auth.example.com")
    with pytest.raises(RuntimeError, match="rejected"):
        await satellite.register_uplink(
            f"ws://unknown.key@127.0.0.1:{live_server.port}"
        )
    with pytest.raises(RuntimeError, match="unreachable"):
        await satellite.register_uplink("ws://some.key@127.0.0.1:1")
    assert not live_server.db.uplinks
    assert not live_server.db.satellite_links


# -------------------------------------------------------------------------
# Established link: sync behavior
# -------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_live_snapshot_and_context_repush(
    live_remote, test_user, remote_credential
):
    """Sessions arrive with their upstream-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 upstream authorizes locally; the /validate
    refresh travels back over the wire into the upstream'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 upstream (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 upstream",
        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_key_rejected(live_remote):
    """An uplink whose key does not match the stored link's never syncs."""
    uplink = Uplink(url=live_remote.url, created=datetime.now(UTC), key=b"not-the-key")
    uplink.uplink_id = "bad-link"
    bad = satellite.UplinkReplica(uplink)
    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

    # Expire the trust window (a copy: the db-owned struct is not mutated)
    uplink_copy = msgspec.structs.replace(live.replica.uplink, cache_ttl=0)
    uplink_copy.uplink_id = LINK_ID
    live.replica.uplink = uplink_copy
    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_unlink_stops_replica(live_remote):
    """Unlinking drops the uplink, its domains and the replica."""
    live = live_remote
    await satellite.manager.unlink(LINK_ID)
    assert LINK_ID not in satellite.manager.replicas
    assert LINK_ID not in live.db.uplinks


@pytest.mark.asyncio
async def test_live_revocation_unlinks_satellite(live_remote):
    """Revoking the link upstream drops the connection; the satellite's
    handshake retry gets a policy close and unlinks its side."""
    live = live_remote
    ops_db.delete_satellite_link(LINK_ID)
    await sync.close_link(LINK_ID)
    await wait_for(
        "satellite to unlink", lambda: LINK_ID not in live.db.uplinks, timeout=25
    )
    assert LINK_ID 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 link's rp-ids: 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 link's scope is ignored
    by the upstream (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_shrink_stops_feed(live_remote, test_user, remote_credential):
    """Narrowing the link's scope on the upstream takes effect on the live
    connection: sessions of removed domains stop flowing."""
    live = live_remote
    ops_db.set_satellite_link_scope(LINK_ID, [])
    await sync.push_link_domains(LINK_ID)

    removed_key, _ = create_test_session(
        user_uuid=test_user.uuid,
        credential_uuid=remote_credential.uuid,
        host=APP_HOST,
        rp_id=REMOTE_DOMAIN,
    )
    # Positive control: widen the scope again; a session created now does
    # arrive, so the earlier one demonstrably passed a closed pipe.
    ops_db.set_satellite_link_scope(LINK_ID, [REMOTE_DOMAIN])
    await sync.push_link_domains(LINK_ID)
    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(
        "re-scoped session sync", lambda: remote_key in live.replica.db.sessions
    )
    assert removed_key not in live.replica.db.sessions


@pytest.mark.asyncio
async def test_live_domain_conflict_reported_upstream(live_remote, test_db):
    """A pushed domain the satellite serves locally is rejected, and the
    rejection is reported into the upstream's link state."""
    # The shared config now holds the domain authoritatively — from the
    # satellite's view this is the local-conflict case.
    ops_db.create_domain(
        REMOTE_DOMAIN,
        DomainConfig(
            rp_name="Example",
            origins={
                f"**.{REMOTE_DOMAIN}": True,
                f"auth.{REMOTE_DOMAIN}": OriginEntry(auth_host=True),
            },
        ),
    )
    await sync.push_link_domains(LINK_ID)
    await wait_for(
        "rejection to arrive",
        lambda: (
            REMOTE_DOMAIN in sync.links_status().get(LINK_ID, {}).get("rejected", {})
        ),
    )
    # The satellite kept its (synthetic) registry — no local overwrite
    assert test_db.config.domains[REMOTE_DOMAIN].remote is None