333 lines
11 KiB
Python
333 lines
11 KiB
Python
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,
|
|
ApiKey,
|
|
AuthSession,
|
|
User,
|
|
)
|
|
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 (
|
|
Tenant,
|
|
create_scope_tables,
|
|
scope_registry,
|
|
)
|
|
|
|
|
|
class ServiceAccountTests(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)
|
|
self.session = self.Session()
|
|
self.tenant = Tenant(
|
|
id="tenant-1",
|
|
slug="tenant-1",
|
|
name="Tenant 1",
|
|
)
|
|
self.account = Account(
|
|
id="account-1",
|
|
email="admin@example.test",
|
|
normalized_email="admin@example.test",
|
|
)
|
|
self.user = User(
|
|
id="user-1",
|
|
tenant_id=self.tenant.id,
|
|
account_id=self.account.id,
|
|
email=self.account.email,
|
|
)
|
|
self.session.add_all(
|
|
(self.tenant, self.account, self.user)
|
|
)
|
|
self.session.commit()
|
|
|
|
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 _principal(
|
|
self,
|
|
scopes: frozenset[str] = frozenset({"tenant:*"}),
|
|
) -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id=self.account.id,
|
|
membership_id=self.user.id,
|
|
tenant_id=self.tenant.id,
|
|
scopes=scopes,
|
|
),
|
|
account=self.account,
|
|
user=self.user,
|
|
)
|
|
|
|
def test_create_builds_a_non_login_identity_without_secrets(self) -> None:
|
|
item = create_service_account(
|
|
self.session,
|
|
tenant=self.tenant,
|
|
principal=self._principal(),
|
|
name=" Monthly import ",
|
|
description="Runs the governed monthly import.",
|
|
scope_ceiling=(
|
|
"dataflow:pipeline:run",
|
|
"datasources:catalogue:read",
|
|
),
|
|
)
|
|
|
|
backing_account = self.session.get(
|
|
Account,
|
|
item.account_id,
|
|
)
|
|
membership = self.session.get(
|
|
User,
|
|
item.membership_id,
|
|
)
|
|
self.assertEqual("Monthly import", item.name)
|
|
self.assertEqual(
|
|
[
|
|
"dataflow:pipeline:run",
|
|
"datasources:catalogue:read",
|
|
],
|
|
item.scope_ceiling,
|
|
)
|
|
self.assertEqual(
|
|
"service_account",
|
|
backing_account.auth_provider,
|
|
)
|
|
self.assertIsNone(backing_account.password_hash)
|
|
self.assertEqual(
|
|
"service_account",
|
|
membership.auth_provider,
|
|
)
|
|
self.assertIsNone(membership.password_hash)
|
|
self.assertEqual(
|
|
0,
|
|
self.session.query(ApiKey)
|
|
.filter(ApiKey.user_id == membership.id)
|
|
.count(),
|
|
)
|
|
self.assertEqual(
|
|
0,
|
|
self.session.query(AuthSession)
|
|
.filter(AuthSession.user_id == membership.id)
|
|
.count(),
|
|
)
|
|
|
|
def test_scope_escalation_and_stale_updates_fail_closed(self) -> None:
|
|
principal = self._principal(
|
|
frozenset({"dataflow:pipeline:run"})
|
|
)
|
|
with self.assertRaises(PermissionError):
|
|
create_service_account(
|
|
self.session,
|
|
tenant=self.tenant,
|
|
principal=principal,
|
|
name="Escalating worker",
|
|
description=None,
|
|
scope_ceiling=("system:settings:write",),
|
|
)
|
|
|
|
item = create_service_account(
|
|
self.session,
|
|
tenant=self.tenant,
|
|
principal=principal,
|
|
name="Bounded worker",
|
|
description=None,
|
|
scope_ceiling=("dataflow:pipeline:run",),
|
|
)
|
|
updated = update_service_account(
|
|
self.session,
|
|
tenant_id=self.tenant.id,
|
|
service_account_id=item.id,
|
|
principal=principal,
|
|
expected_revision=1,
|
|
changes={"description": "Updated"},
|
|
)
|
|
self.assertEqual(2, updated.revision)
|
|
with self.assertRaises(ServiceAccountConflictError):
|
|
update_service_account(
|
|
self.session,
|
|
tenant_id=self.tenant.id,
|
|
service_account_id=item.id,
|
|
principal=principal,
|
|
expected_revision=1,
|
|
changes={"description": "Stale"},
|
|
)
|
|
|
|
def test_retirement_revokes_the_backing_principal(self) -> None:
|
|
principal = self._principal()
|
|
item = create_service_account(
|
|
self.session,
|
|
tenant=self.tenant,
|
|
principal=principal,
|
|
name="Retired worker",
|
|
description=None,
|
|
scope_ceiling=("dataflow:pipeline:run",),
|
|
)
|
|
|
|
retired = retire_service_account(
|
|
self.session,
|
|
tenant_id=self.tenant.id,
|
|
service_account_id=item.id,
|
|
principal=principal,
|
|
expected_revision=1,
|
|
)
|
|
|
|
self.assertFalse(retired.is_active)
|
|
self.assertIsNotNone(retired.retired_at)
|
|
self.assertFalse(
|
|
self.session.get(Account, retired.account_id).is_active
|
|
)
|
|
self.assertFalse(
|
|
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()
|