Add controlled first-admin enrollment
This commit is contained in:
@@ -0,0 +1,386 @@
|
||||
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
|
||||
Reference in New Issue
Block a user