387 lines
13 KiB
Python
387 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
import pytest
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from govoplan_core.commands.first_admin import _write_private_json
|
|
from govoplan_core.core.access import (
|
|
FirstAdminProvisioner,
|
|
FirstAdminProvisioningError,
|
|
FirstSystemAdministratorRef,
|
|
)
|
|
from govoplan_core.core.first_admin import (
|
|
FirstAdminEnrollment,
|
|
FirstAdminEnrollmentConflict,
|
|
FirstAdminEnrollmentCredentialError,
|
|
FirstAdminEnrollmentEvent,
|
|
FirstAdminEnrollmentState,
|
|
FirstAdminEnrollmentUnavailable,
|
|
consume_first_admin_credential,
|
|
first_admin_enrollment_status,
|
|
issue_first_admin_credential,
|
|
)
|
|
from govoplan_core.core.modules import ModuleContext, ModuleManifest
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.db.session import get_session
|
|
from govoplan_core.server.bootstrap import create_bootstrap_router
|
|
from govoplan_core.tenancy.scope import scope_registry
|
|
|
|
|
|
class _Provisioner(FirstAdminProvisioner):
|
|
def __init__(self, *, administrator_exists: bool = False, fail_create: bool = False) -> None:
|
|
self.administrator_exists = administrator_exists
|
|
self.fail_create = fail_create
|
|
self.create_count = 0
|
|
|
|
def has_durable_system_administrator(self, session: object) -> bool:
|
|
del session
|
|
return self.administrator_exists
|
|
|
|
def create_first_system_administrator(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant: object,
|
|
email: str,
|
|
display_name: str | None,
|
|
password: str,
|
|
) -> FirstSystemAdministratorRef:
|
|
del session, password
|
|
if self.fail_create:
|
|
raise FirstAdminProvisioningError("simulated authority failure")
|
|
self.create_count += 1
|
|
self.administrator_exists = True
|
|
return FirstSystemAdministratorRef(
|
|
account_id="account-1",
|
|
email=email,
|
|
display_name=display_name,
|
|
membership_id="membership-1",
|
|
tenant_id=str(getattr(tenant, "id")),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def session() -> Session:
|
|
engine = create_engine(
|
|
"sqlite+pysqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
scope_registry.metadata.create_all(engine)
|
|
Base.metadata.create_all(
|
|
engine,
|
|
tables=[
|
|
FirstAdminEnrollment.__table__,
|
|
FirstAdminEnrollmentEvent.__table__,
|
|
],
|
|
)
|
|
with Session(engine, expire_on_commit=False) as item:
|
|
yield item
|
|
engine.dispose()
|
|
|
|
|
|
def test_single_use_enrollment_creates_authority_and_replays_idempotently(
|
|
session: Session,
|
|
) -> None:
|
|
provisioner = _Provisioner()
|
|
now = datetime(2026, 8, 4, 12, tzinfo=timezone.utc)
|
|
with patch("govoplan_core.core.first_admin.audit_event"):
|
|
issued = issue_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
ttl_seconds=900,
|
|
reason="initial installation",
|
|
now=now,
|
|
)
|
|
session.commit()
|
|
result = consume_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
secret=issued.secret,
|
|
email="owner@example.test",
|
|
display_name="System Owner",
|
|
password="a-production-password",
|
|
tenant_slug="default",
|
|
tenant_name="Default Tenant",
|
|
now=now + timedelta(minutes=1),
|
|
)
|
|
session.commit()
|
|
replay = consume_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
secret=issued.secret,
|
|
email="owner@example.test",
|
|
display_name="System Owner",
|
|
password="a-production-password",
|
|
tenant_slug="default",
|
|
tenant_name="Default Tenant",
|
|
now=now + timedelta(minutes=2),
|
|
)
|
|
|
|
assert not result.replayed
|
|
assert replay.replayed
|
|
assert replay.administrator.account_id == "account-1"
|
|
assert provisioner.create_count == 1
|
|
enrollment = session.get(FirstAdminEnrollment, "installation-1")
|
|
assert enrollment is not None
|
|
assert enrollment.state == FirstAdminEnrollmentState.CONSUMED.value
|
|
assert enrollment.consumed_account_id == "account-1"
|
|
assert enrollment.token_sha256 != issued.secret
|
|
evidence = session.scalars(
|
|
select(FirstAdminEnrollmentEvent).order_by(FirstAdminEnrollmentEvent.sequence)
|
|
).all()
|
|
assert [item.event_type for item in evidence] == [
|
|
"credential_issued",
|
|
"administrator_created",
|
|
]
|
|
assert evidence[1].previous_sha256 == evidence[0].event_sha256
|
|
assert issued.secret not in json.dumps([item.evidence for item in evidence])
|
|
|
|
|
|
def test_consumed_credential_rejects_a_different_request(session: Session) -> None:
|
|
provisioner = _Provisioner()
|
|
with patch("govoplan_core.core.first_admin.audit_event"):
|
|
issued = issue_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
ttl_seconds=900,
|
|
reason="initial installation",
|
|
)
|
|
consume_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
secret=issued.secret,
|
|
email="owner@example.test",
|
|
display_name=None,
|
|
password="a-production-password",
|
|
tenant_slug="default",
|
|
tenant_name="Default Tenant",
|
|
)
|
|
session.commit()
|
|
with pytest.raises(FirstAdminEnrollmentCredentialError, match="already been used"):
|
|
consume_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
secret=issued.secret,
|
|
email="other@example.test",
|
|
display_name=None,
|
|
password="a-production-password",
|
|
tenant_slug="default",
|
|
tenant_name="Default Tenant",
|
|
)
|
|
|
|
|
|
def test_recovery_rotates_lost_or_expired_material(session: Session) -> None:
|
|
provisioner = _Provisioner()
|
|
now = datetime(2026, 8, 4, 12, tzinfo=timezone.utc)
|
|
with patch("govoplan_core.core.first_admin.audit_event"):
|
|
first = issue_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
ttl_seconds=60,
|
|
reason="initial installation",
|
|
now=now,
|
|
)
|
|
with pytest.raises(FirstAdminEnrollmentConflict, match="already exists"):
|
|
issue_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
ttl_seconds=60,
|
|
reason="duplicate issue",
|
|
now=now + timedelta(seconds=30),
|
|
)
|
|
replacement = issue_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
ttl_seconds=300,
|
|
reason="lost credential",
|
|
replace_active=True,
|
|
now=now + timedelta(seconds=30),
|
|
)
|
|
session.commit()
|
|
with pytest.raises(FirstAdminEnrollmentCredentialError, match="invalid"):
|
|
consume_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
secret=first.secret,
|
|
email="owner@example.test",
|
|
display_name=None,
|
|
password="a-production-password",
|
|
tenant_slug="default",
|
|
tenant_name="Default Tenant",
|
|
now=now + timedelta(seconds=40),
|
|
)
|
|
assert replacement.generation == 2
|
|
|
|
|
|
def test_failed_authority_creation_rolls_back_without_consuming_secret(
|
|
session: Session,
|
|
) -> None:
|
|
provisioner = _Provisioner(fail_create=True)
|
|
with patch("govoplan_core.core.first_admin.audit_event"):
|
|
issued = issue_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
ttl_seconds=900,
|
|
reason="initial installation",
|
|
)
|
|
session.commit()
|
|
with pytest.raises(FirstAdminEnrollmentConflict, match="simulated"):
|
|
consume_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
secret=issued.secret,
|
|
email="owner@example.test",
|
|
display_name=None,
|
|
password="a-production-password",
|
|
tenant_slug="default",
|
|
tenant_name="Default Tenant",
|
|
)
|
|
session.rollback()
|
|
|
|
enrollment = session.get(FirstAdminEnrollment, "installation-1")
|
|
assert enrollment is not None
|
|
assert enrollment.state == FirstAdminEnrollmentState.ACTIVE.value
|
|
assert enrollment.consumed_account_id is None
|
|
|
|
|
|
def test_enrollment_is_unavailable_after_durable_admin_exists(session: Session) -> None:
|
|
provisioner = _Provisioner(administrator_exists=True)
|
|
with patch("govoplan_core.core.first_admin.audit_event"):
|
|
with pytest.raises(FirstAdminEnrollmentUnavailable):
|
|
issue_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
ttl_seconds=900,
|
|
reason="must fail",
|
|
)
|
|
readiness = first_admin_enrollment_status(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
)
|
|
assert not readiness.enrollment_required
|
|
assert readiness.state == "completed"
|
|
|
|
|
|
def test_operator_artifact_is_owner_readable_only(tmp_path: Path) -> None:
|
|
output = tmp_path / "bootstrap" / "first-admin.json"
|
|
_write_private_json(output, {"enrollment_token": "never-print-this"})
|
|
|
|
assert stat.S_IMODE(output.stat().st_mode) == 0o600
|
|
assert output.read_text(encoding="utf-8").endswith("\n")
|
|
assert os.geteuid() == output.stat().st_uid
|
|
|
|
|
|
def test_public_api_is_limited_to_readiness_and_single_enrollment(
|
|
session: Session,
|
|
) -> None:
|
|
provisioner = _Provisioner()
|
|
registry = PlatformRegistry()
|
|
registry.register(
|
|
ModuleManifest(
|
|
id="access",
|
|
name="Access",
|
|
version="test",
|
|
capability_factories={
|
|
"access.firstAdminProvisioner": lambda _context: provisioner,
|
|
},
|
|
)
|
|
)
|
|
settings = SimpleNamespace(installation_id="installation-1")
|
|
registry.configure_capability_context(
|
|
ModuleContext(registry=registry, settings=settings)
|
|
)
|
|
app = FastAPI()
|
|
app.state.govoplan_registry = registry
|
|
app.include_router(create_bootstrap_router(settings), prefix="/api/v1")
|
|
|
|
def _session_override():
|
|
yield session
|
|
|
|
app.dependency_overrides[get_session] = _session_override
|
|
with patch("govoplan_core.core.first_admin.audit_event"):
|
|
issued = issue_first_admin_credential(
|
|
session,
|
|
installation_id="installation-1",
|
|
provisioner=provisioner,
|
|
ttl_seconds=900,
|
|
reason="API test",
|
|
)
|
|
session.commit()
|
|
with TestClient(app) as client:
|
|
ready = client.get("/api/v1/bootstrap/status")
|
|
enrolled = client.post(
|
|
"/api/v1/bootstrap/first-admin",
|
|
headers={"X-GovOPlaN-Enrollment-Token": issued.secret},
|
|
json={
|
|
"email": "owner@example.test",
|
|
"display_name": "System Owner",
|
|
"password": "a-production-password",
|
|
"tenant_slug": "default",
|
|
"tenant_name": "Default Tenant",
|
|
},
|
|
)
|
|
completed = client.get("/api/v1/bootstrap/status")
|
|
|
|
assert ready.status_code == 200
|
|
assert ready.json()["credential_active"] is True
|
|
assert enrolled.status_code == 201
|
|
assert enrolled.json()["bootstrap_retired"] is True
|
|
assert completed.json()["state"] == "completed"
|
|
|
|
|
|
def test_readiness_reports_missing_access_without_exposing_an_enrollment_api(
|
|
session: Session,
|
|
) -> None:
|
|
registry = PlatformRegistry()
|
|
settings = SimpleNamespace(installation_id="installation-1")
|
|
app = FastAPI()
|
|
app.state.govoplan_registry = registry
|
|
app.include_router(create_bootstrap_router(settings), prefix="/api/v1")
|
|
|
|
def _session_override():
|
|
yield session
|
|
|
|
app.dependency_overrides[get_session] = _session_override
|
|
with TestClient(app) as client:
|
|
ready = client.get("/api/v1/bootstrap/status")
|
|
rejected = client.post(
|
|
"/api/v1/bootstrap/first-admin",
|
|
headers={"X-GovOPlaN-Enrollment-Token": "x" * 48},
|
|
json={
|
|
"email": "owner@example.test",
|
|
"password": "a-production-password",
|
|
},
|
|
)
|
|
|
|
assert ready.json()["state"] == "not_ready"
|
|
assert ready.json()["readiness"]["access_capability"] is False
|
|
assert rejected.status_code == 503
|