389 lines
12 KiB
Python
389 lines
12 KiB
Python
"""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"]
|