Complete service-account credential administration

tests/test_interface_documentation_contract.py

@@ -23,12 +23,16 @@ class InterfaceDocumentationContractTests(unittest.TestCase):
                 "access.admin.tenant-groups",
                 "access.admin.tenant-roles",
                 "access.admin.api-keys",
+                "access.admin.service-accounts",
                 "access.credentials",
             },
             "access.reference.external-function-role-mappings": {
                 "access.admin.function-mappings",
                 "access.explanation",
             },
+            "access.workflow.manage-service-account-credentials": {
+                "access.admin.service-accounts",
+            },
         }
 
         for topic_id, expected in expected_contexts.items():
@@ -54,6 +58,7 @@ class InterfaceDocumentationContractTests(unittest.TestCase):
                 "access.admin.tenant-users",
                 "access.admin.tenant-credentials",
                 "access.admin.tenant-api-keys",
+                "access.admin.tenant-service-accounts",
                 "access.admin.group-credentials",
                 "access.admin.user-credentials",
                 "access.settings.credentials",

tests/test_service_accounts.py

@@ -15,9 +15,16 @@ from govoplan_access.backend.db.models import (
 from govoplan_access.backend.service_accounts import (
     ServiceAccountConflictError,
     create_service_account,
+    create_service_account_credential,
+    revoke_service_account_credential,
     retire_service_account,
+    rotate_service_account_credential,
+    service_account_credential_summaries,
     update_service_account,
 )
+from govoplan_access.backend.auth.dependencies import (
+    _resolve_api_key_principal_context,
+)
 from govoplan_core.auth import ApiPrincipal
 from govoplan_core.core.access import PrincipalRef
 from govoplan_core.tenancy.scope import (
@@ -197,6 +204,129 @@ class ServiceAccountTests(unittest.TestCase):
             self.session.get(User, retired.membership_id).is_active
         )
 
+    def test_credentials_are_one_time_scope_bounded_and_rotatable(self) -> None:
+        principal = self._principal()
+        item = create_service_account(
+            self.session,
+            tenant=self.tenant,
+            principal=principal,
+            name="Monthly worker",
+            description=None,
+            scope_ceiling=("dataflow:pipeline:run",),
+        )
+        item, first = create_service_account_credential(
+            self.session,
+            tenant_id=self.tenant.id,
+            service_account_id=item.id,
+            principal=principal,
+            expected_revision=1,
+            name="Worker credential",
+            scopes=("dataflow:pipeline:run",),
+            expires_at=None,
+        )
+        self.assertEqual(2, item.revision)
+        self.assertTrue(first.secret.startswith("mm_"))
+        self.assertNotEqual(first.secret, first.model.key_hash)
+
+        item, previous, replacement = rotate_service_account_credential(
+            self.session,
+            tenant_id=self.tenant.id,
+            service_account_id=item.id,
+            credential_id=first.model.id,
+            principal=principal,
+            expected_revision=2,
+            name=None,
+            scopes=None,
+            expires_at=None,
+        )
+        self.assertEqual(3, item.revision)
+        self.assertIsNotNone(previous.revoked_at)
+        self.assertIsNone(replacement.model.revoked_at)
+        self.assertNotEqual(first.secret, replacement.secret)
+
+        with self.assertRaises(ServiceAccountConflictError):
+            revoke_service_account_credential(
+                self.session,
+                tenant_id=self.tenant.id,
+                service_account_id=item.id,
+                credential_id=replacement.model.id,
+                principal=principal,
+                expected_revision=2,
+            )
+
+        item, revoked = revoke_service_account_credential(
+            self.session,
+            tenant_id=self.tenant.id,
+            service_account_id=item.id,
+            credential_id=replacement.model.id,
+            principal=principal,
+            expected_revision=3,
+        )
+        self.assertEqual(4, item.revision)
+        self.assertIsNotNone(revoked.revoked_at)
+        summary = service_account_credential_summaries(
+            self.session,
+            service_accounts=(item,),
+        )[item.id]
+        self.assertEqual(2, summary.credential_count)
+        self.assertEqual(0, summary.active_credential_count)
+
+    def test_service_account_credential_uses_current_ceiling(self) -> None:
+        principal = self._principal()
+        item = create_service_account(
+            self.session,
+            tenant=self.tenant,
+            principal=principal,
+            name="Bounded API worker",
+            description=None,
+            scope_ceiling=("dataflow:pipeline:run",),
+        )
+        item, created = create_service_account_credential(
+            self.session,
+            tenant_id=self.tenant.id,
+            service_account_id=item.id,
+            principal=principal,
+            expected_revision=1,
+            name="Runtime",
+            scopes=("dataflow:pipeline:run",),
+            expires_at=None,
+        )
+        self.session.commit()
+
+        context = _resolve_api_key_principal_context(
+            self.session,
+            token=created.secret,
+            idm_directory=None,
+            identity_directory=None,
+            organization_directory=None,
+        )
+        self.assertIsNotNone(context)
+        self.assertEqual("service_account", context.principal.auth_method)
+        self.assertEqual(item.id, context.principal.service_account_id)
+        self.assertEqual(created.model.id, context.principal.api_key_id)
+        self.assertEqual(
+            frozenset({"dataflow:pipeline:run"}),
+            context.principal.scopes,
+        )
+
+        update_service_account(
+            self.session,
+            tenant_id=self.tenant.id,
+            service_account_id=item.id,
+            principal=principal,
+            expected_revision=2,
+            changes={"scope_ceiling": []},
+        )
+        self.session.commit()
+        narrowed = _resolve_api_key_principal_context(
+            self.session,
+            token=created.secret,
+            idm_directory=None,
+            identity_directory=None,
+            organization_directory=None,
+        )
+        self.assertEqual(frozenset(), narrowed.principal.scopes)
+
 
 if __name__ == "__main__":
     unittest.main()