109 lines
4.5 KiB
Python
109 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from sqlalchemy import event, select
|
|
|
|
from packages.hub_clients import ClientKind, ClientStatus
|
|
from services.control_api.auth.models import AuthenticatedPrincipal, PrincipalKind
|
|
from services.control_api.persistence.models import AuditEventRow, ClientBindingRow, ClientRow
|
|
from services.runtime.client_service import PersistentClientService
|
|
|
|
pytest_plugins = ("tests.test_machine_client_provisioning",)
|
|
|
|
|
|
def _actor() -> AuthenticatedPrincipal:
|
|
now = datetime.now(UTC)
|
|
return AuthenticatedPrincipal(
|
|
subject="operator-audit-fixture",
|
|
issuer="https://identity.fixture.invalid",
|
|
audience=("hub-fixture",),
|
|
kind=PrincipalKind.MANAGEMENT_USER,
|
|
issued_at=now,
|
|
expires_at=now + timedelta(minutes=5),
|
|
)
|
|
|
|
|
|
def _active_client(database: Any) -> tuple[PersistentClientService, Any]:
|
|
service = PersistentClientService(
|
|
database, issuer="https://identity.fixture.invalid", public_host="mcp.fixture.invalid"
|
|
)
|
|
record, _, _ = service.create(
|
|
name="Audit fixture", project="ludarium", environment="prod", kind=ClientKind.STDIO_BRIDGE
|
|
)
|
|
with database() as session, session.begin():
|
|
row = session.get(ClientRow, record.id)
|
|
assert row is not None
|
|
row.status = "active"
|
|
binding = session.scalar(
|
|
select(ClientBindingRow).where(ClientBindingRow.client_id == record.id)
|
|
)
|
|
assert binding is not None
|
|
binding.enabled = True
|
|
return service, record.id
|
|
|
|
|
|
def test_client_disable_and_revoke_are_scoped_and_append_actor_audits(database: Any) -> None:
|
|
service, client_id = _active_client(database)
|
|
actor = _actor()
|
|
disabled = service.transition(client_id, ClientStatus.DISABLED, actor)
|
|
assert disabled.status is ClientStatus.DISABLED
|
|
revoked = service.transition(client_id, ClientStatus.REVOKED, actor)
|
|
assert revoked.status is ClientStatus.REVOKED
|
|
with database() as session:
|
|
binding = session.scalar(
|
|
select(ClientBindingRow).where(ClientBindingRow.client_id == client_id)
|
|
)
|
|
assert binding is not None and binding.enabled is False
|
|
audits = session.scalars(
|
|
select(AuditEventRow)
|
|
.where(AuditEventRow.client_id == client_id)
|
|
.order_by(AuditEventRow.occurred_at)
|
|
).all()
|
|
assert [row.event_type for row in audits] == ["client.disable", "client.revoke"]
|
|
for row in audits:
|
|
assert row.principal_subject == actor.subject
|
|
assert row.principal_issuer == actor.issuer
|
|
assert row.project_id == binding.project_id
|
|
assert row.virtual_server_id == binding.virtual_server_id
|
|
assert row.environment_id is not None
|
|
assert row.policy_decision == "allow" and row.outcome == "success"
|
|
assert row.sanitized_arguments == {"clientId": str(client_id), "action": row.event_type}
|
|
assert row.arguments_hash is not None and len(row.arguments_hash) == 64
|
|
assert audits[0].trace_id != audits[1].trace_id
|
|
with pytest.raises(ValueError, match="revoked"):
|
|
service.transition(client_id, ClientStatus.DISABLED, actor)
|
|
with database() as session:
|
|
row = session.get(ClientRow, client_id)
|
|
assert row is not None and row.status == "revoked"
|
|
assert len(session.scalars(select(AuditEventRow)).all()) == 2
|
|
|
|
|
|
@pytest.mark.parametrize("status", [ClientStatus.DISABLED, ClientStatus.REVOKED])
|
|
def test_audit_failure_rolls_back_client_and_binding_transition(
|
|
database: Any, status: ClientStatus
|
|
) -> None:
|
|
service, client_id = _active_client(database)
|
|
|
|
def reject_audit(*_: Any) -> None:
|
|
raise RuntimeError("fixture audit insert failed")
|
|
|
|
event.listen(AuditEventRow, "before_insert", reject_audit)
|
|
try:
|
|
with pytest.raises(RuntimeError, match="fixture audit insert failed"):
|
|
service.transition(client_id, status, _actor())
|
|
finally:
|
|
event.remove(AuditEventRow, "before_insert", reject_audit)
|
|
with database() as session:
|
|
row = session.get(ClientRow, client_id)
|
|
binding = session.scalar(
|
|
select(ClientBindingRow).where(ClientBindingRow.client_id == client_id)
|
|
)
|
|
assert row is not None and row.status == "active"
|
|
assert row.revoked_at is None
|
|
assert row.record_metadata["operatorDisabled"] is False
|
|
assert binding is not None and binding.enabled is True
|
|
assert session.scalars(select(AuditEventRow)).all() == []
|