From 668f7cf1080fe110ded507baba0142899804d7a7 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 4 Aug 2026 10:07:20 +0200 Subject: [PATCH] Provide first system administrator authority --- src/govoplan_access/backend/manifest.py | 42 +++++++ .../backend/tenancy/provisioning.py | 116 +++++++++++++++++- tests/test_first_admin_provisioning.py | 107 ++++++++++++++++ 3 files changed, 263 insertions(+), 2 deletions(-) create mode 100644 tests/test_first_admin_provisioning.py diff --git a/src/govoplan_access/backend/manifest.py b/src/govoplan_access/backend/manifest.py index c9556cd..89e4b2d 100644 --- a/src/govoplan_access/backend/manifest.py +++ b/src/govoplan_access/backend/manifest.py @@ -9,6 +9,7 @@ from govoplan_core.core.access import ( CAPABILITY_ACCESS_ADMINISTRATION, CAPABILITY_ACCESS_DIRECTORY, CAPABILITY_ACCESS_EXPLANATION, + CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER, CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER, CAPABILITY_ACCESS_PERMISSION_EVALUATOR, CAPABILITY_ACCESS_PRINCIPAL_RESOLVER, @@ -269,6 +270,39 @@ ADMIN_READ_SCOPES = ( ) ACCESS_DOCUMENTATION: tuple[DocumentationTopic, ...] = ( + DocumentationTopic( + id="access.operator.enroll-first-administrator", + title="Enroll the first production administrator", + summary="A local operator can issue one expiring credential while no durable system administrator exists, then use it once to create the protected system owner and an initial tenant membership.", + body=( + "Run the Core first-admin issue command after migrations and after Access is installed. The command stores the random secret in a local mode-0600 artifact and prints only its path, fingerprint, and expiry. " + "The public bootstrap status endpoint exposes only minimum readiness. The enrollment endpoint accepts only the first account and initial tenant fields, creates one protected system owner plus one tenant-owner membership atomically, and retires the credential. " + "Identical retries return the completed account without creating another owner. Lost or expired material can be rotated only by the local recovery command and only while the durable-administrator check remains empty. Development bootstrap settings are a separate dev-only path and are never enabled by enrollment." + ), + layer="always", + documentation_types=("admin",), + audience=("operator", "system_admin"), + order=10, + links=( + DocumentationLink(label="Bootstrap readiness API", href="/api/v1/bootstrap/status", kind="api"), + DocumentationLink(label="First-administrator enrollment API", href="/api/v1/bootstrap/first-admin", kind="api"), + ), + metadata={ + "kind": "operator_workflow", + "commands": [ + "python -m govoplan_core.commands.first_admin status", + "python -m govoplan_core.commands.first_admin issue --reason 'initial production installation'", + "python -m govoplan_core.commands.first_admin recover --reason 'lost or expired handoff'", + ], + "security_properties": [ + "random expiring credential", + "mode-0600 local artifact", + "single-use idempotent enrollment", + "empty-install authority gate", + "hash-chained and audit evidence", + ], + }, + ), DocumentationTopic( id="access.workflow.grant-user-access", title="Grant a person access", @@ -670,6 +704,13 @@ def _tenant_provisioner(context: ModuleContext) -> object: return LegacyTenantAccessProvisioner() +def _first_admin_provisioner(context: ModuleContext) -> object: + del context + from govoplan_access.backend.tenancy.provisioning import LegacyFirstAdminProvisioner + + return LegacyFirstAdminProvisioner() + + def _access_administration(context: ModuleContext) -> object: del context from govoplan_access.backend.administration import SqlAccessAdministration @@ -801,6 +842,7 @@ manifest = ModuleManifest( CAPABILITY_ACCESS_SEMANTIC_DIRECTORY: _access_semantic_directory, CAPABILITY_ACCESS_EXPLANATION: _access_explanation_service, CAPABILITY_ACCESS_TENANT_PROVISIONER: _tenant_provisioner, + CAPABILITY_ACCESS_FIRST_ADMIN_PROVISIONER: _first_admin_provisioner, CAPABILITY_ACCESS_ADMINISTRATION: _access_administration, CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER: _governance_materializer, CAPABILITY_ACCESS_PEOPLE_SEARCH: _people_search, diff --git a/src/govoplan_access/backend/tenancy/provisioning.py b/src/govoplan_access/backend/tenancy/provisioning.py index 5704853..a0ad3e0 100644 --- a/src/govoplan_access/backend/tenancy/provisioning.py +++ b/src/govoplan_access/backend/tenancy/provisioning.py @@ -5,11 +5,21 @@ from collections.abc import Mapping, Sequence from sqlalchemy.orm import Session from govoplan_access.backend.admin.service import ensure_default_roles, get_or_create_account -from govoplan_access.backend.db.models import Account, SystemRoleAssignment, User, UserRoleAssignment +from govoplan_access.backend.db.models import Account, Role, SystemRoleAssignment, User, UserRoleAssignment +from govoplan_access.backend.permissions.catalog import normalize_email, scopes_grant from govoplan_access.backend.security.api_keys import create_api_key from govoplan_access.backend.security.passwords import hash_password from govoplan_core.admin.common import AdminValidationError -from govoplan_core.core.access import CreatedApiKeyRef, DevelopmentBootstrapRef, TenantAccessProvisioner, TenantOwnerCandidateRef, UserRef +from govoplan_core.core.access import ( + CreatedApiKeyRef, + DevelopmentBootstrapRef, + FirstAdminProvisioner, + FirstAdminProvisioningError, + FirstSystemAdministratorRef, + TenantAccessProvisioner, + TenantOwnerCandidateRef, + UserRef, +) class LegacyTenantAccessProvisioner(TenantAccessProvisioner): @@ -169,6 +179,108 @@ class LegacyTenantAccessProvisioner(TenantAccessProvisioner): ) +class LegacyFirstAdminProvisioner(FirstAdminProvisioner): + def has_durable_system_administrator(self, session: object) -> bool: + db = _session(session) + roles = ( + db.query(Role) + .join(SystemRoleAssignment, SystemRoleAssignment.role_id == Role.id) + .join(Account, Account.id == SystemRoleAssignment.account_id) + .filter(Role.tenant_id.is_(None), Account.is_active.is_(True)) + .all() + ) + return any( + role.slug in {"system_owner", "system_admin"} + or scopes_grant(role.permissions or (), "access:system_setting:write") + for role in roles + ) + + def create_first_system_administrator( + self, + session: object, + *, + tenant: object, + email: str, + display_name: str | None, + password: str, + ) -> FirstSystemAdministratorRef: + db = _session(session) + tenant_id = getattr(tenant, "id", None) + if not tenant_id: + raise FirstAdminProvisioningError( + "First-administrator enrollment requires a persisted initial tenant." + ) + if len(password) < 12: + raise FirstAdminProvisioningError( + "The administrator password must contain at least 12 characters." + ) + if self.has_durable_system_administrator(db): + raise FirstAdminProvisioningError( + "A durable system administrator already exists." + ) + + normalized_email = normalize_email(email) + if not normalized_email or "@" not in normalized_email: + raise FirstAdminProvisioningError("Enter a valid administrator email address.") + existing = ( + db.query(Account) + .filter(Account.normalized_email == normalized_email) + .with_for_update() + .one_or_none() + ) + if existing is not None: + raise FirstAdminProvisioningError( + "The enrollment email already belongs to an account. Use a new address for the first system owner." + ) + + tenant_roles = ensure_default_roles(db, tenant) # type: ignore[arg-type] + system_roles = ensure_default_roles(db, None) + account, created, _temporary_password = get_or_create_account( + db, + email=email, + display_name=display_name, + password=password, + password_reset_required=False, + ) + if not created: + raise FirstAdminProvisioningError( + "The enrollment email already belongs to an account." + ) + membership = User( + tenant_id=tenant_id, + account_id=account.id, + email=account.email, + display_name=display_name or account.display_name, + is_active=True, + is_tenant_admin=True, + auth_provider=account.auth_provider, + password_hash=account.password_hash, + ) + db.add(membership) + db.flush() + db.add( + UserRoleAssignment( + tenant_id=tenant_id, + user_id=membership.id, + role_id=tenant_roles["owner"].id, + ) + ) + db.add( + SystemRoleAssignment( + account_id=account.id, + role_id=system_roles["system_owner"].id, + ) + ) + db.flush() + return FirstSystemAdministratorRef( + account_id=account.id, + email=account.email, + display_name=account.display_name, + membership_id=membership.id, + tenant_id=tenant_id, + ) + + def _session(session: object) -> Session: if not isinstance(session, Session): raise TypeError("Tenant access provisioner requires a SQLAlchemy Session") diff --git a/tests/test_first_admin_provisioning.py b/tests/test_first_admin_provisioning.py new file mode 100644 index 0000000..186b0fe --- /dev/null +++ b/tests/test_first_admin_provisioning.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from govoplan_access.backend.db.base import AccessBase +from govoplan_access.backend.db.models import ( + Account, + Role, + SystemRoleAssignment, + User, + UserRoleAssignment, +) +from govoplan_access.backend.security.passwords import verify_password +from govoplan_access.backend.tenancy.provisioning import LegacyFirstAdminProvisioner +from govoplan_core.core.access import FirstAdminProvisioningError +from govoplan_core.tenancy.scope import Tenant, create_scope_tables, scope_registry + + +class FirstAdminProvisioningTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + create_scope_tables(self.engine) + AccessBase.metadata.create_all(bind=self.engine) + self.Session = sessionmaker(bind=self.engine, expire_on_commit=False) + self.session = self.Session() + self.tenant = Tenant(id="tenant-1", slug="default", name="Default Tenant") + self.session.add(self.tenant) + self.session.flush() + self.provisioner = LegacyFirstAdminProvisioner() + + def tearDown(self) -> None: + self.session.close() + AccessBase.metadata.drop_all(bind=self.engine) + scope_registry.metadata.drop_all(bind=self.engine) + self.engine.dispose() + + def test_creates_one_system_owner_with_a_login_membership(self) -> None: + created = self.provisioner.create_first_system_administrator( + self.session, + tenant=self.tenant, + email="Owner@Example.test", + display_name="System Owner", + password="a-production-password", + ) + self.session.flush() + + account = self.session.get(Account, created.account_id) + membership = self.session.get(User, created.membership_id) + self.assertIsNotNone(account) + self.assertIsNotNone(membership) + assert account is not None + assert membership is not None + self.assertTrue(verify_password("a-production-password", account.password_hash)) + self.assertEqual(membership.tenant_id, self.tenant.id) + self.assertTrue(membership.is_tenant_admin) + system_role = ( + self.session.query(Role) + .join(SystemRoleAssignment, SystemRoleAssignment.role_id == Role.id) + .filter(SystemRoleAssignment.account_id == account.id) + .one() + ) + tenant_role = ( + self.session.query(Role) + .join(UserRoleAssignment, UserRoleAssignment.role_id == Role.id) + .filter(UserRoleAssignment.user_id == membership.id) + .one() + ) + self.assertEqual(system_role.slug, "system_owner") + self.assertEqual(tenant_role.slug, "owner") + self.assertTrue( + self.provisioner.has_durable_system_administrator(self.session) + ) + + with self.assertRaisesRegex(FirstAdminProvisioningError, "already exists"): + self.provisioner.create_first_system_administrator( + self.session, + tenant=self.tenant, + email="second@example.test", + display_name=None, + password="another-production-password", + ) + + def test_refuses_to_promote_or_reset_an_existing_account(self) -> None: + self.session.add( + Account( + email="existing@example.test", + normalized_email="existing@example.test", + is_active=True, + ) + ) + self.session.flush() + + with self.assertRaisesRegex(FirstAdminProvisioningError, "already belongs"): + self.provisioner.create_first_system_administrator( + self.session, + tenant=self.tenant, + email="existing@example.test", + display_name=None, + password="a-production-password", + ) + + +if __name__ == "__main__": + unittest.main()