From 43f0128e6a3845688a878795e8b0d56bbaacb242 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 19 Aug 2026 23:19:10 +0200 Subject: [PATCH] feat(identity): enforce lifecycle audit semantics --- docs/IDENTITY_MODEL.md | 30 ++++ src/govoplan_identity/backend/lifecycle.py | 168 +++++++++++++++++++++ src/govoplan_identity/backend/manifest.py | 22 +++ tests/test_lifecycle.py | 125 +++++++++++++++ 4 files changed, 345 insertions(+) create mode 100644 src/govoplan_identity/backend/lifecycle.py create mode 100644 tests/test_lifecycle.py diff --git a/docs/IDENTITY_MODEL.md b/docs/IDENTITY_MODEL.md index e97ec42..5bea2cc 100644 --- a/docs/IDENTITY_MODEL.md +++ b/docs/IDENTITY_MODEL.md @@ -10,6 +10,32 @@ authenticate and independent of what they may do. - Primary account: the account used as the default display/explainability anchor when multiple accounts exist. +## Lifecycle semantics + +An active identity is eligible for ordinary directory search. Deactivation +removes it from default search results but does not delete the identity, its +account links, their source provenance, or the primary-account marker. Direct +identifier/account resolution retains the record with an explicit `inactive` +status so Access and reconcilers do not mistake deactivation for absence. +Authorized lifecycle owners may also include inactive identities in search and +may reactivate them. Deactivation is therefore a reversible directory-state +change, not account suspension or erasure; Access owns those separate +consequences. + +Each account link records the origin of the accepted association in `source` +(for example `local` or an IDM reconciliation source). The source is provenance, +not authorization and not proof that the external source remains reachable. +Changing the primary account never rewrites this origin. + +An identity may retain multiple account links but has at most one primary +account. A primary-account change may select only an existing link belonging to +that identity, atomically demotes the previous primary, preserves every link, +and records old/new account ids plus link-source provenance in the audit log. +The lifecycle service does not commit: its caller authorizes the operation and +commits the state and audit record together, or rolls both back on validation, +audit, or persistence failure. Repeating the already-effective selection is a +no-op and does not create misleading audit activity. + ## Boundary With Access Access owns authorization. Identity only tells access which identity is behind @@ -48,3 +74,7 @@ Rollout plan: The close-out condition is that Access works with canonical Identity installed and still works without it through the projection fallback. + +The current lifecycle service is intentionally not an administration API. +Identity administration screens and endpoint permissions remain tracked +separately; IDM continues to own external import/reconciliation decisions. diff --git a/src/govoplan_identity/backend/lifecycle.py b/src/govoplan_identity/backend/lifecycle.py new file mode 100644 index 0000000..6748c1e --- /dev/null +++ b/src/govoplan_identity/backend/lifecycle.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +from dataclasses import dataclass + +from sqlalchemy.orm import Session + +from govoplan_core.audit.logging import audit_event +from govoplan_identity.backend.db.models import Identity, IdentityAccountLink + + +class IdentityLifecycleError(ValueError): + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + + +@dataclass(frozen=True, slots=True) +class IdentityLifecycleResult: + identity_id: str + changed: bool + previous_primary_account_id: str | None = None + primary_account_id: str | None = None + active: bool | None = None + + +def set_identity_active( + session: Session, + *, + identity_id: str, + active: bool, + actor_tenant_id: str | None, + actor_user_id: str | None, + reason: str | None = None, +) -> IdentityLifecycleResult: + """Change directory visibility without deleting identity or link evidence. + + The caller owns authorization and the outer transaction. No commit occurs + here, so an API, import, or reconciliation owner can roll the state and its + audit record back together. + """ + + identity = session.get(Identity, identity_id) + if identity is None: + raise IdentityLifecycleError("identity_not_found", "Identity not found.") + desired = bool(active) + if identity.is_active == desired: + return IdentityLifecycleResult( + identity_id=identity.id, + changed=False, + active=identity.is_active, + ) + with session.begin_nested(): + identity.is_active = desired + audit_event( + session, + tenant_id=actor_tenant_id, + user_id=actor_user_id, + action="identity.activated" if desired else "identity.deactivated", + object_type="identity", + object_id=identity.id, + details={"active": desired, "reason": _bounded_reason(reason)}, + ) + session.flush() + return IdentityLifecycleResult( + identity_id=identity.id, + changed=True, + active=identity.is_active, + ) + + +def set_primary_account( + session: Session, + *, + identity_id: str, + account_id: str, + actor_tenant_id: str | None, + actor_user_id: str | None, + reason: str | None = None, +) -> IdentityLifecycleResult: + """Atomically promote an existing link and retain every other account link.""" + + identity = session.get(Identity, identity_id) + if identity is None: + raise IdentityLifecycleError("identity_not_found", "Identity not found.") + target = ( + session.query(IdentityAccountLink) + .filter( + IdentityAccountLink.identity_id == identity.id, + IdentityAccountLink.account_id == account_id, + ) + .one_or_none() + ) + if target is None: + raise IdentityLifecycleError( + "account_not_linked", + "The requested account is not linked to this identity.", + ) + conflicting_primary = ( + session.query(IdentityAccountLink) + .filter( + IdentityAccountLink.account_id == account_id, + IdentityAccountLink.identity_id != identity.id, + IdentityAccountLink.is_primary.is_(True), + ) + .first() + ) + if conflicting_primary is not None: + raise IdentityLifecycleError( + "account_primary_elsewhere", + "The requested account is already primary for another identity.", + ) + previous = ( + session.query(IdentityAccountLink) + .filter( + IdentityAccountLink.identity_id == identity.id, + IdentityAccountLink.is_primary.is_(True), + ) + .one_or_none() + ) + previous_account_id = previous.account_id if previous is not None else None + if previous is not None and previous.id == target.id: + return IdentityLifecycleResult( + identity_id=identity.id, + changed=False, + previous_primary_account_id=previous_account_id, + primary_account_id=target.account_id, + active=identity.is_active, + ) + with session.begin_nested(): + if previous is not None: + previous.is_primary = False + target.is_primary = True + audit_event( + session, + tenant_id=actor_tenant_id, + user_id=actor_user_id, + action="identity.primary_account_changed", + object_type="identity", + object_id=identity.id, + details={ + "previous_primary_account_id": previous_account_id, + "primary_account_id": target.account_id, + "link_id": target.id, + "link_source": target.source, + "reason": _bounded_reason(reason), + }, + ) + session.flush() + return IdentityLifecycleResult( + identity_id=identity.id, + changed=True, + previous_primary_account_id=previous_account_id, + primary_account_id=target.account_id, + active=identity.is_active, + ) + + +def _bounded_reason(reason: str | None) -> str | None: + normalized = str(reason or "").strip() + return normalized[:500] or None + + +__all__ = [ + "IdentityLifecycleError", + "IdentityLifecycleResult", + "set_identity_active", + "set_primary_account", +] diff --git a/src/govoplan_identity/backend/manifest.py b/src/govoplan_identity/backend/manifest.py index 70aabac..7434ea8 100644 --- a/src/govoplan_identity/backend/manifest.py +++ b/src/govoplan_identity/backend/manifest.py @@ -92,6 +92,28 @@ manifest = ModuleManifest( audience=("tenant_admin", "access_admin", "operator"), order=24, ), + DocumentationTopic( + id="identity.lifecycle", + title="Administer identity and account-link lifecycle", + summary="Deactivate identities reversibly and change the primary account without discarding link provenance.", + body=( + "Deactivation removes an identity from ordinary search while direct resolution retains an explicit inactive record; it preserves all account links and is not account suspension or erasure. " + "A primary-account change selects an existing link, atomically demotes the previous primary, preserves multiple-account compatibility, and records actor, old/new account, and link-source evidence. The authorized caller commits or rolls back state and audit evidence together." + ), + layer="configured", + documentation_types=("admin",), + audience=("tenant_admin", "access_admin", "operator"), + order=25, + metadata={ + "kind": "reference", + "prerequisites": [ + "The caller has separately established lifecycle authority.", + "The replacement primary account is already linked to the identity.", + ], + "outcome": "Identity visibility or primary-account state changes without deleting account-link provenance.", + "verification": "Confirm ordinary versus include-inactive directory results, inspect every retained account link, and review the matching identity lifecycle audit record.", + }, + ), ), architecture=declared_module_architecture( layer="institutional_foundation", diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py new file mode 100644 index 0000000..3638922 --- /dev/null +++ b/tests/test_lifecycle.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from govoplan_core.db.base import Base +from govoplan_core.db.session import configure_database, reset_database +from govoplan_identity.backend.db.models import Identity, IdentityAccountLink +from govoplan_identity.backend.directory import SqlIdentityDirectory +from govoplan_identity.backend.lifecycle import ( + IdentityLifecycleError, + set_identity_active, + set_primary_account, +) + + +class IdentityLifecycleTests(unittest.TestCase): + def setUp(self) -> None: + self.database = configure_database("sqlite:///:memory:") + Base.metadata.create_all( + self.database.engine, + tables=[Identity.__table__, IdentityAccountLink.__table__], + ) + with self.database.session() as session: + session.add(Identity(id="identity-1", display_name="Ada", source="local", is_active=True, settings={})) + session.add_all( + [ + IdentityAccountLink(id="link-1", identity_id="identity-1", account_id="account-1", is_primary=True, source="local"), + IdentityAccountLink(id="link-2", identity_id="identity-1", account_id="account-2", is_primary=False, source="idm:accepted"), + ] + ) + session.commit() + + def tearDown(self) -> None: + reset_database(dispose=True) + + @patch("govoplan_identity.backend.lifecycle.audit_event") + def test_primary_account_change_preserves_all_links_and_audits_source(self, audit) -> None: + with self.database.session() as session: + result = set_primary_account( + session, + identity_id="identity-1", + account_id="account-2", + actor_tenant_id="tenant-1", + actor_user_id="user-1", + reason="Preferred institutional account", + ) + session.commit() + + self.assertTrue(result.changed) + self.assertEqual("account-1", result.previous_primary_account_id) + self.assertEqual("account-2", result.primary_account_id) + with self.database.session() as session: + links = session.query(IdentityAccountLink).filter_by(identity_id="identity-1").order_by(IdentityAccountLink.account_id).all() + self.assertEqual([False, True], [item.is_primary for item in links]) + resolved = SqlIdentityDirectory().get_identity("identity-1") + self.assertIsNotNone(resolved) + self.assertEqual(("account-1", "account-2"), tuple(sorted(resolved.account_ids))) + self.assertEqual("account-2", resolved.primary_account_id) + self.assertEqual("identity.primary_account_changed", audit.call_args.kwargs["action"]) + self.assertEqual("idm:accepted", audit.call_args.kwargs["details"]["link_source"]) + + @patch("govoplan_identity.backend.lifecycle.audit_event") + def test_deactivation_is_reversible_and_preserves_account_links(self, audit) -> None: + with self.database.session() as session: + changed = set_identity_active( + session, + identity_id="identity-1", + active=False, + actor_tenant_id="tenant-1", + actor_user_id="user-1", + ) + session.commit() + self.assertTrue(changed.changed) + inactive = SqlIdentityDirectory().get_identity("identity-1") + self.assertIsNotNone(inactive) + self.assertEqual("inactive", inactive.status) + self.assertEqual((), SqlIdentityDirectory().search_identities()) + with self.database.session() as session: + self.assertEqual(2, session.query(IdentityAccountLink).filter_by(identity_id="identity-1").count()) + set_identity_active( + session, + identity_id="identity-1", + active=True, + actor_tenant_id="tenant-1", + actor_user_id="user-1", + ) + session.commit() + self.assertIsNotNone(SqlIdentityDirectory().get_identity("identity-1")) + self.assertEqual(["identity.deactivated", "identity.activated"], [call.kwargs["action"] for call in audit.call_args_list]) + + @patch("govoplan_identity.backend.lifecycle.audit_event") + def test_invalid_primary_change_does_not_mutate_or_audit(self, audit) -> None: + with self.database.session() as session: + with self.assertRaises(IdentityLifecycleError) as raised: + set_primary_account( + session, + identity_id="identity-1", + account_id="not-linked", + actor_tenant_id="tenant-1", + actor_user_id="user-1", + ) + self.assertEqual("account_not_linked", raised.exception.code) + links = session.query(IdentityAccountLink).filter_by(identity_id="identity-1").order_by(IdentityAccountLink.account_id).all() + self.assertEqual([True, False], [item.is_primary for item in links]) + audit.assert_not_called() + + @patch("govoplan_identity.backend.lifecycle.audit_event", side_effect=RuntimeError("audit unavailable")) + def test_primary_change_rolls_back_when_audit_cannot_be_recorded(self, _audit) -> None: + with self.database.session() as session: + with self.assertRaisesRegex(RuntimeError, "audit unavailable"): + set_primary_account( + session, + identity_id="identity-1", + account_id="account-2", + actor_tenant_id="tenant-1", + actor_user_id="user-1", + ) + session.expire_all() + links = session.query(IdentityAccountLink).filter_by(identity_id="identity-1").order_by(IdentityAccountLink.account_id).all() + self.assertEqual([True, False], [item.is_primary for item in links]) + + +if __name__ == "__main__": + unittest.main()