diff --git a/README.md b/README.md index 0c7d65a..b3402a2 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,7 @@ CI runs the `ci` profile in report-only mode and uploads `audit-reports/` as an artifact. Once the baseline is clean, set `SECURITY_AUDIT_FAIL_ON_FINDINGS=1` or pass `--strict` locally to turn findings into a failing gate. -`govoplan_core.devserver` enables the development bootstrap before loading settings. In dev, startup migrations create or upgrade the schema and the bootstrap creates the default development login if needed. Explicitly setting `DEV_BOOTSTRAP_ENABLED=false` disables this convenience. Production deployments should use migrations and managed database provisioning instead. +`govoplan_core.devserver` enables the development bootstrap before loading settings. In dev, startup migrations create or upgrade the schema and the bootstrap creates the default development login if needed. Explicitly setting `DEV_BOOTSTRAP_ENABLED=false` disables this convenience. Production deployments use the separate, expiring single-use flow exposed by `python -m govoplan_core.commands.first_admin`; it cannot enable or consume development bootstrap credentials. To verify the effective runtime paths and bootstrap behavior without starting uvicorn, run the smoke mode: diff --git a/alembic/dev_versions/f25c9d3e7a01_first_admin_enrollment.py b/alembic/dev_versions/f25c9d3e7a01_first_admin_enrollment.py new file mode 100644 index 0000000..6bb2b10 --- /dev/null +++ b/alembic/dev_versions/f25c9d3e7a01_first_admin_enrollment.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +from importlib.util import module_from_spec, spec_from_file_location +from pathlib import Path + + +_path = ( + Path(__file__).resolve().parents[1] + / "versions" + / "f25c9d3e7a01_first_admin_enrollment.py" +) +_spec = spec_from_file_location("govoplan_first_admin_enrollment_migration", _path) +if _spec is None or _spec.loader is None: + raise RuntimeError(f"Unable to load migration implementation from {_path}") +_module = module_from_spec(_spec) +_spec.loader.exec_module(_module) + +revision = _module.revision +down_revision = _module.down_revision +branch_labels = _module.branch_labels +depends_on = _module.depends_on +upgrade = _module.upgrade +downgrade = _module.downgrade diff --git a/alembic/env.py b/alembic/env.py index 533f0f6..c6cc472 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -12,7 +12,10 @@ except ModuleNotFoundError as exc: raise from govoplan_core.admin import models as core_admin_models # noqa: F401 - populate core admin metadata from govoplan_core.core import change_sequence as core_change_sequence_models # noqa: F401 - populate core metadata +from govoplan_core.core import first_admin as core_first_admin_models # noqa: F401 - populate core metadata from govoplan_core.core import ownership as core_ownership_models # noqa: F401 - populate core metadata +from govoplan_core.core import recovery as core_recovery_models # noqa: F401 - populate core metadata +from govoplan_core.core import runtime_coordination as core_runtime_models # noqa: F401 - populate core metadata from govoplan_core.security import credential_envelopes as core_credential_models # noqa: F401 - populate core metadata from govoplan_core.core.migrations import migration_metadata_plan from govoplan_core.db.base import Base diff --git a/alembic/versions/f25c9d3e7a01_first_admin_enrollment.py b/alembic/versions/f25c9d3e7a01_first_admin_enrollment.py new file mode 100644 index 0000000..16171a0 --- /dev/null +++ b/alembic/versions/f25c9d3e7a01_first_admin_enrollment.py @@ -0,0 +1,110 @@ +"""add controlled first-administrator enrollment evidence + +Revision ID: f25c9d3e7a01 +Revises: e14b8c2d6f90 +Create Date: 2026-08-04 00:00:00.000000 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "f25c9d3e7a01" +down_revision = "e14b8c2d6f90" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + inspector = sa.inspect(op.get_bind()) + tables = set(inspector.get_table_names()) + if "core_first_admin_enrollments" not in tables: + op.create_table( + "core_first_admin_enrollments", + sa.Column("installation_id", sa.String(length=100), nullable=False), + sa.Column("state", sa.String(length=24), nullable=False), + sa.Column("generation", sa.Integer(), nullable=False), + sa.Column("token_sha256", sa.String(length=64), nullable=True), + sa.Column("token_fingerprint", sa.String(length=16), nullable=True), + sa.Column("issued_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("consumed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("consumed_account_id", sa.String(length=36), nullable=True), + sa.Column("consumed_membership_id", sa.String(length=36), nullable=True), + sa.Column("consumed_tenant_id", sa.String(length=36), nullable=True), + sa.Column("consumed_email", sa.String(length=320), nullable=True), + sa.Column("consumed_display_name", sa.String(length=255), nullable=True), + sa.Column("consumed_request_sha256", sa.String(length=64), nullable=True), + sa.Column("issue_reason", sa.String(length=500), nullable=True), + sa.Column("event_count", sa.Integer(), nullable=False), + sa.Column("evidence_head_sha256", sa.String(length=64), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint( + "installation_id", + name=op.f("pk_core_first_admin_enrollments"), + ), + ) + op.create_index( + op.f("ix_core_first_admin_enrollments_state"), + "core_first_admin_enrollments", + ["state"], + unique=False, + ) + op.create_index( + op.f("ix_core_first_admin_enrollments_expires_at"), + "core_first_admin_enrollments", + ["expires_at"], + unique=False, + ) + + inspector = sa.inspect(op.get_bind()) + tables = set(inspector.get_table_names()) + if "core_first_admin_enrollment_events" not in tables: + op.create_table( + "core_first_admin_enrollment_events", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("installation_id", sa.String(length=100), nullable=False), + sa.Column("sequence", sa.Integer(), nullable=False), + sa.Column("event_type", sa.String(length=80), nullable=False), + sa.Column("generation", sa.Integer(), nullable=False), + sa.Column("evidence", sa.JSON(), nullable=False), + sa.Column("previous_sha256", sa.String(length=64), nullable=True), + sa.Column("event_sha256", sa.String(length=64), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["installation_id"], + ["core_first_admin_enrollments.installation_id"], + name=op.f( + "fk_core_first_admin_enrollment_events_installation_id_core_first_admin_enrollments" + ), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint( + "id", + name=op.f("pk_core_first_admin_enrollment_events"), + ), + sa.UniqueConstraint( + "installation_id", + "sequence", + name="uq_core_first_admin_enrollment_event_sequence", + ), + ) + for column in ("installation_id", "event_type", "event_sha256"): + op.create_index( + op.f(f"ix_core_first_admin_enrollment_events_{column}"), + "core_first_admin_enrollment_events", + [column], + unique=False, + ) + + +def downgrade() -> None: + inspector = sa.inspect(op.get_bind()) + tables = set(inspector.get_table_names()) + if "core_first_admin_enrollment_events" in tables: + op.drop_table("core_first_admin_enrollment_events") + if "core_first_admin_enrollments" in tables: + op.drop_table("core_first_admin_enrollments") diff --git a/docs/DEPLOYMENT_OPERATOR_GUIDE.md b/docs/DEPLOYMENT_OPERATOR_GUIDE.md index 3741fc3..d4ab2e5 100644 --- a/docs/DEPLOYMENT_OPERATOR_GUIDE.md +++ b/docs/DEPLOYMENT_OPERATOR_GUIDE.md @@ -57,6 +57,8 @@ PY | `GOVOPLAN_MIGRATION_TRACK` | `release` | Use the release track for normal runtime and deployments. Use `dev` only for fresh/disposable databases that intentionally replay detailed development migrations. | | `DEV_AUTO_MIGRATE_ENABLED` | `true` | Dev convenience only. Production should run migration commands explicitly during deployment. | | `DEV_BOOTSTRAP_ENABLED` | `false` | Dev bootstrap only. `govoplan_core.devserver` and `govoplan/tools/launch/launch-dev.sh` default it to `true`; use controlled first-admin creation outside dev. | +| `FIRST_ADMIN_ENROLLMENT_TTL_SECONDS` | `1800` | Lifetime of a locally issued production enrollment credential. Allowed range: 60 seconds to 24 hours. | +| `FIRST_ADMIN_ENROLLMENT_FILE` | `/run/govoplan/first-admin-enrollment.json` | Local operator artifact. The command creates it with mode `0600` and never prints the secret. | Operator rule: take a database backup before applying migrations or destructive module retirement. For non-SQLite databases, configure deployment-specific @@ -300,8 +302,34 @@ configuration, not the core runtime contract. Store them in a local ignored 3. Build the WebUI from `webui/package.release.json` or deploy a prebuilt artifact from the same release tag. 4. Run database migrations with the target `DATABASE_URL`. -5. Create the first tenant and system owner through the controlled bootstrap or - one-time admin command for the deployment. +5. Create the first tenant and system owner through the controlled bootstrap: + +```bash +python -m govoplan_core.commands.first_admin status +python -m govoplan_core.commands.first_admin issue \ + --reason "initial production installation" +``` + + The issue command fails when an active system administrator already exists, + writes the random credential only to `FIRST_ADMIN_ENROLLMENT_FILE`, and does + not print it. Check `GET /api/v1/bootstrap/status`, then submit the account + and initial tenant fields to `POST /api/v1/bootstrap/first-admin` with the + secret in `X-GovOPlaN-Enrollment-Token`. The operation creates the protected + system owner and initial tenant-owner membership in one transaction and + retires the credential. A repeated identical request returns the same result + without creating another owner. + + If the artifact is lost or expires before use, a local operator may rotate + it only while no durable system administrator exists: + +```bash +python -m govoplan_core.commands.first_admin recover \ + --reason "expired installation handoff" +``` + + Issue and recovery write hash-chained Core evidence and an audit event. They + never enable or reuse `DEV_BOOTSTRAP_ENABLED`, `DEV_BOOTSTRAP_PASSWORD`, or + `DEV_BOOTSTRAP_API_KEY`. 6. Start the API service with `govoplan_core.server.app:app`. 7. Start workers when `CELERY_ENABLED=true`. 8. Start the WebUI/reverse proxy and verify CORS/cookie settings. diff --git a/pyproject.toml b/pyproject.toml index f6d5f21..afaefb0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,6 +37,7 @@ govoplan_core = ["py.typed"] [project.scripts] govoplan-config = "govoplan_core.commands.config:main" govoplan-devserver = "govoplan_core.devserver:main" +govoplan-first-admin = "govoplan_core.commands.first_admin:main" govoplan-module-install-plan = "govoplan_core.commands.module_install_plan:main" govoplan-module-installer = "govoplan_core.commands.module_installer:main" diff --git a/src/govoplan_core/commands/first_admin.py b/src/govoplan_core/commands/first_admin.py new file mode 100644 index 0000000..fde45af --- /dev/null +++ b/src/govoplan_core/commands/first_admin.py @@ -0,0 +1,233 @@ +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path +import stat +from typing import Any + +from govoplan_core.core.access import ( + CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER, + FirstAdminProvisioner, +) +from govoplan_core.core.first_admin import ( + FirstAdminEnrollmentError, + first_admin_enrollment_status, + issue_first_admin_credential, +) +from govoplan_core.core.module_management import ( + load_startup_enabled_modules, + startup_candidate_module_ids, +) +from govoplan_core.core.modules import ModuleContext +from govoplan_core.core.runtime import configure_runtime +from govoplan_core.db.session import configure_database, get_database +from govoplan_core.server.registry import ( + available_module_manifests, + build_platform_registry, +) +from govoplan_core.settings import settings + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Manage the single-use production first-administrator credential", + ) + parser.add_argument( + "command", + choices=("status", "issue", "recover"), + help="Inspect readiness, issue the initial credential, or rotate lost/expired material.", + ) + parser.add_argument("--database-url", default=settings.database_url) + parser.add_argument("--installation-id", default=settings.installation_id) + parser.add_argument( + "--output", + type=Path, + default=Path(settings.first_admin_enrollment_file), + help="Root-readable/equivalent JSON credential artifact.", + ) + parser.add_argument( + "--ttl-seconds", + type=int, + default=settings.first_admin_enrollment_ttl_seconds, + ) + parser.add_argument( + "--reason", + default=None, + help="Audited local-operator reason for issue or recovery.", + ) + args = parser.parse_args() + + configure_database(args.database_url) + provisioner = _configure_first_admin_provisioner() + with get_database().SessionLocal() as session: + if args.command == "status": + enrollment = first_admin_enrollment_status( + session, + installation_id=args.installation_id, + provisioner=provisioner, + ) + print( + json.dumps( + { + "installation_id": args.installation_id, + "enrollment_required": enrollment.enrollment_required, + "credential_active": enrollment.credential_active, + "state": enrollment.state, + "generation": enrollment.generation, + "expires_at": ( + enrollment.expires_at.isoformat() + if enrollment.expires_at is not None + else None + ), + "readiness": enrollment.readiness, + }, + indent=2, + sort_keys=True, + ) + ) + return + + reason = args.reason or ( + "initial production administrator enrollment" + if args.command == "issue" + else "local operator recovery of first-administrator enrollment" + ) + try: + credential = issue_first_admin_credential( + session, + installation_id=args.installation_id, + provisioner=provisioner, + ttl_seconds=args.ttl_seconds, + reason=reason, + replace_active=args.command == "recover", + ) + payload = { + "schema_version": 1, + "installation_id": args.installation_id, + "endpoint": "/api/v1/bootstrap/first-admin", + "header": "X-GovOPlaN-Enrollment-Token", + "enrollment_token": credential.secret, + "fingerprint": credential.fingerprint, + "generation": credential.generation, + "expires_at": credential.expires_at.isoformat(), + } + previous = _secure_file_snapshot(args.output) + _write_private_json(args.output, payload) + try: + session.commit() + except Exception: + session.rollback() + _restore_secure_file(args.output, previous) + raise + except FirstAdminEnrollmentError as exc: + session.rollback() + parser.error(str(exc)) + + print(f"First-administrator credential written to {args.output}") + print(f"Fingerprint: {credential.fingerprint}") + print(f"Expires: {credential.expires_at.isoformat()}") + print("The secret was not printed. Read it from the restricted artifact on the host.") + + +def _configure_first_admin_provisioner() -> FirstAdminProvisioner: + raw_enabled = load_startup_enabled_modules(settings.enabled_modules) + candidates = startup_candidate_module_ids(settings.enabled_modules, raw_enabled) + available = available_module_manifests( + enabled_modules=candidates, + ignore_load_errors=True, + ) + enabled = load_startup_enabled_modules( + settings.enabled_modules, + available=available, + ) + registry = build_platform_registry(enabled) + context = ModuleContext(registry=registry, settings=settings) + registry.configure_capability_context(context) + configure_runtime(context) + if not registry.has_capability(CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER): + raise RuntimeError( + "Install and enable the Access module before issuing a first-administrator credential." + ) + capability = registry.require_capability(CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER) + if not isinstance(capability, FirstAdminProvisioner): + raise RuntimeError("The Access first-administrator capability is invalid.") + return capability + + +def _secure_file_snapshot(path: Path) -> tuple[bytes, int] | None: + try: + metadata = path.lstat() + except FileNotFoundError: + return None + if not stat.S_ISREG(metadata.st_mode): + raise RuntimeError(f"Refusing to replace non-regular credential artifact: {path}") + if metadata.st_uid != os.geteuid(): + raise RuntimeError(f"Credential artifact is not owned by the current operator: {path}") + if stat.S_IMODE(metadata.st_mode) & 0o077: + raise RuntimeError(f"Credential artifact permissions are too broad: {path}") + return path.read_bytes(), stat.S_IMODE(metadata.st_mode) + + +def _write_private_json(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp") + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + descriptor = os.open(temporary, flags, 0o600) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as stream: + json.dump(payload, stream, indent=2, sort_keys=True) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + os.chmod(path, 0o600) + _fsync_directory(path.parent) + except Exception: + try: + temporary.unlink() + except FileNotFoundError: + pass + raise + + +def _restore_secure_file(path: Path, snapshot: tuple[bytes, int] | None) -> None: + if snapshot is None: + try: + path.unlink() + except FileNotFoundError: + return + return + content, mode = snapshot + temporary = path.with_name(f".{path.name}.{os.getpid()}.restore") + descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + with os.fdopen(descriptor, "wb") as stream: + stream.write(content) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + os.chmod(path, mode) + _fsync_directory(path.parent) + finally: + try: + temporary.unlink() + except FileNotFoundError: + pass + + +def _fsync_directory(path: Path) -> None: + if not hasattr(os, "O_DIRECTORY"): + return + descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + +if __name__ == "__main__": + main() diff --git a/src/govoplan_core/core/access.py b/src/govoplan_core/core/access.py index 44ba31d..5be8f9d 100644 --- a/src/govoplan_core/core/access.py +++ b/src/govoplan_core/core/access.py @@ -21,6 +21,7 @@ CAPABILITY_ACCESS_RESOURCE_ACCESS = "access.resourceAccess" CAPABILITY_ACCESS_SEMANTIC_DIRECTORY = "access.semanticDirectory" CAPABILITY_ACCESS_EXPLANATION = "access.explanation" CAPABILITY_ACCESS_TENANT_PROVISIONER = "access.tenantProvisioner" +CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER = "access.firstAdminProvisioner" CAPABILITY_ACCESS_ADMINISTRATION = "access.administration" CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER = "access.governanceMaterializer" CAPABILITY_TENANCY_TENANT_RESOLVER = "tenancy.tenantResolver" @@ -45,6 +46,7 @@ ACCESS_CAPABILITY_NAMES = frozenset( CAPABILITY_ACCESS_SEMANTIC_DIRECTORY, CAPABILITY_ACCESS_EXPLANATION, CAPABILITY_ACCESS_TENANT_PROVISIONER, + CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER, CAPABILITY_ACCESS_ADMINISTRATION, CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER, CAPABILITY_TENANCY_TENANT_RESOLVER, @@ -342,6 +344,19 @@ class DevelopmentBootstrapRef: created_api_key: CreatedApiKeyRef | None = None +@dataclass(frozen=True, slots=True) +class FirstSystemAdministratorRef: + account_id: str + email: str + display_name: str | None = None + membership_id: str | None = None + tenant_id: str | None = None + + +class FirstAdminProvisioningError(RuntimeError): + """Safe, user-facing rejection from the Access enrollment boundary.""" + + @dataclass(frozen=True, slots=True) class TenantContextSwitchRef: account_id: str @@ -579,6 +594,25 @@ class TenantAccessProvisioner(Protocol): ... +@runtime_checkable +class FirstAdminProvisioner(Protocol): + """Narrow Access boundary used only by the production bootstrap flow.""" + + def has_durable_system_administrator(self, session: object) -> bool: + ... + + def create_first_system_administrator( + self, + session: object, + *, + tenant: object, + email: str, + display_name: str | None, + password: str, + ) -> FirstSystemAdministratorRef: + ... + + @runtime_checkable class AccessAdministration(Protocol): def tenant_counts(self, session: object, tenant_id: str) -> Mapping[str, int]: diff --git a/src/govoplan_core/core/first_admin.py b/src/govoplan_core/core/first_admin.py new file mode 100644 index 0000000..b9b25f6 --- /dev/null +++ b/src/govoplan_core/core/first_admin.py @@ -0,0 +1,532 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from enum import StrEnum +import hashlib +import hmac +import json +import re +import secrets +from typing import Any +from uuid import uuid4 + +from sqlalchemy import DateTime, ForeignKey, Integer, JSON, String, UniqueConstraint, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Mapped, Session, mapped_column + +from govoplan_core.audit.logging import audit_event +from govoplan_core.core.access import ( + FirstAdminProvisioner, + FirstAdminProvisioningError, + FirstSystemAdministratorRef, +) +from govoplan_core.db.base import Base, TimestampMixin, utcnow +from govoplan_core.tenancy.scope import Tenant + + +_TENANT_SLUG_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") + + +class FirstAdminEnrollmentState(StrEnum): + INACTIVE = "inactive" + ACTIVE = "active" + CONSUMED = "consumed" + REVOKED = "revoked" + + +class FirstAdminEnrollmentError(RuntimeError): + pass + + +class FirstAdminEnrollmentUnavailable(FirstAdminEnrollmentError): + pass + + +class FirstAdminEnrollmentCredentialError(FirstAdminEnrollmentError): + pass + + +class FirstAdminEnrollmentConflict(FirstAdminEnrollmentError): + pass + + +class FirstAdminEnrollment(Base, TimestampMixin): + __tablename__ = "core_first_admin_enrollments" + + installation_id: Mapped[str] = mapped_column(String(100), primary_key=True) + state: Mapped[str] = mapped_column( + String(24), + default=FirstAdminEnrollmentState.INACTIVE.value, + nullable=False, + index=True, + ) + generation: Mapped[int] = mapped_column(Integer, default=0, nullable=False) + token_sha256: Mapped[str | None] = mapped_column(String(64)) + token_fingerprint: Mapped[str | None] = mapped_column(String(16)) + issued_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), index=True) + consumed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + consumed_account_id: Mapped[str | None] = mapped_column(String(36)) + consumed_membership_id: Mapped[str | None] = mapped_column(String(36)) + consumed_tenant_id: Mapped[str | None] = mapped_column(String(36)) + consumed_email: Mapped[str | None] = mapped_column(String(320)) + consumed_display_name: Mapped[str | None] = mapped_column(String(255)) + consumed_request_sha256: Mapped[str | None] = mapped_column(String(64)) + issue_reason: Mapped[str | None] = mapped_column(String(500)) + event_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) + evidence_head_sha256: Mapped[str | None] = mapped_column(String(64)) + + +class FirstAdminEnrollmentEvent(Base): + __tablename__ = "core_first_admin_enrollment_events" + __table_args__ = ( + UniqueConstraint( + "installation_id", + "sequence", + name="uq_core_first_admin_enrollment_event_sequence", + ), + ) + + id: Mapped[str] = mapped_column( + String(36), + primary_key=True, + default=lambda: str(uuid4()), + ) + installation_id: Mapped[str] = mapped_column( + ForeignKey( + "core_first_admin_enrollments.installation_id", + ondelete="CASCADE", + ), + nullable=False, + index=True, + ) + sequence: Mapped[int] = mapped_column(Integer, nullable=False) + event_type: Mapped[str] = mapped_column(String(80), nullable=False, index=True) + generation: Mapped[int] = mapped_column(Integer, nullable=False) + evidence: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False) + previous_sha256: Mapped[str | None] = mapped_column(String(64)) + event_sha256: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + default=utcnow, + nullable=False, + ) + + +@dataclass(frozen=True, slots=True) +class IssuedFirstAdminCredential: + secret: str + fingerprint: str + generation: int + expires_at: datetime + + +@dataclass(frozen=True, slots=True) +class FirstAdminEnrollmentStatus: + enrollment_required: bool + credential_active: bool + state: str + generation: int + expires_at: datetime | None + completed_account_id: str | None + readiness: dict[str, bool] + + +@dataclass(frozen=True, slots=True) +class FirstAdminEnrollmentResult: + administrator: FirstSystemAdministratorRef + replayed: bool + + +def issue_first_admin_credential( + session: Session, + *, + installation_id: str, + provisioner: FirstAdminProvisioner, + ttl_seconds: int, + reason: str, + replace_active: bool = False, + now: datetime | None = None, +) -> IssuedFirstAdminCredential: + current_time = _utc(now) + if ttl_seconds < 60 or ttl_seconds > 24 * 60 * 60: + raise ValueError("First-admin enrollment expiry must be between 60 seconds and 24 hours.") + if provisioner.has_durable_system_administrator(session): + raise FirstAdminEnrollmentUnavailable( + "A durable system administrator already exists. Bootstrap enrollment is disabled." + ) + + enrollment = _locked_enrollment(session, installation_id) + if ( + enrollment.state == FirstAdminEnrollmentState.ACTIVE.value + and _is_future(enrollment.expires_at, current_time) + and not replace_active + ): + raise FirstAdminEnrollmentConflict( + "An unexpired first-admin credential already exists. Use the recovery command to rotate it." + ) + + secret = secrets.token_urlsafe(48) + token_sha256 = _secret_sha256(secret) + fingerprint = token_sha256[:12] + expires_at = current_time + timedelta(seconds=ttl_seconds) + generation = enrollment.generation + 1 + if enrollment.state == FirstAdminEnrollmentState.ACTIVE.value: + _append_event( + session, + enrollment, + event_type="credential_revoked", + generation=enrollment.generation, + created_at=current_time, + evidence={"reason": "local_operator_recovery"}, + ) + enrollment.state = FirstAdminEnrollmentState.ACTIVE.value + enrollment.generation = generation + enrollment.token_sha256 = token_sha256 + enrollment.token_fingerprint = fingerprint + enrollment.issued_at = current_time + enrollment.expires_at = expires_at + enrollment.consumed_at = None + enrollment.consumed_account_id = None + enrollment.consumed_membership_id = None + enrollment.consumed_tenant_id = None + enrollment.consumed_email = None + enrollment.consumed_display_name = None + enrollment.consumed_request_sha256 = None + enrollment.issue_reason = _bounded_reason(reason) + session.add(enrollment) + _append_event( + session, + enrollment, + event_type="credential_issued", + generation=generation, + created_at=current_time, + evidence={ + "fingerprint": fingerprint, + "expires_at": expires_at.isoformat(), + "reason": enrollment.issue_reason, + }, + ) + audit_event( + session, + tenant_id=None, + scope="system", + action="access.first_admin_enrollment.issued", + object_type="first_admin_enrollment", + object_id=installation_id, + details={ + "generation": generation, + "fingerprint": fingerprint, + "expires_at": expires_at.isoformat(), + "reason": enrollment.issue_reason, + }, + ) + return IssuedFirstAdminCredential( + secret=secret, + fingerprint=fingerprint, + generation=generation, + expires_at=expires_at, + ) + + +def first_admin_enrollment_status( + session: Session, + *, + installation_id: str, + provisioner: FirstAdminProvisioner, + now: datetime | None = None, +) -> FirstAdminEnrollmentStatus: + current_time = _utc(now) + administrator_exists = provisioner.has_durable_system_administrator(session) + enrollment = session.get(FirstAdminEnrollment, installation_id) + state = enrollment.state if enrollment is not None else FirstAdminEnrollmentState.INACTIVE.value + active = bool( + not administrator_exists + and enrollment is not None + and state == FirstAdminEnrollmentState.ACTIVE.value + and enrollment.token_sha256 + and _is_future(enrollment.expires_at, current_time) + ) + if ( + not administrator_exists + and enrollment is not None + and state == FirstAdminEnrollmentState.ACTIVE.value + and not active + ): + state = "expired" + return FirstAdminEnrollmentStatus( + enrollment_required=not administrator_exists, + credential_active=active, + state="completed" if administrator_exists else state, + generation=enrollment.generation if enrollment is not None else 0, + expires_at=enrollment.expires_at if enrollment is not None else None, + completed_account_id=( + enrollment.consumed_account_id if enrollment is not None else None + ), + readiness={ + "database": True, + "access_capability": True, + "administrator_absent": not administrator_exists, + }, + ) + + +def consume_first_admin_credential( + session: Session, + *, + installation_id: str, + provisioner: FirstAdminProvisioner, + secret: str, + email: str, + display_name: str | None, + password: str, + tenant_slug: str, + tenant_name: str, + now: datetime | None = None, +) -> FirstAdminEnrollmentResult: + current_time = _utc(now) + normalized_email = email.strip().casefold() + clean_display_name = display_name.strip() if display_name and display_name.strip() else None + clean_tenant_slug = tenant_slug.strip().casefold() + clean_tenant_name = tenant_name.strip() + if not normalized_email or "@" not in normalized_email: + raise FirstAdminEnrollmentConflict("Enter a valid administrator email address.") + if len(password) < 12: + raise FirstAdminEnrollmentConflict("The administrator password must contain at least 12 characters.") + if not _TENANT_SLUG_RE.fullmatch(clean_tenant_slug): + raise FirstAdminEnrollmentConflict( + "The initial tenant slug may contain lowercase letters, numbers, and single hyphens." + ) + if not clean_tenant_name: + raise FirstAdminEnrollmentConflict("Enter a name for the initial tenant.") + + request_sha256 = _request_sha256( + email=normalized_email, + display_name=clean_display_name, + tenant_slug=clean_tenant_slug, + tenant_name=clean_tenant_name, + ) + supplied_sha256 = _secret_sha256(secret) + enrollment = session.execute( + select(FirstAdminEnrollment) + .where(FirstAdminEnrollment.installation_id == installation_id) + .with_for_update() + ).scalar_one_or_none() + if enrollment is None: + raise FirstAdminEnrollmentCredentialError("First-admin enrollment is not active.") + + if enrollment.state == FirstAdminEnrollmentState.CONSUMED.value: + if ( + enrollment.token_sha256 + and hmac.compare_digest(enrollment.token_sha256, supplied_sha256) + and enrollment.consumed_request_sha256 == request_sha256 + and enrollment.consumed_account_id + and enrollment.consumed_email + ): + return FirstAdminEnrollmentResult( + administrator=FirstSystemAdministratorRef( + account_id=enrollment.consumed_account_id, + email=enrollment.consumed_email, + display_name=enrollment.consumed_display_name, + membership_id=enrollment.consumed_membership_id, + tenant_id=enrollment.consumed_tenant_id, + ), + replayed=True, + ) + raise FirstAdminEnrollmentCredentialError("The first-admin credential has already been used.") + + if enrollment.state != FirstAdminEnrollmentState.ACTIVE.value or not enrollment.token_sha256: + raise FirstAdminEnrollmentCredentialError("First-admin enrollment is not active.") + if not _is_future(enrollment.expires_at, current_time): + raise FirstAdminEnrollmentCredentialError( + "The first-admin credential has expired. A local operator must issue a replacement." + ) + if not hmac.compare_digest(enrollment.token_sha256, supplied_sha256): + raise FirstAdminEnrollmentCredentialError("The first-admin credential is invalid.") + if provisioner.has_durable_system_administrator(session): + raise FirstAdminEnrollmentUnavailable( + "A durable system administrator already exists. Bootstrap enrollment is disabled." + ) + + tenant = session.execute( + select(Tenant).where(Tenant.slug == clean_tenant_slug).with_for_update() + ).scalar_one_or_none() + if tenant is None: + tenant = Tenant( + slug=clean_tenant_slug, + name=clean_tenant_name, + default_locale="en", + settings={}, + is_active=True, + ) + session.add(tenant) + session.flush() + elif not tenant.is_active: + raise FirstAdminEnrollmentConflict("The selected initial tenant is inactive.") + + try: + administrator = provisioner.create_first_system_administrator( + session, + tenant=tenant, + email=normalized_email, + display_name=clean_display_name, + password=password, + ) + except FirstAdminProvisioningError as exc: + raise FirstAdminEnrollmentConflict(str(exc)) from exc + enrollment.state = FirstAdminEnrollmentState.CONSUMED.value + enrollment.consumed_at = current_time + enrollment.consumed_account_id = administrator.account_id + enrollment.consumed_membership_id = administrator.membership_id + enrollment.consumed_tenant_id = administrator.tenant_id + enrollment.consumed_email = administrator.email + enrollment.consumed_display_name = administrator.display_name + enrollment.consumed_request_sha256 = request_sha256 + session.add(enrollment) + _append_event( + session, + enrollment, + event_type="administrator_created", + generation=enrollment.generation, + created_at=current_time, + evidence={ + "account_id": administrator.account_id, + "membership_id": administrator.membership_id, + "tenant_id": administrator.tenant_id, + "email_sha256": hashlib.sha256(normalized_email.encode("utf-8")).hexdigest(), + }, + ) + audit_event( + session, + tenant_id=None, + scope="system", + action="access.first_admin_enrollment.completed", + object_type="access_account", + object_id=administrator.account_id, + details={ + "generation": enrollment.generation, + "membership_id": administrator.membership_id, + "tenant_id": administrator.tenant_id, + "credential_invalidated": True, + }, + ) + return FirstAdminEnrollmentResult(administrator=administrator, replayed=False) + + +def _locked_enrollment(session: Session, installation_id: str) -> FirstAdminEnrollment: + enrollment = session.execute( + select(FirstAdminEnrollment) + .where(FirstAdminEnrollment.installation_id == installation_id) + .with_for_update() + ).scalar_one_or_none() + if enrollment is not None: + return enrollment + enrollment = FirstAdminEnrollment(installation_id=installation_id) + try: + with session.begin_nested(): + session.add(enrollment) + session.flush() + except IntegrityError: + enrollment = session.execute( + select(FirstAdminEnrollment) + .where(FirstAdminEnrollment.installation_id == installation_id) + .with_for_update() + ).scalar_one() + return enrollment + + +def _append_event( + session: Session, + enrollment: FirstAdminEnrollment, + *, + event_type: str, + generation: int, + created_at: datetime, + evidence: dict[str, Any], +) -> None: + sequence = enrollment.event_count + 1 + payload = { + "installation_id": enrollment.installation_id, + "sequence": sequence, + "event_type": event_type, + "generation": generation, + "created_at": created_at.isoformat(), + "evidence": evidence, + "previous_sha256": enrollment.evidence_head_sha256, + } + event_sha256 = hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + session.add( + FirstAdminEnrollmentEvent( + installation_id=enrollment.installation_id, + sequence=sequence, + event_type=event_type, + generation=generation, + evidence=evidence, + previous_sha256=enrollment.evidence_head_sha256, + event_sha256=event_sha256, + created_at=created_at, + ) + ) + enrollment.event_count = sequence + enrollment.evidence_head_sha256 = event_sha256 + session.add(enrollment) + + +def _request_sha256( + *, + email: str, + display_name: str | None, + tenant_slug: str, + tenant_name: str, +) -> str: + payload = { + "email": email, + "display_name": display_name, + "tenant_slug": tenant_slug, + "tenant_name": tenant_name, + } + return hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + + +def _secret_sha256(secret: str) -> str: + return hashlib.sha256(secret.encode("utf-8")).hexdigest() + + +def _utc(value: datetime | None) -> datetime: + candidate = value or datetime.now(timezone.utc) + if candidate.tzinfo is None: + return candidate.replace(tzinfo=timezone.utc) + return candidate.astimezone(timezone.utc) + + +def _is_future(value: datetime | None, now: datetime) -> bool: + return value is not None and _utc(value) > now + + +def _bounded_reason(value: str) -> str: + clean = value.strip() + if not clean: + raise ValueError("A local operator reason is required.") + return clean[:500] + + +__all__ = [ + "FirstAdminEnrollment", + "FirstAdminEnrollmentConflict", + "FirstAdminEnrollmentCredentialError", + "FirstAdminEnrollmentError", + "FirstAdminEnrollmentEvent", + "FirstAdminEnrollmentResult", + "FirstAdminEnrollmentState", + "FirstAdminEnrollmentStatus", + "FirstAdminEnrollmentUnavailable", + "IssuedFirstAdminCredential", + "consume_first_admin_credential", + "first_admin_enrollment_status", + "issue_first_admin_credential", +] diff --git a/src/govoplan_core/db/bootstrap.py b/src/govoplan_core/db/bootstrap.py index 9959517..c6484a2 100644 --- a/src/govoplan_core/db/bootstrap.py +++ b/src/govoplan_core/db/bootstrap.py @@ -32,6 +32,7 @@ def create_all_tables() -> None: # model metadata with the shared SQLAlchemy base before create_all runs. from govoplan_core.admin import models as core_admin_models # noqa: F401 from govoplan_core.core import change_sequence as core_change_sequence_models # noqa: F401 + from govoplan_core.core import first_admin as core_first_admin_models # noqa: F401 from govoplan_core.core import recovery as core_recovery_models # noqa: F401 from govoplan_core.core import runtime_coordination as core_runtime_models # noqa: F401 from govoplan_core.security import credential_envelopes as core_credential_models # noqa: F401 diff --git a/src/govoplan_core/db/migrations.py b/src/govoplan_core/db/migrations.py index f7b610a..9bfd0de 100644 --- a/src/govoplan_core/db/migrations.py +++ b/src/govoplan_core/db/migrations.py @@ -19,6 +19,7 @@ from sqlalchemy import create_engine, inspect, text from govoplan_core.core.migrations import MigrationMetadataPlan, migration_metadata_plan from govoplan_core.core import change_sequence as core_change_sequence_models # noqa: F401 - populate core metadata +from govoplan_core.core import first_admin as core_first_admin_models # noqa: F401 - populate core metadata from govoplan_core.core import recovery as core_recovery_models # noqa: F401 - populate core metadata from govoplan_core.core import runtime_coordination as core_runtime_models # noqa: F401 - populate core metadata from govoplan_core.security import credential_envelopes as core_credential_models # noqa: F401 - populate core metadata diff --git a/src/govoplan_core/server/app.py b/src/govoplan_core/server/app.py index 26453a7..22dd7c2 100644 --- a/src/govoplan_core/server/app.py +++ b/src/govoplan_core/server/app.py @@ -8,6 +8,7 @@ from govoplan_core.db.session import configure_database from govoplan_core.server.config import GovoplanServerConfig, load_server_config from govoplan_core.server.fastapi import create_govoplan_app from govoplan_core.server.platform import create_platform_router +from govoplan_core.server.bootstrap import create_bootstrap_router from govoplan_core.server.credentials import router as credential_router from govoplan_core.server.ownership import router as ownership_router from govoplan_core.server.registry import available_module_manifests, build_platform_registry @@ -69,6 +70,7 @@ def _server_api_router(server_config: GovoplanServerConfig, registry) -> APIRout for router in server_config.base_routers: api_router.include_router(router) api_router.include_router(create_platform_router(settings=server_config.settings)) + api_router.include_router(create_bootstrap_router(server_config.settings)) api_router.include_router(credential_router) api_router.include_router(ownership_router) for router in server_config.post_module_routers: diff --git a/src/govoplan_core/server/bootstrap.py b/src/govoplan_core/server/bootstrap.py new file mode 100644 index 0000000..e03be4a --- /dev/null +++ b/src/govoplan_core/server/bootstrap.py @@ -0,0 +1,171 @@ +from __future__ import annotations + +from datetime import datetime + +from fastapi import APIRouter, Depends, Header, HTTPException, Request, status +from pydantic import BaseModel, Field, SecretStr +from sqlalchemy.orm import Session + +from govoplan_core.core.access import ( + CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER, + FirstAdminProvisioner, +) +from govoplan_core.core.first_admin import ( + FirstAdminEnrollmentConflict, + FirstAdminEnrollmentCredentialError, + FirstAdminEnrollmentUnavailable, + consume_first_admin_credential, + first_admin_enrollment_status, +) +from govoplan_core.core.registry import PlatformRegistry +from govoplan_core.db.session import get_session + + +class FirstAdminReadinessResponse(BaseModel): + enrollment_required: bool + credential_active: bool + state: str + generation: int = 0 + expires_at: datetime | None = None + readiness: dict[str, bool] = Field(default_factory=dict) + + +class FirstAdminEnrollmentRequest(BaseModel): + email: str = Field(min_length=3, max_length=320) + display_name: str | None = Field(default=None, max_length=255) + password: SecretStr = Field(min_length=12, max_length=1024) + tenant_slug: str = Field(default="default", min_length=1, max_length=100) + tenant_name: str = Field(default="Default Tenant", min_length=1, max_length=255) + + +class FirstAdminEnrollmentResponse(BaseModel): + account_id: str + membership_id: str | None = None + tenant_id: str | None = None + email: str + display_name: str | None = None + replayed: bool = False + bootstrap_retired: bool = True + + +def create_bootstrap_router(settings: object) -> APIRouter: + router = APIRouter(prefix="/bootstrap", tags=["bootstrap"]) + + @router.get("/status", response_model=FirstAdminReadinessResponse) + def bootstrap_status( + request: Request, + session: Session = Depends(get_session), + ) -> FirstAdminReadinessResponse: + provisioner = _first_admin_provisioner(request, required=False) + if provisioner is None: + return FirstAdminReadinessResponse( + enrollment_required=False, + credential_active=False, + state="not_ready", + readiness={ + "database": True, + "access_capability": False, + "administrator_absent": False, + }, + ) + enrollment = first_admin_enrollment_status( + session, + installation_id=str(getattr(settings, "installation_id", "govoplan-local")), + provisioner=provisioner, + ) + return FirstAdminReadinessResponse( + enrollment_required=enrollment.enrollment_required, + credential_active=enrollment.credential_active, + state=enrollment.state, + generation=enrollment.generation, + expires_at=enrollment.expires_at, + readiness=enrollment.readiness, + ) + + @router.post( + "/first-admin", + response_model=FirstAdminEnrollmentResponse, + status_code=status.HTTP_201_CREATED, + ) + def enroll_first_admin( + payload: FirstAdminEnrollmentRequest, + request: Request, + x_govoplan_enrollment_token: str = Header( + min_length=32, + max_length=512, + alias="X-GovOPlaN-Enrollment-Token", + ), + session: Session = Depends(get_session), + ) -> FirstAdminEnrollmentResponse: + provisioner = _first_admin_provisioner(request, required=True) + assert provisioner is not None + try: + result = consume_first_admin_credential( + session, + installation_id=str(getattr(settings, "installation_id", "govoplan-local")), + provisioner=provisioner, + secret=x_govoplan_enrollment_token, + email=payload.email, + display_name=payload.display_name, + password=payload.password.get_secret_value(), + tenant_slug=payload.tenant_slug, + tenant_name=payload.tenant_name, + ) + session.commit() + except FirstAdminEnrollmentCredentialError as exc: + session.rollback() + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc + except FirstAdminEnrollmentUnavailable as exc: + session.rollback() + raise HTTPException(status_code=status.HTTP_410_GONE, detail=str(exc)) from exc + except FirstAdminEnrollmentConflict as exc: + session.rollback() + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc + administrator = result.administrator + return FirstAdminEnrollmentResponse( + account_id=administrator.account_id, + membership_id=administrator.membership_id, + tenant_id=administrator.tenant_id, + email=administrator.email, + display_name=administrator.display_name, + replayed=result.replayed, + ) + + return router + + +def _first_admin_provisioner( + request: Request, + *, + required: bool, +) -> FirstAdminProvisioner | None: + registry = getattr(request.app.state, "govoplan_registry", None) + if not isinstance(registry, PlatformRegistry): + if required: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="The module registry is not ready.", + ) + return None + if not registry.has_capability(CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER): + if required: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Install and enable the Access module before enrolling the first administrator.", + ) + return None + capability = registry.require_capability(CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER) + if not isinstance(capability, FirstAdminProvisioner): + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="The Access first-administrator capability is invalid.", + ) + return capability + + +__all__ = [ + "FirstAdminEnrollmentRequest", + "FirstAdminEnrollmentResponse", + "FirstAdminReadinessResponse", + "create_bootstrap_router", +] diff --git a/src/govoplan_core/settings.py b/src/govoplan_core/settings.py index 6cb0ec6..3e35647 100644 --- a/src/govoplan_core/settings.py +++ b/src/govoplan_core/settings.py @@ -271,6 +271,19 @@ class Settings(BaseSettings): dev_bootstrap_password: str = Field(default="dev-admin", alias="DEV_BOOTSTRAP_PASSWORD") dev_mailbox_api_enabled: bool = Field(default=False, alias="DEV_MAILBOX_API_ENABLED") + # Production first-administrator enrollment. The credential is issued only + # by the local operator command and is unrelated to development bootstrap. + first_admin_enrollment_ttl_seconds: int = Field( + default=30 * 60, + ge=60, + le=24 * 60 * 60, + alias="FIRST_ADMIN_ENROLLMENT_TTL_SECONDS", + ) + first_admin_enrollment_file: str = Field( + default="/run/govoplan/first-admin-enrollment.json", + alias="FIRST_ADMIN_ENROLLMENT_FILE", + ) + # Comma-separated list. Use * only for local development. cors_origins: str = Field(default="http://localhost:5173,http://127.0.0.1:5173,http://localhost:8080", alias="CORS_ORIGINS") diff --git a/tests/test_first_admin_enrollment.py b/tests/test_first_admin_enrollment.py new file mode 100644 index 0000000..ad39af8 --- /dev/null +++ b/tests/test_first_admin_enrollment.py @@ -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