diff --git a/docs/FUNCTION_ASSIGNMENT_WORKFLOWS.md b/docs/FUNCTION_ASSIGNMENT_WORKFLOWS.md index 912b17a..ee36a5c 100644 --- a/docs/FUNCTION_ASSIGNMENT_WORKFLOWS.md +++ b/docs/FUNCTION_ASSIGNMENT_WORKFLOWS.md @@ -54,7 +54,7 @@ effective decision and provenance rather than IDM duplicating policy logic. ## IDM Aggregate -IDM should persist one function-assignment change aggregate for both journeys: +IDM persists one function-assignment change aggregate for both journeys: - change ID, tenant, kind (`request` or `grant`), function, unit, candidate identity/account, requested validity, and assignment source @@ -67,9 +67,9 @@ IDM should persist one function-assignment change aggregate for both journeys: - idempotency key and optimistic-concurrency revision Candidate states are `draft`, `submitted`, `awaiting_holder`, -`awaiting_authority`, `awaiting_recipient`, `approved`, `accepted`, `applied`, -`rejected`, `withdrawn`, `expired`, `cancelled`, and `failed_manual_review`. -Not every profile uses every state. +`awaiting_authority`, `awaiting_recipient`, `changes_requested`, `blocked`, +`approved`, `accepted`, `applied`, `rejected`, `withdrawn`, `expired`, +`cancelled`, and `failed_manual_review`. Not every profile uses every state. The workflow instance coordinates the process, but the IDM change record is the business source of truth. A workflow callback applies the assignment exactly @@ -97,6 +97,30 @@ reconciled without starting a second grant. functions or treated as an explicit emergency override with reason, provenance, and equivalent evidence. +## Configuration And Runtime Contract + +Function settings use `assignment_governance`. Tenant defaults may be supplied +through `settings.function_assignment_governance_defaults`; function values +override only the corresponding defaults. Supported keys include +`request_profile`, `grant_profile`, `authority_function_id`, +`recipient_acceptance_required`, `evidence_required`, +`separation_of_duties`, `quorum`, `maximum_validity_days`, and +`request_expiry_hours`. Missing or malformed profiles fail closed. + +IDM exposes governed changes at +`/api/v1/idm/function-assignment-changes`. Mutations require a strong `If-Match` +precondition and the aggregate revision. Requests and grants pin the exact +Workflow definition revision and hash, retain append-only transition evidence, +support review change requests and responses, and apply an assignment exactly +once. Vacant holder or authority functions create a visible `blocked` state; +the lifecycle worker expires overdue open changes durably. + +Direct administration remains available for independently deployed IDM. When +a function has an enabled governance profile, however, direct create or update +requires `idm:function_change:admin` plus an emergency override reason. IDM +stores the actor, time, reason, and evidence references with the assignment and +includes them in the normal Audit change record. + ## Module Boundaries - Organizations: function definitions and assignment-policy profile reference. diff --git a/src/govoplan_idm/backend/api/v1/function_changes.py b/src/govoplan_idm/backend/api/v1/function_changes.py new file mode 100644 index 0000000..dc97e04 --- /dev/null +++ b/src/govoplan_idm/backend/api/v1/function_changes.py @@ -0,0 +1,551 @@ +from __future__ import annotations + +from math import ceil + +from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response, status +from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from govoplan_core.audit.logging import audit_from_principal +from govoplan_core.auth import ApiPrincipal, get_api_principal +from govoplan_core.core.access import CAPABILITY_AUDIT_RECORDER +from govoplan_core.core.concurrency import ( + ConcurrencyError, + MissingPreconditionError, + RevisionConflictError, + assert_revision_precondition, +) +from govoplan_core.core.identity import ( + CAPABILITY_IDENTITY_DIRECTORY, + IdentityDirectory, +) +from govoplan_core.core.organizations import ( + CAPABILITY_ORGANIZATION_DIRECTORY, + OrganizationDirectory, + OrganizationFunctionRef, +) +from govoplan_core.core.policy import ( + CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE, +) +from govoplan_core.core.runtime import get_registry +from govoplan_core.core.workflows import CAPABILITY_WORKFLOW_ORCHESTRATION +from govoplan_core.db.session import get_session +from govoplan_idm.backend.db.models import IdmFunctionAssignmentChange +from govoplan_idm.backend.function_assignment_changes import ( + FunctionAssignmentChangeConflict, + FunctionAssignmentChangeUnavailable, + available_change_actions, + change_events, + create_function_assignment_change, + resolve_submission_capability, + transition_function_assignment_change, + visible_change_filter, +) + +from .schemas import ( + FunctionAssignmentCapabilityItem, + FunctionAssignmentChangeActionRequest, + FunctionAssignmentChangeCreateRequest, + FunctionAssignmentChangeEventItem, + FunctionAssignmentChangeItem, + FunctionAssignmentChangeKind, + FunctionAssignmentChangeList, +) + + +router = APIRouter(prefix="/function-assignment-changes") +READ_SCOPES = ( + "idm:function_change:read", + "idm:function_request:create", + "idm:function_grant:create", + "idm:function_change:decide", + "idm:function_change:admin", + "idm:organization_assignment:write", +) + + +def _require_any(principal: ApiPrincipal, *scopes: str) -> None: + if not any(principal.has(scope) for scope in scopes): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=f"Requires one of: {', '.join(scopes)}", + ) + + +def _registry(): + registry = get_registry() + if registry is None: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="The module registry is unavailable.", + ) + return registry + + +def _organization_directory() -> OrganizationDirectory: + registry = _registry() + capability = registry.capability(CAPABILITY_ORGANIZATION_DIRECTORY) + if not isinstance(capability, OrganizationDirectory): + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="The Organizations directory is unavailable.", + ) + return capability + + +def _identity_directory() -> IdentityDirectory: + registry = _registry() + capability = registry.capability(CAPABILITY_IDENTITY_DIRECTORY) + if not isinstance(capability, IdentityDirectory): + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="The Identity directory is unavailable.", + ) + return capability + + +def _function(function_id: str, tenant_id: str) -> OrganizationFunctionRef: + function = _organization_directory().get_function(function_id) + if function is None or function.tenant_id != tenant_id: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Organization function not found.", + ) + return function + + +def _historical_function( + function_id: str, + tenant_id: str, +) -> OrganizationFunctionRef | None: + try: + function = _organization_directory().get_function(function_id) + except HTTPException: + return None + if function is None or function.tenant_id != tenant_id: + return None + return function + + +def _validate_candidate(identity_id: str, account_id: str | None) -> None: + directory = _identity_directory() + if directory.get_identity(identity_id) is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Identity not found.", + ) + if account_id is not None and not any( + link.account_id == account_id + for link in directory.accounts_for_identity(identity_id) + ): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail="Account is not linked to the selected identity.", + ) + + +def _change_query( + session: Session, + principal: ApiPrincipal, +): + statement = select(IdmFunctionAssignmentChange).where( + IdmFunctionAssignmentChange.tenant_id == principal.tenant_id + ) + visibility = visible_change_filter(session, principal) + return statement.where(visibility) if visibility is not None else statement + + +def _get_change( + session: Session, + principal: ApiPrincipal, + change_id: str, +) -> IdmFunctionAssignmentChange: + change = session.scalar( + _change_query(session, principal).where( + IdmFunctionAssignmentChange.id == change_id + ) + ) + if change is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Function assignment change not found.", + ) + return change + + +def _event_item(event) -> FunctionAssignmentChangeEventItem: + return FunctionAssignmentChangeEventItem( + id=event.id, + sequence=event.sequence, + action=event.action, + from_state=event.from_state, + to_state=event.to_state, + actor_account_id=event.actor_account_id, + actor_identity_id=event.actor_identity_id, + actor_assignment_id=event.actor_assignment_id, + comment=event.comment, + evidence=list(event.evidence), + policy_decision=dict(event.policy_decision), + workflow_step_id=event.workflow_step_id, + details=dict(event.details), + created_at=event.created_at, + ) + + +def _change_item( + session: Session, + principal: ApiPrincipal, + change: IdmFunctionAssignmentChange, + *, + include_events: bool, +) -> FunctionAssignmentChangeItem: + function = _historical_function(change.function_id, change.tenant_id) + if function is None: + actions, reason = [], "The referenced organization function is no longer available." + else: + try: + actions, reason = available_change_actions( + session, + principal=principal, + registry=get_registry(), + change=change, + function=function, + ) + except FunctionAssignmentChangeUnavailable as exc: + actions, reason = [], str(exc) + return FunctionAssignmentChangeItem( + id=change.id, + tenant_id=change.tenant_id, + kind=change.kind, + state=change.state, + profile=change.profile, + function_id=change.function_id, + organization_unit_id=change.organization_unit_id, + candidate_identity_id=change.candidate_identity_id, + candidate_account_id=change.candidate_account_id, + initiator_account_id=change.initiator_account_id, + initiator_identity_id=change.initiator_identity_id, + represented_assignment_id=change.represented_assignment_id, + justification=change.justification, + evidence=list(change.evidence), + requested_valid_from=change.requested_valid_from, + requested_valid_until=change.requested_valid_until, + applies_to_subunits=change.applies_to_subunits, + assignment_source=change.assignment_source, + required_steps=list(change.required_steps), + completed_steps=list(change.completed_steps), + policy_decision=dict(change.policy_decision), + workflow_definition_id=change.workflow_definition_id, + workflow_definition_revision_id=change.workflow_definition_revision_id, + workflow_definition_revision=change.workflow_definition_revision, + workflow_definition_hash=change.workflow_definition_hash, + workflow_instance_id=change.workflow_instance_id, + workflow_current_step_id=change.workflow_current_step_id, + resulting_assignment_id=change.resulting_assignment_id, + expires_at=change.expires_at, + outcome_reason=change.outcome_reason, + resource_revision=change.resource_revision, + etag=change.strong_etag, + metadata=dict(change.metadata_), + events=( + [ + _event_item(event) + for event in change_events(session, change_id=change.id) + ] + if include_events + else [] + ), + available_actions=actions, + availability_reason=reason, + created_at=change.created_at, + updated_at=change.updated_at, + ) + + +def _set_etag(response: Response, change: IdmFunctionAssignmentChange) -> None: + response.headers["ETag"] = change.strong_etag + + +def _record_change_audit( + session: Session, + principal: ApiPrincipal, + *, + change: IdmFunctionAssignmentChange, + action: str, + from_state: str | None, +) -> None: + registry = get_registry() + if registry is None or not registry.has_capability(CAPABILITY_AUDIT_RECORDER): + return + audit_from_principal( + session, + principal, + action=f"idm.function_assignment_change.{action}", + object_type="function_assignment_change", + object_id=change.id, + details={ + "kind": change.kind, + "from_state": from_state, + "to_state": change.state, + "function_id": change.function_id, + "candidate_identity_id": change.candidate_identity_id, + "represented_assignment_id": change.represented_assignment_id, + "policy_decision": dict(change.policy_decision), + "workflow_definition_id": change.workflow_definition_id, + "workflow_definition_revision_id": change.workflow_definition_revision_id, + "workflow_instance_id": change.workflow_instance_id, + "evidence": list(change.evidence), + "resulting_assignment_id": change.resulting_assignment_id, + "resource_revision": change.resource_revision, + }, + commit=False, + ) + + +def _mutation_error(exc: Exception) -> HTTPException: + if isinstance(exc, MissingPreconditionError): + return HTTPException( + status_code=status.HTTP_428_PRECONDITION_REQUIRED, + detail=exc.as_dict(), + ) + if isinstance(exc, RevisionConflictError): + return HTTPException( + status_code=status.HTTP_412_PRECONDITION_FAILED, + detail=exc.as_dict(), + ) + if isinstance(exc, ConcurrencyError): + return HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={"code": "concurrency_conflict", "message": str(exc)}, + ) + if isinstance(exc, FunctionAssignmentChangeConflict): + return HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) + return HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=str(exc), + ) + + +@router.get("/capability", response_model=FunctionAssignmentCapabilityItem) +def get_function_assignment_capability( + kind: FunctionAssignmentChangeKind, + function_id: str = Query(min_length=1, max_length=36), + candidate_identity_id: str | None = Query(default=None, max_length=36), + candidate_account_id: str | None = Query(default=None, max_length=36), + has_evidence: bool = False, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> FunctionAssignmentCapabilityItem: + _require_any(principal, *READ_SCOPES) + candidate_identity_id = candidate_identity_id or principal.identity_id + if candidate_identity_id is None: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail="A candidate identity is required.", + ) + function = _function(function_id, principal.tenant_id) + registry = _registry() + decision, reason = resolve_submission_capability( + session, + principal=principal, + registry=registry, + kind=kind, + function=function, + candidate_identity_id=candidate_identity_id, + candidate_account_id=candidate_account_id, + has_evidence=has_evidence, + ) + return FunctionAssignmentCapabilityItem( + kind=kind, + function_id=function_id, + available=bool(decision and decision.allowed), + reason=reason, + profile=decision.profile if decision else "unavailable", + required_steps=list(decision.required_steps) if decision else [], + requirements=list(decision.requirements) if decision else [], + authority_function_id=(decision.authority_function_id if decision else None), + evidence_required=bool(decision and decision.evidence_required), + recipient_acceptance_required=bool( + decision and decision.recipient_acceptance_required + ), + maximum_validity_days=(decision.maximum_validity_days if decision else None), + workflow_available=registry.has_capability(CAPABILITY_WORKFLOW_ORCHESTRATION), + policy_available=registry.has_capability( + CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE + ), + ) + + +@router.get("", response_model=FunctionAssignmentChangeList) +def list_function_assignment_changes( + kind: FunctionAssignmentChangeKind | None = None, + state_filter: str | None = Query(default=None, alias="state", max_length=40), + page: int = Query(default=1, ge=1), + page_size: int = Query(default=50, ge=1, le=200), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> FunctionAssignmentChangeList: + _require_any(principal, *READ_SCOPES) + statement = _change_query(session, principal) + if kind is not None: + statement = statement.where(IdmFunctionAssignmentChange.kind == kind) + if state_filter: + statement = statement.where(IdmFunctionAssignmentChange.state == state_filter) + total = int( + session.scalar(select(func.count()).select_from(statement.subquery())) or 0 + ) + rows = list( + session.scalars( + statement.order_by( + IdmFunctionAssignmentChange.updated_at.desc(), + IdmFunctionAssignmentChange.id.desc(), + ) + .offset((page - 1) * page_size) + .limit(page_size) + ) + ) + return FunctionAssignmentChangeList( + changes=[ + _change_item(session, principal, change, include_events=False) + for change in rows + ], + total=total, + page=page, + page_size=page_size, + pages=max(1, ceil(total / page_size)), + ) + + +@router.get("/{change_id}", response_model=FunctionAssignmentChangeItem) +def get_function_assignment_change( + change_id: str, + response: Response, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> FunctionAssignmentChangeItem: + _require_any(principal, *READ_SCOPES) + change = _get_change(session, principal, change_id) + _set_etag(response, change) + return _change_item(session, principal, change, include_events=True) + + +@router.post("", response_model=FunctionAssignmentChangeItem, status_code=201) +def create_governed_function_assignment_change( + payload: FunctionAssignmentChangeCreateRequest, + response: Response, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> FunctionAssignmentChangeItem: + required = ( + "idm:function_request:create" + if payload.kind == "request" + else "idm:function_grant:create" + ) + _require_any( + principal, + required, + "idm:function_change:admin", + "idm:organization_assignment:write", + ) + _validate_candidate(payload.candidate_identity_id, payload.candidate_account_id) + try: + change, replayed = create_function_assignment_change( + session, + principal=principal, + registry=_registry(), + function=_function(payload.function_id, principal.tenant_id), + payload=payload, + ) + if not replayed: + _record_change_audit( + session, + principal, + change=change, + action="created", + from_state=None, + ) + session.commit() + except ( + FunctionAssignmentChangeConflict, + FunctionAssignmentChangeUnavailable, + ) as exc: + session.rollback() + raise _mutation_error(exc) from exc + except IntegrityError as exc: + session.rollback() + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="The function assignment change conflicts with existing data.", + ) from exc + session.refresh(change) + response.status_code = status.HTTP_200_OK if replayed else status.HTTP_201_CREATED + _set_etag(response, change) + return _change_item(session, principal, change, include_events=True) + + +@router.post("/{change_id}/actions", response_model=FunctionAssignmentChangeItem) +def act_on_function_assignment_change( + change_id: str, + payload: FunctionAssignmentChangeActionRequest, + response: Response, + if_match: str | None = Header(default=None, alias="If-Match"), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> FunctionAssignmentChangeItem: + change = _get_change(session, principal, change_id) + from_state = change.state + if payload.action == "recover": + _require_any(principal, "idm:function_change:admin") + elif payload.action not in {"withdraw", "respond"}: + _require_any( + principal, + "idm:function_change:decide", + "idm:function_change:admin", + "idm:organization_assignment:write", + ) + try: + assert_revision_precondition( + if_match, + resource_type="idm_function_assignment_change", + resource_id=change.id, + submitted_base_revision=payload.base_revision, + ) + change = transition_function_assignment_change( + session, + principal=principal, + registry=_registry(), + change=change, + function=_function(change.function_id, change.tenant_id), + action=payload.action, + base_revision=payload.base_revision, + comment=payload.comment, + evidence=payload.evidence, + ) + _record_change_audit( + session, + principal, + change=change, + action=payload.action, + from_state=from_state, + ) + session.commit() + except ( + ConcurrencyError, + FunctionAssignmentChangeConflict, + FunctionAssignmentChangeUnavailable, + ) as exc: + session.rollback() + raise _mutation_error(exc) from exc + except IntegrityError as exc: + session.rollback() + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="The function assignment transition conflicts with existing data.", + ) from exc + session.refresh(change) + _set_etag(response, change) + return _change_item(session, principal, change, include_events=True) + + +__all__ = ["router"] diff --git a/src/govoplan_idm/backend/api/v1/routes.py b/src/govoplan_idm/backend/api/v1/routes.py index bd6497f..343255a 100644 --- a/src/govoplan_idm/backend/api/v1/routes.py +++ b/src/govoplan_idm/backend/api/v1/routes.py @@ -284,7 +284,14 @@ def _plan_assignment_update( *, tenant_id: str, ) -> AssignmentMutationPlan: - values = payload.model_dump(exclude_unset=True, exclude={"change_request_id"}) + values = payload.model_dump( + exclude_unset=True, + exclude={ + "change_request_id", + "governance_override_reason", + "governance_override_evidence", + }, + ) organization_unit_id: str | None = None if "function_id" in values and values["function_id"] is not None: function = _organization_function(str(values["function_id"]), tenant_id) @@ -330,16 +337,32 @@ def _actor_id(principal: ApiPrincipal) -> str: return principal.membership_id or principal.account_id -def _payload_for_control(resource_type: str, operation: str, payload: object) -> dict[str, Any]: +def _payload_for_control( + resource_type: str, operation: str, payload: object +) -> dict[str, Any]: if hasattr(payload, "model_dump"): - values = payload.model_dump(mode="json", exclude={"change_request_id"}, exclude_unset=True) # type: ignore[attr-defined] + values = payload.model_dump( # type: ignore[attr-defined] + mode="json", + exclude={ + "change_request_id", + "governance_override_reason", + "governance_override_evidence", + }, + exclude_unset=True, + ) else: values = {} return {"resource_type": resource_type, "operation": operation, "payload": values} -def _target_for_control(tenant_id: str, resource_type: str, operation: str, resource_id: str | None = None) -> dict[str, Any]: - target: dict[str, Any] = {"tenant_id": tenant_id, "resource_type": resource_type, "operation": operation} +def _target_for_control( + tenant_id: str, resource_type: str, operation: str, resource_id: str | None = None +) -> dict[str, Any]: + target: dict[str, Any] = { + "tenant_id": tenant_id, + "resource_type": resource_type, + "operation": operation, + } if resource_id is not None: target["resource_id"] = resource_id return target @@ -440,6 +463,48 @@ def _publish_assignment_event( ) +def _apply_governance_override( + principal: ApiPrincipal, + function: OrganizationFunctionRef, + payload: object, + settings: dict[str, Any], +) -> dict[str, Any]: + governance = function.settings.get("assignment_governance") + if not isinstance(governance, dict) or not any( + str(governance.get(key) or "unavailable").strip().casefold() != "unavailable" + for key in ("request_profile", "grant_profile") + ): + return settings + if not principal.has("idm:function_change:admin"): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=( + "Direct assignment to a governed function requires function " + "change recovery permission." + ), + ) + reason = str(getattr(payload, "governance_override_reason", None) or "").strip() + if not reason: + raise _invalid( + "Direct assignment to a governed function requires an emergency override reason." + ) + evidence = [ + str(item).strip() + for item in getattr(payload, "governance_override_evidence", ()) + if str(item).strip() + ] + return { + **settings, + "governance_override": { + "reason": reason, + "evidence": evidence, + "actor_account_id": principal.account_id, + "recorded_at": utc_now().isoformat(), + "function_id": function.id, + }, + } + + @router.get("/settings", response_model=IdmSettingsItem) def get_idm_settings( session: Session = Depends(get_session), @@ -545,6 +610,12 @@ def create_organization_function_assignment( _ensure_identity(payload.identity_id) _ensure_account_link(payload.identity_id, payload.account_id) function = _organization_function(payload.function_id, tenant_id) + item_settings = _apply_governance_override( + principal, + function, + payload, + dict(payload.settings), + ) item = IdmOrganizationFunctionAssignment( tenant_id=tenant_id, identity_id=payload.identity_id, @@ -558,7 +629,7 @@ def create_organization_function_assignment( valid_from=payload.valid_from, valid_until=payload.valid_until, is_active=payload.is_active, - settings=payload.settings, + settings=item_settings, ) if item.delegated_from_assignment_id is not None: _get_tenant_row(session, IdmOrganizationFunctionAssignment, item.delegated_from_assignment_id, tenant_id, "Delegated function assignment") @@ -614,6 +685,13 @@ def update_organization_function_assignment( tenant_id=tenant_id, ) plan.apply(item) + function = _organization_function(item.function_id, tenant_id) + item.settings = _apply_governance_override( + principal, + function, + payload, + dict(item.settings), + ) now = utc_now() event_types = lifecycle_event_types( plan.before, @@ -678,3 +756,9 @@ def _identity_candidate(identity: IdentityRef) -> OrganizationIdentityCandidate: account_ids=list(identity.account_ids), status=identity.status, ) + + +from .function_changes import router as function_changes_router # noqa: E402 + + +router.include_router(function_changes_router) diff --git a/src/govoplan_idm/backend/api/v1/schemas.py b/src/govoplan_idm/backend/api/v1/schemas.py index e4014ff..a670910 100644 --- a/src/govoplan_idm/backend/api/v1/schemas.py +++ b/src/govoplan_idm/backend/api/v1/schemas.py @@ -65,6 +65,10 @@ class OrganizationFunctionAssignmentCreateRequest(BaseModel): is_active: bool = True settings: dict[str, Any] = Field(default_factory=dict) change_request_id: str | None = None + governance_override_reason: str | None = Field(default=None, max_length=4_000) + governance_override_evidence: list[str] = Field( + default_factory=list, max_length=100 + ) class OrganizationFunctionAssignmentUpdateRequest(BaseModel): @@ -80,6 +84,10 @@ class OrganizationFunctionAssignmentUpdateRequest(BaseModel): is_active: bool | None = None settings: dict[str, Any] | None = None change_request_id: str | None = None + governance_override_reason: str | None = Field(default=None, max_length=4_000) + governance_override_evidence: list[str] = Field( + default_factory=list, max_length=100 + ) class IdmSettingsItem(BaseModel): @@ -97,3 +105,120 @@ class IdmSettingsUpdateRequest(BaseModel): audit_detail_level: AuditDetailLevel | None = None change_retention_days: int | None = Field(default=None, ge=0) settings: dict[str, Any] | None = None + + +FunctionAssignmentChangeKind = Literal["request", "grant"] +FunctionAssignmentChangeAction = Literal[ + "approve", + "reject", + "accept", + "request_changes", + "respond", + "withdraw", + "recover", +] + + +class FunctionAssignmentChangeCreateRequest(BaseModel): + kind: FunctionAssignmentChangeKind + function_id: str = Field(min_length=1, max_length=36) + candidate_identity_id: str = Field(min_length=1, max_length=36) + candidate_account_id: str | None = Field(default=None, max_length=36) + justification: str = Field(min_length=1, max_length=8_000) + evidence: list[str] = Field(default_factory=list, max_length=100) + requested_valid_from: datetime | None = None + requested_valid_until: datetime | None = None + applies_to_subunits: bool = False + assignment_source: Literal["governance", "delegated"] = "governance" + represented_assignment_id: str | None = Field(default=None, max_length=36) + idempotency_key: str = Field(min_length=1, max_length=255) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class FunctionAssignmentChangeActionRequest(BaseModel): + action: FunctionAssignmentChangeAction + base_revision: int = Field(ge=1) + comment: str | None = Field(default=None, max_length=4_000) + evidence: list[str] = Field(default_factory=list, max_length=100) + + +class FunctionAssignmentChangeEventItem(BaseModel): + id: str + sequence: int + action: str + from_state: str | None = None + to_state: str + actor_account_id: str | None = None + actor_identity_id: str | None = None + actor_assignment_id: str | None = None + comment: str | None = None + evidence: list[str] + policy_decision: dict[str, Any] + workflow_step_id: str | None = None + details: dict[str, Any] + created_at: datetime + + +class FunctionAssignmentChangeItem(BaseModel): + id: str + tenant_id: str + kind: FunctionAssignmentChangeKind + state: str + profile: str + function_id: str + organization_unit_id: str + candidate_identity_id: str + candidate_account_id: str | None = None + initiator_account_id: str + initiator_identity_id: str | None = None + represented_assignment_id: str | None = None + justification: str + evidence: list[str] + requested_valid_from: datetime | None = None + requested_valid_until: datetime | None = None + applies_to_subunits: bool + assignment_source: str + required_steps: list[str] + completed_steps: list[str] + policy_decision: dict[str, Any] + workflow_definition_id: str | None = None + workflow_definition_revision_id: str | None = None + workflow_definition_revision: int | None = None + workflow_definition_hash: str | None = None + workflow_instance_id: str | None = None + workflow_current_step_id: str | None = None + resulting_assignment_id: str | None = None + expires_at: datetime | None = None + outcome_reason: str | None = None + resource_revision: int + etag: str + metadata: dict[str, Any] + events: list[FunctionAssignmentChangeEventItem] = Field(default_factory=list) + available_actions: list[str] = Field(default_factory=list) + availability_reason: str | None = None + created_at: datetime + updated_at: datetime + + +class FunctionAssignmentChangeList(BaseModel): + changes: list[FunctionAssignmentChangeItem] + total: int + page: int + page_size: int + pages: int + + +class FunctionAssignmentCapabilityItem(BaseModel): + kind: FunctionAssignmentChangeKind + function_id: str + available: bool + reason: str | None = None + profile: str = "unavailable" + required_steps: list[str] = Field(default_factory=list) + requirements: list[str] = Field(default_factory=list) + authority_function_id: str | None = None + evidence_required: bool = False + recipient_acceptance_required: bool = False + maximum_validity_days: int | None = None + workflow_available: bool = False + policy_available: bool = False diff --git a/src/govoplan_idm/backend/assignment_lifecycle.py b/src/govoplan_idm/backend/assignment_lifecycle.py index e31c9ef..a4fcd2c 100644 --- a/src/govoplan_idm/backend/assignment_lifecycle.py +++ b/src/govoplan_idm/backend/assignment_lifecycle.py @@ -2,17 +2,37 @@ from __future__ import annotations from datetime import datetime +from sqlalchemy import func from sqlalchemy.orm import Session +from govoplan_core.core.events import ( + EventActorRef, + EventObjectRef, + EventTenantRef, + PlatformEvent, + emit_platform_event, +) from govoplan_core.core.principal_cache import invalidate_auth_principals +from govoplan_core.core.notifications import ( + NotificationDispatchRequest, + notification_dispatch_provider, +) from govoplan_core.security.time import ensure_aware_utc, utc_now from govoplan_idm.backend.assignment_events import emit_assignment_event -from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment +from govoplan_idm.backend.db.models import ( + IdmFunctionAssignmentChange, + IdmFunctionAssignmentChangeEvent, + IdmOrganizationFunctionAssignment, +) +from govoplan_idm.backend.function_assignment_changes import OPEN_STATES class SqlIdmAssignmentLifecycle: """Claim and publish elapsed assignments exactly once per validity window.""" + def __init__(self, *, registry: object | None = None) -> None: + self._registry = registry + def process_expired( self, session: object, @@ -83,11 +103,157 @@ class SqlIdmAssignmentLifecycle: resource_type="organization_function_assignment_expiry", resource_id=touched_tenant_id, ) + expired_change_ids = self._expire_open_changes( + session, + tenant_id=tenant_id, + effective_at=now, + limit=limit, + ) return { "selected": len(candidates), "expired": len(expired_ids), "assignment_ids": expired_ids, + "expired_changes": len(expired_change_ids), + "change_ids": expired_change_ids, } + def _expire_open_changes( + self, + session: Session, + *, + tenant_id: str | None, + effective_at: datetime, + limit: int, + ) -> list[str]: + query = session.query(IdmFunctionAssignmentChange).filter( + IdmFunctionAssignmentChange.state.in_(OPEN_STATES), + IdmFunctionAssignmentChange.expires_at.is_not(None), + IdmFunctionAssignmentChange.expires_at <= effective_at, + ) + if tenant_id is not None: + query = query.filter(IdmFunctionAssignmentChange.tenant_id == tenant_id) + candidates = ( + query.order_by( + IdmFunctionAssignmentChange.expires_at.asc(), + IdmFunctionAssignmentChange.id.asc(), + ) + .limit(limit) + .all() + ) + expired_ids: list[str] = [] + for change in candidates: + previous_state = change.state + claimed = ( + session.query(IdmFunctionAssignmentChange) + .filter( + IdmFunctionAssignmentChange.id == change.id, + IdmFunctionAssignmentChange.state == previous_state, + IdmFunctionAssignmentChange.expires_at.is_not(None), + IdmFunctionAssignmentChange.expires_at <= effective_at, + ) + .update( + { + IdmFunctionAssignmentChange.state: "expired", + IdmFunctionAssignmentChange.outcome_reason: ( + "The governed function assignment change expired." + ), + IdmFunctionAssignmentChange.resource_revision: ( + IdmFunctionAssignmentChange.resource_revision + 1 + ), + }, + synchronize_session=False, + ) + ) + if claimed != 1: + continue + session.refresh(change) + sequence = ( + int( + session.scalar( + session.query( + func.max(IdmFunctionAssignmentChangeEvent.sequence) + ) + .filter(IdmFunctionAssignmentChangeEvent.change_id == change.id) + .statement + ) + or 0 + ) + + 1 + ) + session.add( + IdmFunctionAssignmentChangeEvent( + tenant_id=change.tenant_id, + change_id=change.id, + sequence=sequence, + action="expired", + from_state=previous_state, + to_state="expired", + policy_decision=dict(change.policy_decision), + details={"effective_at": effective_at.isoformat()}, + created_at=effective_at, + ) + ) + emit_platform_event( + session, + PlatformEvent( + type="idm.function_change.expired.v1", + module_id="idm", + payload={ + "kind": change.kind, + "state": change.state, + "function_id": change.function_id, + "candidate_identity_id": change.candidate_identity_id, + "resource_revision": change.resource_revision, + }, + actor=EventActorRef(type="system"), + tenant=EventTenantRef(id=change.tenant_id), + subject=EventObjectRef( + type="organization_function", + id=change.function_id, + ), + resource=EventObjectRef( + type="function_assignment_change", + id=change.id, + ), + classification="internal", + ), + ) + self._notify_expiry(session, change) + expired_ids.append(change.id) + return expired_ids + + def _notify_expiry( + self, + session: Session, + change: IdmFunctionAssignmentChange, + ) -> None: + provider = notification_dispatch_provider(self._registry) + if provider is None: + return + recipient_ids = { + change.initiator_account_id, + change.candidate_account_id, + } + for account_id in sorted(item for item in recipient_ids if item): + provider.enqueue_notification( + session, + NotificationDispatchRequest( + tenant_id=change.tenant_id, + source_module="idm", + source_resource_type="function_assignment_change", + source_resource_id=change.id, + event_kind="function_assignment_change.expired", + recipient_type="account", + recipient_id=account_id, + subject="Function assignment change expired", + body_text=( + "The governed function assignment change expired " + "before all required decisions were completed." + ), + action_url=f"/idm?change={change.id}", + payload={"change_id": change.id, "state": "expired"}, + ), + ) + __all__ = ["SqlIdmAssignmentLifecycle"] diff --git a/src/govoplan_idm/backend/db/models.py b/src/govoplan_idm/backend/db/models.py index cc8599b..b373bd0 100644 --- a/src/govoplan_idm/backend/db/models.py +++ b/src/govoplan_idm/backend/db/models.py @@ -4,9 +4,20 @@ import uuid from datetime import datetime from typing import Any -from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, JSON, String, UniqueConstraint +from sqlalchemy import ( + Boolean, + DateTime, + ForeignKey, + Index, + Integer, + JSON, + String, + Text, + UniqueConstraint, +) from sqlalchemy.orm import Mapped, mapped_column +from govoplan_core.core.concurrency import strong_resource_etag from govoplan_core.db.base import Base, TimestampMixin @@ -59,4 +70,180 @@ class IdmTenantSettings(Base, TimestampMixin): settings: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False) -__all__ = ["IdmOrganizationFunctionAssignment", "IdmTenantSettings", "new_uuid"] +class IdmFunctionAssignmentChange(Base, TimestampMixin): + __tablename__ = "idm_function_assignment_changes" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "kind", + "initiator_account_id", + "idempotency_key", + name="uq_idm_function_assignment_change_idempotency", + ), + Index( + "ix_idm_function_assignment_changes_tenant_state", + "tenant_id", + "state", + "updated_at", + ), + Index( + "ix_idm_function_assignment_changes_candidate", + "tenant_id", + "candidate_identity_id", + "state", + ), + Index( + "ix_idm_function_assignment_changes_expiry", + "state", + "expires_at", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + kind: Mapped[str] = mapped_column(String(20), nullable=False, index=True) + state: Mapped[str] = mapped_column(String(40), nullable=False, index=True) + profile: Mapped[str] = mapped_column(String(60), nullable=False) + function_id: Mapped[str] = mapped_column( + ForeignKey("organizations_functions.id", ondelete="RESTRICT"), + nullable=False, + index=True, + ) + organization_unit_id: Mapped[str] = mapped_column( + String(36), nullable=False, index=True + ) + candidate_identity_id: Mapped[str] = mapped_column( + ForeignKey("identity_identities.id", ondelete="RESTRICT"), + nullable=False, + index=True, + ) + candidate_account_id: Mapped[str | None] = mapped_column( + String(36), nullable=True, index=True + ) + initiator_account_id: Mapped[str] = mapped_column( + String(36), nullable=False, index=True + ) + initiator_identity_id: Mapped[str | None] = mapped_column( + String(36), nullable=True, index=True + ) + represented_assignment_id: Mapped[str | None] = mapped_column( + ForeignKey("idm_organization_function_assignments.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + justification: Mapped[str] = mapped_column(Text, nullable=False) + evidence: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False) + requested_valid_from: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + requested_valid_until: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + applies_to_subunits: Mapped[bool] = mapped_column( + Boolean, default=False, nullable=False + ) + assignment_source: Mapped[str] = mapped_column( + String(50), default="governance", nullable=False + ) + required_steps: Mapped[list[str]] = mapped_column( + JSON, default=list, nullable=False + ) + completed_steps: Mapped[list[str]] = mapped_column( + JSON, default=list, nullable=False + ) + policy_decision: Mapped[dict[str, Any]] = mapped_column( + JSON, default=dict, nullable=False + ) + workflow_definition_id: Mapped[str | None] = mapped_column( + String(36), nullable=True, index=True + ) + workflow_definition_revision_id: Mapped[str | None] = mapped_column( + String(36), nullable=True + ) + workflow_definition_revision: Mapped[int | None] = mapped_column( + Integer, nullable=True + ) + workflow_definition_hash: Mapped[str | None] = mapped_column( + String(64), nullable=True + ) + workflow_instance_id: Mapped[str | None] = mapped_column( + String(36), nullable=True, unique=True + ) + workflow_current_step_id: Mapped[str | None] = mapped_column( + String(36), nullable=True + ) + resulting_assignment_id: Mapped[str | None] = mapped_column( + ForeignKey("idm_organization_function_assignments.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False) + expires_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + outcome_reason: Mapped[str | None] = mapped_column(Text, nullable=True) + resource_revision: Mapped[int] = mapped_column(Integer, default=1, nullable=False) + metadata_: Mapped[dict[str, Any]] = mapped_column( + "metadata", JSON, default=dict, nullable=False + ) + + @property + def strong_etag(self) -> str: + return strong_resource_etag( + "idm_function_assignment_change", + self.id, + self.resource_revision, + ) + + +class IdmFunctionAssignmentChangeEvent(Base): + __tablename__ = "idm_function_assignment_change_events" + __table_args__ = ( + UniqueConstraint( + "change_id", + "sequence", + name="uq_idm_function_assignment_change_event_sequence", + ), + Index( + "ix_idm_function_assignment_change_events_tenant_change", + "tenant_id", + "change_id", + "sequence", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + change_id: Mapped[str] = mapped_column( + ForeignKey("idm_function_assignment_changes.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + sequence: Mapped[int] = mapped_column(Integer, nullable=False) + action: Mapped[str] = mapped_column(String(50), nullable=False, index=True) + from_state: Mapped[str | None] = mapped_column(String(40), nullable=True) + to_state: Mapped[str] = mapped_column(String(40), nullable=False) + actor_account_id: Mapped[str | None] = mapped_column( + String(36), nullable=True, index=True + ) + actor_identity_id: Mapped[str | None] = mapped_column(String(36), nullable=True) + actor_assignment_id: Mapped[str | None] = mapped_column(String(36), nullable=True) + comment: Mapped[str | None] = mapped_column(Text, nullable=True) + evidence: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False) + policy_decision: Mapped[dict[str, Any]] = mapped_column( + JSON, default=dict, nullable=False + ) + workflow_step_id: Mapped[str | None] = mapped_column(String(36), nullable=True) + details: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False + ) + + +__all__ = [ + "IdmFunctionAssignmentChange", + "IdmFunctionAssignmentChangeEvent", + "IdmOrganizationFunctionAssignment", + "IdmTenantSettings", + "new_uuid", +] diff --git a/src/govoplan_idm/backend/function_assignment_changes.py b/src/govoplan_idm/backend/function_assignment_changes.py new file mode 100644 index 0000000..7a2626d --- /dev/null +++ b/src/govoplan_idm/backend/function_assignment_changes.py @@ -0,0 +1,1343 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from datetime import timedelta +import hashlib +import json + +from sqlalchemy import func, or_, select +from sqlalchemy.orm import Session + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.concurrency import claim_revision +from govoplan_core.core.events import ( + EventActorRef, + EventObjectRef, + EventTenantRef, + PlatformEvent, + emit_platform_event, +) +from govoplan_core.core.notifications import ( + NotificationDispatchRequest, + notification_dispatch_provider, +) +from govoplan_core.core.organizations import OrganizationFunctionRef +from govoplan_core.core.policy import ( + FunctionAssignmentGovernanceAction, + FunctionAssignmentGovernanceDecision, + FunctionAssignmentGovernanceRequest, + function_assignment_governance_policy, +) +from govoplan_core.core.principal_cache import invalidate_auth_principals +from govoplan_core.core.workflows import ( + WorkflowCurrentStepResolution, + WorkflowInstanceRef, + WorkflowStandardStartRequest, + workflow_orchestration_provider, +) +from govoplan_core.db.base import utcnow +from govoplan_core.security.time import utc_now +from govoplan_idm.backend.assignment_events import emit_assignment_event +from govoplan_idm.backend.assignment_transitions import ( + AssignmentTransitionError, + validate_assignment_shape, +) +from govoplan_idm.backend.db.models import ( + IdmFunctionAssignmentChange, + IdmFunctionAssignmentChangeEvent, + IdmOrganizationFunctionAssignment, + IdmTenantSettings, + new_uuid, +) + + +OPEN_STATES = { + "submitted", + "awaiting_holder", + "awaiting_authority", + "awaiting_recipient", + "changes_requested", + "blocked", + "failed_manual_review", +} +TERMINAL_STATES = { + "applied", + "rejected", + "withdrawn", + "expired", + "cancelled", +} +STEP_STATE = { + "holder": "awaiting_holder", + "authority": "awaiting_authority", + "recipient": "awaiting_recipient", +} +NODE_STEP = { + "holder_review": "holder", + "authority_review": "authority", + "recipient_review": "recipient", +} + + +class FunctionAssignmentChangeError(ValueError): + pass + + +class FunctionAssignmentChangeUnavailable(FunctionAssignmentChangeError): + pass + + +class FunctionAssignmentChangeConflict(FunctionAssignmentChangeError): + pass + + +def resolve_submission_capability( + session: Session, + *, + principal: ApiPrincipal, + registry: object | None, + kind: str, + function: OrganizationFunctionRef, + candidate_identity_id: str, + candidate_account_id: str | None, + has_evidence: bool, +) -> tuple[FunctionAssignmentGovernanceDecision | None, str | None]: + workflow = workflow_orchestration_provider(registry) + if workflow is None: + return None, "Workflow Engine is not installed or enabled." + policy = function_assignment_governance_policy(registry) + if policy is None: + return ( + None, + "Function assignment governance Policy is not installed or enabled.", + ) + settings = _effective_function_settings( + session, + tenant_id=principal.tenant_id, + function=function, + ) + authority_function_id = _authority_function_id(settings) + context = _actor_context( + session, + principal=principal, + function_id=function.id, + authority_function_id=authority_function_id, + candidate_identity_id=candidate_identity_id, + candidate_account_id=candidate_account_id, + initiator_account_id=principal.account_id, + has_evidence=has_evidence, + ) + decision = policy.resolve_function_assignment_action( + session, + request=FunctionAssignmentGovernanceRequest( + tenant_id=principal.tenant_id, + kind=kind, # type: ignore[arg-type] + action="submit", + function_id=function.id, + actor=principal.to_platform_principal(), + candidate_identity_id=candidate_identity_id, + candidate_account_id=candidate_account_id, + function_settings=settings, + context=context, + ), + ) + return decision, decision.reason + + +def create_function_assignment_change( + session: Session, + *, + principal: ApiPrincipal, + registry: object | None, + function: OrganizationFunctionRef, + payload: object, +) -> tuple[IdmFunctionAssignmentChange, bool]: + kind = str(getattr(payload, "kind")) + candidate_identity_id = str(getattr(payload, "candidate_identity_id")) + candidate_account_id = _text(getattr(payload, "candidate_account_id", None)) + idempotency_key = str(getattr(payload, "idempotency_key")).strip() + fingerprint = _request_fingerprint(payload) + existing = session.scalar( + select(IdmFunctionAssignmentChange).where( + IdmFunctionAssignmentChange.tenant_id == principal.tenant_id, + IdmFunctionAssignmentChange.kind == kind, + IdmFunctionAssignmentChange.initiator_account_id == principal.account_id, + IdmFunctionAssignmentChange.idempotency_key == idempotency_key, + ) + ) + if existing is not None: + if existing.metadata_.get("request_fingerprint") != fingerprint: + raise FunctionAssignmentChangeConflict( + "The idempotency key was already used with different request data." + ) + return existing, True + + decision, unavailable_reason = resolve_submission_capability( + session, + principal=principal, + registry=registry, + kind=kind, + function=function, + candidate_identity_id=candidate_identity_id, + candidate_account_id=candidate_account_id, + has_evidence=bool(getattr(payload, "evidence", ())), + ) + if decision is None: + raise FunctionAssignmentChangeUnavailable( + unavailable_reason + or "Governed function assignment changes are unavailable." + ) + if not decision.allowed: + raise FunctionAssignmentChangeUnavailable( + decision.reason or "The function assignment change is not allowed." + ) + _validate_requested_validity(payload, decision) + represented_assignment_id = _text( + getattr(payload, "represented_assignment_id", None) + ) + if represented_assignment_id is not None: + _require_actor_assignment( + session, + principal=principal, + assignment_id=represented_assignment_id, + function_id=function.id, + ) + now = utc_now() + change = IdmFunctionAssignmentChange( + id=new_uuid(), + tenant_id=principal.tenant_id, + kind=kind, + state="submitted", + profile=decision.profile, + function_id=function.id, + organization_unit_id=function.organization_unit_id, + candidate_identity_id=candidate_identity_id, + candidate_account_id=candidate_account_id, + initiator_account_id=principal.account_id, + initiator_identity_id=principal.identity_id, + represented_assignment_id=represented_assignment_id, + justification=str(getattr(payload, "justification")).strip(), + evidence=list(getattr(payload, "evidence", ())), + requested_valid_from=getattr(payload, "requested_valid_from", None), + requested_valid_until=getattr(payload, "requested_valid_until", None), + applies_to_subunits=bool(getattr(payload, "applies_to_subunits", False)), + assignment_source=str(getattr(payload, "assignment_source", "governance")), + required_steps=list(decision.required_steps), + completed_steps=[], + policy_decision=decision.to_dict(), + idempotency_key=idempotency_key, + expires_at=now + timedelta(hours=decision.request_expiry_hours), + resource_revision=1, + metadata_={ + **dict(getattr(payload, "metadata", {})), + "request_fingerprint": fingerprint, + "step_approvals": {}, + }, + ) + workflow = workflow_orchestration_provider(registry) + assert workflow is not None + try: + workflow_ref = workflow.start_standard( + session, + principal, + request=WorkflowStandardStartRequest( + tenant_id=principal.tenant_id, + origin_module_id="idm", + definition_key=f"function-assignment-{kind}", + idempotency_key=f"idm-function-change:{change.id}", + input={ + "change_id": change.id, + "kind": kind, + "function_id": function.id, + "candidate_identity_id": candidate_identity_id, + "profile": decision.profile, + "required_steps": list(decision.required_steps), + }, + actor_id=principal.account_id, + correlation_id=change.id, + ), + ) + workflow_ref = _align_workflow_to_required_step( + session, + workflow=workflow, + principal=principal, + change=change, + reference=workflow_ref, + ) + except (TypeError, ValueError) as exc: + raise FunctionAssignmentChangeUnavailable(str(exc)) from exc + _pin_workflow(change, workflow_ref) + next_step = _next_required_step(change) + blocked_reason = _missing_reviewer_reason(session, change) + if blocked_reason is not None: + change.state = "blocked" + change.outcome_reason = blocked_reason + elif next_step is None: + _apply_assignment( + session, + change=change, + principal=principal, + registry=registry, + ) + else: + change.state = STEP_STATE[next_step] + session.add(change) + session.flush() + _append_event( + session, + change=change, + action="submitted", + from_state=None, + actor=principal, + decision=decision, + details={"replayed": workflow_ref.replayed}, + ) + _emit_change_event( + session, + change, + event_type="idm.function_change.submitted.v1", + principal=principal, + ) + _notify_current_state(session, change=change, registry=registry) + return change, False + + +def transition_function_assignment_change( + session: Session, + *, + principal: ApiPrincipal, + registry: object | None, + change: IdmFunctionAssignmentChange, + function: OrganizationFunctionRef, + action: str, + base_revision: int, + comment: str | None, + evidence: Sequence[str], +) -> IdmFunctionAssignmentChange: + if change.state in TERMINAL_STATES: + raise FunctionAssignmentChangeConflict( + f"Function assignment change is already {change.state}." + ) + governance_action = _governance_action(change, action) + decision = _resolve_transition_decision( + session, + principal=principal, + registry=registry, + change=change, + function=function, + action=governance_action, + has_evidence=bool(evidence or change.evidence), + ) + if not decision.allowed: + raise FunctionAssignmentChangeUnavailable( + decision.reason or "This transition is not allowed." + ) + if ( + decision.separation_of_duties + and action in {"approve", "accept"} + and principal.account_id == change.initiator_account_id + and not ( + action == "accept" and principal.identity_id == change.candidate_identity_id + ) + ): + raise FunctionAssignmentChangeUnavailable( + "The initiator cannot perform this approval because separation of duties is enabled." + ) + next_revision = claim_revision( + session, + model=IdmFunctionAssignmentChange, + filters=( + IdmFunctionAssignmentChange.id == change.id, + IdmFunctionAssignmentChange.tenant_id == change.tenant_id, + ), + revision_attribute="resource_revision", + expected_revision=base_revision, + resource_type="idm_function_assignment_change", + resource_id=change.id, + refresh_path=f"/api/v1/idm/function-assignment-changes/{change.id}", + ) + change.resource_revision = next_revision + previous_state = change.state + if action in {"reject", "withdraw"}: + _finish_negative_transition( + session, + principal=principal, + registry=registry, + change=change, + action=action, + comment=comment, + evidence=evidence, + ) + elif action in {"approve", "accept"}: + _record_step_approval( + change, + step=_current_required_step(change), + actor_id=principal.account_id, + ) + if _step_approval_count(change, _current_required_step(change)) >= ( + 1 if action == "accept" else decision.quorum + ): + _complete_current_step( + session, + principal=principal, + registry=registry, + change=change, + comment=comment, + evidence=evidence, + ) + else: + change.policy_decision = decision.to_dict() + elif action == "request_changes": + _request_changes( + session, + principal=principal, + registry=registry, + change=change, + comment=comment, + evidence=evidence, + ) + elif action == "respond": + if not comment or not comment.strip(): + raise FunctionAssignmentChangeConflict( + "A response to requested changes requires a comment." + ) + resume_state = _text(change.metadata_.get("resume_state")) + if resume_state not in STEP_STATE.values(): + raise FunctionAssignmentChangeConflict( + "The previous review state is unavailable." + ) + change.state = resume_state + change.outcome_reason = None + change.metadata_ = { + **dict(change.metadata_), + "last_response": comment.strip(), + } + elif action == "recover": + change.outcome_reason = None + _resume_change( + session, + principal=principal, + registry=registry, + change=change, + ) + else: + raise FunctionAssignmentChangeConflict( + f"Unsupported function assignment change action: {action}." + ) + _append_event( + session, + change=change, + action=action, + from_state=previous_state, + actor=principal, + decision=decision, + comment=comment, + evidence=evidence, + ) + _emit_change_event( + session, + change, + event_type=f"idm.function_change.{action}.v1", + principal=principal, + ) + _notify_current_state(session, change=change, registry=registry) + session.flush() + return change + + +def available_change_actions( + session: Session, + *, + principal: ApiPrincipal, + registry: object | None, + change: IdmFunctionAssignmentChange, + function: OrganizationFunctionRef, +) -> tuple[list[str], str | None]: + if change.state in TERMINAL_STATES: + return [], f"The change is {change.state}." + candidates = ( + ["approve", "request_changes", "reject", "withdraw"] + if change.state in {"awaiting_holder", "awaiting_authority"} + else ["accept", "request_changes", "reject", "withdraw"] + if change.state == "awaiting_recipient" + else ["respond", "withdraw"] + if change.state == "changes_requested" + else ["recover", "withdraw"] + ) + available: list[str] = [] + reasons: list[str] = [] + for action in candidates: + decision = _resolve_transition_decision( + session, + principal=principal, + registry=registry, + change=change, + function=function, + action=_governance_action(change, action), + has_evidence=bool(change.evidence), + ) + if decision.allowed: + available.append(action) + elif decision.reason: + reasons.append(decision.reason) + return available, reasons[0] if not available and reasons else None + + +def change_events( + session: Session, + *, + change_id: str, +) -> list[IdmFunctionAssignmentChangeEvent]: + return list( + session.scalars( + select(IdmFunctionAssignmentChangeEvent) + .where(IdmFunctionAssignmentChangeEvent.change_id == change_id) + .order_by(IdmFunctionAssignmentChangeEvent.sequence.asc()) + ) + ) + + +def visible_change_filter( + session: Session, + principal: ApiPrincipal, +): + if principal.has("idm:function_change:admin") or principal.has( + "idm:organization_assignment:write" + ): + return None + actor_function_ids = tuple( + sorted({item.function_id for item in _actor_assignments(session, principal)}) + ) + clauses = [ + IdmFunctionAssignmentChange.initiator_account_id == principal.account_id, + IdmFunctionAssignmentChange.candidate_identity_id == principal.identity_id, + IdmFunctionAssignmentChange.candidate_account_id == principal.account_id, + ] + if actor_function_ids: + clauses.extend( + ( + IdmFunctionAssignmentChange.function_id.in_(actor_function_ids), + IdmFunctionAssignmentChange.policy_decision["authority_function_id"] + .as_string() + .in_(actor_function_ids), + ) + ) + return or_(*clauses) + + +def _resolve_transition_decision( + session: Session, + *, + principal: ApiPrincipal, + registry: object | None, + change: IdmFunctionAssignmentChange, + function: OrganizationFunctionRef, + action: FunctionAssignmentGovernanceAction, + has_evidence: bool, +) -> FunctionAssignmentGovernanceDecision: + policy = function_assignment_governance_policy(registry) + if policy is None: + raise FunctionAssignmentChangeUnavailable( + "Function assignment governance Policy is unavailable." + ) + settings = _effective_function_settings( + session, + tenant_id=change.tenant_id, + function=function, + ) + authority_function_id = _authority_function_id(settings) + context = _actor_context( + session, + principal=principal, + function_id=change.function_id, + authority_function_id=authority_function_id, + candidate_identity_id=change.candidate_identity_id, + candidate_account_id=change.candidate_account_id, + initiator_account_id=change.initiator_account_id, + has_evidence=has_evidence, + ) + context["approvals_complete"] = _next_required_step(change) is None + return policy.resolve_function_assignment_action( + session, + request=FunctionAssignmentGovernanceRequest( + tenant_id=change.tenant_id, + kind=change.kind, # type: ignore[arg-type] + action=action, + function_id=change.function_id, + actor=principal.to_platform_principal(), + candidate_identity_id=change.candidate_identity_id, + candidate_account_id=change.candidate_account_id, + current_state=change.state, + function_settings=settings, + context=context, + ), + ) + + +def _complete_current_step( + session: Session, + *, + principal: ApiPrincipal, + registry: object | None, + change: IdmFunctionAssignmentChange, + comment: str | None, + evidence: Sequence[str], +) -> None: + step = _current_required_step(change) + workflow = workflow_orchestration_provider(registry) + if workflow is None or change.workflow_instance_id is None: + raise FunctionAssignmentChangeUnavailable( + "The pinned Workflow instance is unavailable." + ) + reference = workflow.resolve_current_step( + session, + principal, + tenant_id=change.tenant_id, + instance_id=change.workflow_instance_id, + resolution=WorkflowCurrentStepResolution( + action="approve", + expected_step_id=change.workflow_current_step_id, + actor_id=principal.account_id, + output={"change_id": change.id, "completed_step": step}, + evidence=tuple(evidence), + comment=comment, + ), + ) + change.completed_steps = [*change.completed_steps, step] + reference = _align_workflow_to_required_step( + session, + workflow=workflow, + principal=principal, + change=change, + reference=reference, + ) + _pin_workflow(change, reference) + next_step = _next_required_step(change) + if next_step is None: + try: + _apply_assignment( + session, + change=change, + principal=principal, + registry=registry, + ) + except FunctionAssignmentChangeConflict as exc: + change.state = "failed_manual_review" + change.outcome_reason = str(exc) + else: + blocked_reason = _missing_reviewer_reason(session, change) + if blocked_reason is not None: + change.state = "blocked" + change.outcome_reason = blocked_reason + else: + change.state = STEP_STATE[next_step] + change.outcome_reason = None + + +def _finish_negative_transition( + session: Session, + *, + principal: ApiPrincipal, + registry: object | None, + change: IdmFunctionAssignmentChange, + action: str, + comment: str | None, + evidence: Sequence[str], +) -> None: + workflow = workflow_orchestration_provider(registry) + if workflow is None or change.workflow_instance_id is None: + raise FunctionAssignmentChangeUnavailable( + "The pinned Workflow instance is unavailable." + ) + resolution_action = "reject" if action == "reject" else "cancel" + reference = workflow.resolve_current_step( + session, + principal, + tenant_id=change.tenant_id, + instance_id=change.workflow_instance_id, + resolution=WorkflowCurrentStepResolution( + action=resolution_action, + expected_step_id=change.workflow_current_step_id, + actor_id=principal.account_id, + output={"change_id": change.id}, + evidence=tuple(evidence), + comment=comment, + ), + ) + _pin_workflow(change, reference) + change.state = "rejected" if action == "reject" else "withdrawn" + change.outcome_reason = comment + + +def _request_changes( + session: Session, + *, + principal: ApiPrincipal, + registry: object | None, + change: IdmFunctionAssignmentChange, + comment: str | None, + evidence: Sequence[str], +) -> None: + if not comment or not comment.strip(): + raise FunctionAssignmentChangeConflict("Requesting changes requires a comment.") + workflow = workflow_orchestration_provider(registry) + if workflow is None or change.workflow_instance_id is None: + raise FunctionAssignmentChangeUnavailable( + "The pinned Workflow instance is unavailable." + ) + previous_state = change.state + reference = workflow.resolve_current_step( + session, + principal, + tenant_id=change.tenant_id, + instance_id=change.workflow_instance_id, + resolution=WorkflowCurrentStepResolution( + action="changes", + expected_step_id=change.workflow_current_step_id, + actor_id=principal.account_id, + output={"change_id": change.id}, + evidence=tuple(evidence), + comment=comment, + ), + ) + _pin_workflow(change, reference) + change.state = "changes_requested" + change.outcome_reason = comment.strip() + change.metadata_ = { + **dict(change.metadata_), + "resume_state": previous_state, + } + + +def _resume_change( + session: Session, + *, + principal: ApiPrincipal, + registry: object | None, + change: IdmFunctionAssignmentChange, +) -> None: + workflow = workflow_orchestration_provider(registry) + if workflow is None or change.workflow_instance_id is None: + raise FunctionAssignmentChangeUnavailable( + "The pinned Workflow instance is unavailable." + ) + reference = workflow.get_instance( + session, + tenant_id=change.tenant_id, + instance_id=change.workflow_instance_id, + ) + reference = _align_workflow_to_required_step( + session, + workflow=workflow, + principal=principal, + change=change, + reference=reference, + ) + _pin_workflow(change, reference) + step = _next_required_step(change) + blocked_reason = _missing_reviewer_reason(session, change) + if blocked_reason is not None: + change.state = "blocked" + change.outcome_reason = blocked_reason + elif step is not None: + change.state = STEP_STATE[step] + change.outcome_reason = None + else: + try: + _apply_assignment( + session, + change=change, + principal=principal, + registry=registry, + ) + except FunctionAssignmentChangeConflict as exc: + change.state = "failed_manual_review" + change.outcome_reason = str(exc) + + +def _align_workflow_to_required_step( + session: Session, + *, + workflow: object, + principal: ApiPrincipal, + change: IdmFunctionAssignmentChange, + reference: WorkflowInstanceRef, +) -> WorkflowInstanceRef: + required = set(change.required_steps) + completed = set(change.completed_steps) + current = reference + for _ in range(4): + step = NODE_STEP.get(current.current_node_id or "") + if step is None or (step in required and step not in completed): + return current + current = workflow.resolve_current_step( # type: ignore[attr-defined] + session, + principal, + tenant_id=change.tenant_id, + instance_id=current.id, + resolution=WorkflowCurrentStepResolution( + action="approve", + expected_step_id=current.current_step_id, + actor_id=principal.account_id, + output={"skipped": True, "step": step}, + comment="Skipped by the effective assignment governance profile.", + ), + ) + raise FunctionAssignmentChangeConflict( + "Workflow baseline did not converge on the required governance step." + ) + + +def _apply_assignment( + session: Session, + *, + change: IdmFunctionAssignmentChange, + principal: ApiPrincipal, + registry: object | None, +) -> None: + if change.resulting_assignment_id: + change.state = "applied" + return + existing = session.scalar( + select(IdmOrganizationFunctionAssignment).where( + IdmOrganizationFunctionAssignment.tenant_id == change.tenant_id, + IdmOrganizationFunctionAssignment.identity_id + == change.candidate_identity_id, + IdmOrganizationFunctionAssignment.function_id == change.function_id, + IdmOrganizationFunctionAssignment.organization_unit_id + == change.organization_unit_id, + ) + ) + if existing is not None and existing.is_active: + raise FunctionAssignmentChangeConflict( + "The candidate already holds this organization function." + ) + assignment = existing or IdmOrganizationFunctionAssignment( + tenant_id=change.tenant_id, + identity_id=change.candidate_identity_id, + function_id=change.function_id, + organization_unit_id=change.organization_unit_id, + ) + assignment.account_id = change.candidate_account_id + assignment.applies_to_subunits = change.applies_to_subunits + assignment.source = change.assignment_source + assignment.delegated_from_assignment_id = change.represented_assignment_id + assignment.valid_from = change.requested_valid_from + assignment.valid_until = change.requested_valid_until + assignment.is_active = True + assignment.settings = { + **dict(assignment.settings or {}), + "governance": { + "change_id": change.id, + "workflow_instance_id": change.workflow_instance_id, + "profile": change.profile, + "applied_by": principal.account_id, + "applied_at": utc_now().isoformat(), + }, + } + try: + validate_assignment_shape(assignment) + except AssignmentTransitionError as exc: + raise FunctionAssignmentChangeConflict(str(exc)) from exc + if existing is None: + session.add(assignment) + session.flush() + change.resulting_assignment_id = assignment.id + change.state = "applied" + change.outcome_reason = None + emit_assignment_event( + session, + assignment, + event_type="idm.function_assignment.created.v1", + actor_type="account", + actor_id=principal.account_id, + ) + invalidate_auth_principals( + session, + tenant_id=change.tenant_id, + source_module="idm", + resource_type="organization_function_assignment", + resource_id=assignment.id, + ) + _emit_change_event( + session, + change, + event_type="idm.function_change.applied.v1", + principal=principal, + ) + del registry + + +def _actor_context( + session: Session, + *, + principal: ApiPrincipal, + function_id: str, + authority_function_id: str | None, + candidate_identity_id: str, + candidate_account_id: str | None, + initiator_account_id: str, + has_evidence: bool, +) -> dict[str, object]: + actor_assignments = _actor_assignments(session, principal) + return { + "actor_is_holder": any( + item.function_id == function_id for item in actor_assignments + ), + "actor_is_authority": bool(authority_function_id) + and any( + item.function_id == authority_function_id for item in actor_assignments + ), + "candidate_is_actor": ( + principal.identity_id == candidate_identity_id + or ( + candidate_account_id is not None + and principal.account_id == candidate_account_id + ) + ), + "actor_is_initiator": principal.account_id == initiator_account_id, + "has_evidence": has_evidence, + } + + +def _actor_assignments( + session: Session, + principal: ApiPrincipal, +) -> list[IdmOrganizationFunctionAssignment]: + now = utc_now() + return list( + session.scalars( + select(IdmOrganizationFunctionAssignment).where( + IdmOrganizationFunctionAssignment.tenant_id == principal.tenant_id, + IdmOrganizationFunctionAssignment.is_active.is_(True), + or_( + IdmOrganizationFunctionAssignment.account_id + == principal.account_id, + IdmOrganizationFunctionAssignment.identity_id + == principal.identity_id, + ), + or_( + IdmOrganizationFunctionAssignment.valid_from.is_(None), + IdmOrganizationFunctionAssignment.valid_from <= now, + ), + or_( + IdmOrganizationFunctionAssignment.valid_until.is_(None), + IdmOrganizationFunctionAssignment.valid_until > now, + ), + ) + ) + ) + + +def _effective_function_settings( + session: Session, + *, + tenant_id: str, + function: OrganizationFunctionRef, +) -> dict[str, object]: + tenant_settings = session.get(IdmTenantSettings, tenant_id) + defaults: dict[str, object] = {} + if tenant_settings is not None: + raw_defaults = tenant_settings.settings.get( + "function_assignment_governance_defaults" + ) + if isinstance(raw_defaults, Mapping): + defaults = dict(raw_defaults) + function_policy = function.settings.get("assignment_governance") + return { + **dict(function.settings), + "assignment_governance": { + **defaults, + **(dict(function_policy) if isinstance(function_policy, Mapping) else {}), + }, + } + + +def _authority_function_id(settings: Mapping[str, object]) -> str | None: + raw = settings.get("assignment_governance") + policy = raw if isinstance(raw, Mapping) else {} + return _text(policy.get("authority_function_id")) + + +def _validate_requested_validity( + payload: object, + decision: FunctionAssignmentGovernanceDecision, +) -> None: + valid_from = getattr(payload, "requested_valid_from", None) + valid_until = getattr(payload, "requested_valid_until", None) + if valid_from is not None and valid_until is not None and valid_until <= valid_from: + raise FunctionAssignmentChangeConflict( + "Requested valid until must be after valid from." + ) + if decision.maximum_validity_days is not None and valid_until is not None: + start = valid_from or utc_now() + if valid_until > start + timedelta(days=decision.maximum_validity_days): + raise FunctionAssignmentChangeUnavailable( + f"Requested validity exceeds the Policy limit of " + f"{decision.maximum_validity_days} days." + ) + + +def _require_actor_assignment( + session: Session, + *, + principal: ApiPrincipal, + assignment_id: str, + function_id: str, +) -> None: + assignment = session.get(IdmOrganizationFunctionAssignment, assignment_id) + if ( + assignment is None + or assignment.tenant_id != principal.tenant_id + or assignment.function_id != function_id + or not assignment.is_active + or ( + assignment.account_id != principal.account_id + and assignment.identity_id != principal.identity_id + ) + ): + raise FunctionAssignmentChangeUnavailable( + "The represented function assignment is not an effective assignment of the actor." + ) + + +def _next_required_step(change: IdmFunctionAssignmentChange) -> str | None: + completed = set(change.completed_steps) + return next( + (step for step in change.required_steps if step not in completed), + None, + ) + + +def _current_required_step(change: IdmFunctionAssignmentChange) -> str: + step = _next_required_step(change) + if step is None: + raise FunctionAssignmentChangeConflict( + "All required governance steps are already complete." + ) + return step + + +def _governance_action( + change: IdmFunctionAssignmentChange, + action: str, +) -> FunctionAssignmentGovernanceAction: + if action == "approve": + step = _current_required_step(change) + if step == "holder": + return "approve_holder" + if step == "authority": + return "approve_authority" + raise FunctionAssignmentChangeConflict( + "Recipient approval must use the accept action." + ) + if action == "accept": + if _current_required_step(change) != "recipient": + raise FunctionAssignmentChangeConflict( + "The change is not awaiting recipient acceptance." + ) + return "accept_recipient" + if action in { + "reject", + "request_changes", + "respond", + "withdraw", + "recover", + }: + return action # type: ignore[return-value] + raise FunctionAssignmentChangeConflict( + f"Unsupported function assignment change action: {action}." + ) + + +def _record_step_approval( + change: IdmFunctionAssignmentChange, + *, + step: str, + actor_id: str, +) -> None: + approvals = dict(change.metadata_.get("step_approvals") or {}) + actors = [str(item) for item in approvals.get(step, ())] + if actor_id in actors: + raise FunctionAssignmentChangeConflict( + "This actor already approved the current governance step." + ) + approvals[step] = [*actors, actor_id] + change.metadata_ = {**dict(change.metadata_), "step_approvals": approvals} + + +def _step_approval_count( + change: IdmFunctionAssignmentChange, + step: str, +) -> int: + approvals = change.metadata_.get("step_approvals") or {} + return len(approvals.get(step, ())) if isinstance(approvals, Mapping) else 0 + + +def _pin_workflow( + change: IdmFunctionAssignmentChange, + reference: WorkflowInstanceRef, +) -> None: + change.workflow_definition_id = reference.definition_id + change.workflow_definition_revision_id = reference.definition_revision_id + change.workflow_definition_revision = reference.definition_revision + change.workflow_definition_hash = reference.definition_hash + change.workflow_instance_id = reference.id + change.workflow_current_step_id = reference.current_step_id + + +def _append_event( + session: Session, + *, + change: IdmFunctionAssignmentChange, + action: str, + from_state: str | None, + actor: ApiPrincipal, + decision: FunctionAssignmentGovernanceDecision, + comment: str | None = None, + evidence: Sequence[str] = (), + details: Mapping[str, object] | None = None, +) -> None: + sequence = ( + int( + session.scalar( + select(func.max(IdmFunctionAssignmentChangeEvent.sequence)).where( + IdmFunctionAssignmentChangeEvent.change_id == change.id + ) + ) + or 0 + ) + + 1 + ) + session.add( + IdmFunctionAssignmentChangeEvent( + tenant_id=change.tenant_id, + change_id=change.id, + sequence=sequence, + action=action, + from_state=from_state, + to_state=change.state, + actor_account_id=actor.account_id, + actor_identity_id=actor.identity_id, + actor_assignment_id=( + change.represented_assignment_id + if actor.account_id == change.initiator_account_id + else None + ), + comment=comment, + evidence=list(evidence), + policy_decision=decision.to_dict(), + workflow_step_id=change.workflow_current_step_id, + details=dict(details or {}), + created_at=utcnow(), + ) + ) + + +def _emit_change_event( + session: Session, + change: IdmFunctionAssignmentChange, + *, + event_type: str, + principal: ApiPrincipal, +) -> None: + emit_platform_event( + session, + PlatformEvent( + type=event_type, + module_id="idm", + payload={ + "kind": change.kind, + "state": change.state, + "profile": change.profile, + "function_id": change.function_id, + "candidate_identity_id": change.candidate_identity_id, + "workflow_instance_id": change.workflow_instance_id, + "resulting_assignment_id": change.resulting_assignment_id, + "resource_revision": change.resource_revision, + }, + actor=EventActorRef(type="account", id=principal.account_id), + tenant=EventTenantRef(id=change.tenant_id), + subject=EventObjectRef( + type="organization_function", + id=change.function_id, + ), + resource=EventObjectRef( + type="function_assignment_change", + id=change.id, + ), + classification="internal", + ), + ) + + +def _notify_current_state( + session: Session, + *, + change: IdmFunctionAssignmentChange, + registry: object | None, +) -> None: + provider = notification_dispatch_provider(registry) + if provider is None: + return + recipients = _notification_recipients(session, change) + for account_id in recipients: + provider.enqueue_notification( + session, + NotificationDispatchRequest( + tenant_id=change.tenant_id, + source_module="idm", + source_resource_type="function_assignment_change", + source_resource_id=change.id, + event_kind=f"function_assignment_change.{change.state}", + recipient_type="account", + recipient_id=account_id, + subject=f"Function assignment change: {change.state.replace('_', ' ')}", + body_text=( + f"A {change.kind} for organization function " + f"{change.function_id} is {change.state.replace('_', ' ')}." + ), + action_url=f"/idm?change={change.id}", + payload={"change_id": change.id, "state": change.state}, + ), + ) + + +def _notification_recipients( + session: Session, + change: IdmFunctionAssignmentChange, +) -> tuple[str, ...]: + recipients = {change.initiator_account_id} + if change.candidate_account_id: + recipients.add(change.candidate_account_id) + if change.state == "awaiting_holder": + recipients.update( + _function_holder_accounts( + session, + tenant_id=change.tenant_id, + function_id=change.function_id, + ) + ) + elif change.state == "awaiting_authority": + authority_id = change.policy_decision.get("authority_function_id") + if authority_id: + recipients.update( + _function_holder_accounts( + session, + tenant_id=change.tenant_id, + function_id=str(authority_id), + ) + ) + return tuple(sorted(item for item in recipients if item)) + + +def _function_holder_accounts( + session: Session, + *, + tenant_id: str, + function_id: str, +) -> tuple[str, ...]: + now = utc_now() + return tuple( + item + for item in session.scalars( + select(IdmOrganizationFunctionAssignment.account_id).where( + IdmOrganizationFunctionAssignment.tenant_id == tenant_id, + IdmOrganizationFunctionAssignment.function_id == function_id, + IdmOrganizationFunctionAssignment.account_id.is_not(None), + IdmOrganizationFunctionAssignment.is_active.is_(True), + or_( + IdmOrganizationFunctionAssignment.valid_from.is_(None), + IdmOrganizationFunctionAssignment.valid_from <= now, + ), + or_( + IdmOrganizationFunctionAssignment.valid_until.is_(None), + IdmOrganizationFunctionAssignment.valid_until > now, + ), + ) + ) + if item is not None + ) + + +def _missing_reviewer_reason( + session: Session, + change: IdmFunctionAssignmentChange, +) -> str | None: + step = _next_required_step(change) + if step == "holder" and not _function_has_incumbent( + session, + tenant_id=change.tenant_id, + function_id=change.function_id, + ): + return "The function is vacant; no effective holder can review this change." + if step == "authority": + authority_id = _text(change.policy_decision.get("authority_function_id")) + if authority_id is None: + return "The effective Policy does not designate an authority function." + if not _function_has_incumbent( + session, + tenant_id=change.tenant_id, + function_id=authority_id, + ): + return "The designated authority function is vacant." + if step == "recipient" and not ( + change.candidate_account_id or change.candidate_identity_id + ): + return "Recipient acceptance has no resolvable candidate." + return None + + +def _function_has_incumbent( + session: Session, + *, + tenant_id: str, + function_id: str, +) -> bool: + now = utc_now() + return bool( + session.scalar( + select(func.count(IdmOrganizationFunctionAssignment.id)).where( + IdmOrganizationFunctionAssignment.tenant_id == tenant_id, + IdmOrganizationFunctionAssignment.function_id == function_id, + IdmOrganizationFunctionAssignment.is_active.is_(True), + or_( + IdmOrganizationFunctionAssignment.valid_from.is_(None), + IdmOrganizationFunctionAssignment.valid_from <= now, + ), + or_( + IdmOrganizationFunctionAssignment.valid_until.is_(None), + IdmOrganizationFunctionAssignment.valid_until > now, + ), + ) + ) + ) + + +def _request_fingerprint(payload: object) -> str: + if hasattr(payload, "model_dump"): + value = payload.model_dump(mode="json", exclude={"idempotency_key"}) + else: + value = vars(payload) + encoded = json.dumps(value, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def _text(value: object) -> str | None: + text = str(value).strip() if value is not None else "" + return text or None + + +__all__ = [ + "FunctionAssignmentChangeConflict", + "FunctionAssignmentChangeError", + "FunctionAssignmentChangeUnavailable", + "OPEN_STATES", + "TERMINAL_STATES", + "available_change_actions", + "change_events", + "create_function_assignment_change", + "resolve_submission_capability", + "transition_function_assignment_change", + "visible_change_filter", +] diff --git a/src/govoplan_idm/backend/manifest.py b/src/govoplan_idm/backend/manifest.py index 3993cdc..1668d1a 100644 --- a/src/govoplan_idm/backend/manifest.py +++ b/src/govoplan_idm/backend/manifest.py @@ -2,14 +2,29 @@ from __future__ import annotations from pathlib import Path -from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER -from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY, CAPABILITY_IDENTITY_SEARCH, IdentityDirectory +from govoplan_core.core.access import ( + CAPABILITY_AUTH_PERMISSION_EVALUATOR, + CAPABILITY_AUTH_PRINCIPAL_RESOLVER, +) +from govoplan_core.core.identity import ( + CAPABILITY_IDENTITY_DIRECTORY, + CAPABILITY_IDENTITY_SEARCH, + IdentityDirectory, +) from govoplan_core.core.idm import ( CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE, CAPABILITY_IDM_DIRECTORY, CAPABILITY_IDM_FUNCTION_ASSIGNMENTS, ) -from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY, OrganizationDirectory +from govoplan_core.core.notifications import CAPABILITY_NOTIFICATIONS_DISPATCH +from govoplan_core.core.policy import ( + CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE, +) +from govoplan_core.core.workflows import CAPABILITY_WORKFLOW_ORCHESTRATION +from govoplan_core.core.organizations import ( + CAPABILITY_ORGANIZATION_DIRECTORY, + OrganizationDirectory, +) from govoplan_core.core.views import ViewSurface from govoplan_core.core.module_guards import persistent_table_uninstall_guard from govoplan_core.core.modules import ( @@ -26,11 +41,22 @@ from govoplan_core.core.modules import ( ) from govoplan_core.db.base import Base from govoplan_idm.backend.db import models as idm_models # noqa: F401 - populate metadata +from govoplan_idm.backend.workflow_definitions import ( + function_assignment_workflow_definitions, +) + + +MODULE_VERSION = "0.1.8" IDM_READ_SCOPES = ( "idm:organization_assignment:read", "idm:organization_assignment:write", "idm:settings:read", + "idm:function_change:read", + "idm:function_request:create", + "idm:function_grant:create", + "idm:function_change:decide", + "idm:function_change:admin", "organizations:function:assign", ) @@ -75,6 +101,31 @@ PERMISSIONS = ( "Manage IDM settings", "Update IDM governance and assignment-change policy settings.", ), + _permission( + "idm:function_change:read", + "View function assignment changes", + "View governed function requests, grants, decisions, and outcomes.", + ), + _permission( + "idm:function_request:create", + "Request organization functions", + "Request assignment to an eligible organization function.", + ), + _permission( + "idm:function_grant:create", + "Propose organization function grants", + "Bestow an organization function through its governed grant profile.", + ), + _permission( + "idm:function_change:decide", + "Decide function assignment changes", + "Approve, reject, or accept governed function assignment changes when eligible.", + ), + _permission( + "idm:function_change:admin", + "Recover function assignment changes", + "Inspect and recover blocked or failed function assignment workflows.", + ), ) ROLE_TEMPLATES = ( @@ -88,6 +139,21 @@ ROLE_TEMPLATES = ( "idm:organization_assignment:write", "idm:settings:read", "idm:settings:write", + "idm:function_change:read", + "idm:function_request:create", + "idm:function_grant:create", + "idm:function_change:decide", + "idm:function_change:admin", + ), + ), + RoleTemplate( + slug="idm_function_participant", + name="IDM function participant", + description="Request functions and participate in governed assignment decisions.", + permissions=( + "idm:function_change:read", + "idm:function_request:create", + "idm:function_change:decide", ), ), ) @@ -113,20 +179,30 @@ def _idm_directory(context: ModuleContext) -> object: def _assignment_lifecycle(context: ModuleContext) -> object: - del context from govoplan_idm.backend.assignment_lifecycle import ( SqlIdmAssignmentLifecycle, ) - return SqlIdmAssignmentLifecycle() + return SqlIdmAssignmentLifecycle(registry=context.registry) manifest = ModuleManifest( id="idm", name="IDM", - version="0.1.8", + version=MODULE_VERSION, dependencies=("identity", "organizations"), - optional_dependencies=("access", "audit"), + optional_dependencies=( + "access", + "audit", + "notifications", + "policy", + "workflow_engine", + ), + optional_capabilities=( + CAPABILITY_NOTIFICATIONS_DISPATCH, + CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE, + CAPABILITY_WORKFLOW_ORCHESTRATION, + ), required_capabilities=( CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, @@ -141,7 +217,11 @@ manifest = ModuleManifest( ), ModuleInterfaceProvider( name=CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE, - version="0.1.8", + version=MODULE_VERSION, + ), + ModuleInterfaceProvider( + name="idm.function_assignment_changes", + version="1.0.0", ), ), permissions=PERMISSIONS, @@ -172,6 +252,8 @@ manifest = ModuleManifest( persistent_table_uninstall_guard( idm_models.IdmOrganizationFunctionAssignment, idm_models.IdmTenantSettings, + idm_models.IdmFunctionAssignmentChange, + idm_models.IdmFunctionAssignmentChangeEvent, label="IDM", ), ), @@ -180,6 +262,9 @@ manifest = ModuleManifest( CAPABILITY_IDM_DIRECTORY: _idm_directory, CAPABILITY_IDM_FUNCTION_ASSIGNMENTS: _idm_directory, }, + workflow_definitions=function_assignment_workflow_definitions( + module_version=MODULE_VERSION, + ), documentation=( DocumentationTopic( id="idm.organization_identity_bridge", diff --git a/src/govoplan_idm/backend/migrations/versions/a0b1c2d3e4f5_function_assignment_changes.py b/src/govoplan_idm/backend/migrations/versions/a0b1c2d3e4f5_function_assignment_changes.py new file mode 100644 index 0000000..d8a89bd --- /dev/null +++ b/src/govoplan_idm/backend/migrations/versions/a0b1c2d3e4f5_function_assignment_changes.py @@ -0,0 +1,226 @@ +"""Add governed function assignment change aggregates. + +Revision ID: a0b1c2d3e4f5 +Revises: 9a0b1c2d3e4f +Create Date: 2026-07-31 00:00:00.000000 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "a0b1c2d3e4f5" +down_revision = "9a0b1c2d3e4f" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "idm_function_assignment_changes", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("kind", sa.String(length=20), nullable=False), + sa.Column("state", sa.String(length=40), nullable=False), + sa.Column("profile", sa.String(length=60), nullable=False), + sa.Column("function_id", sa.String(length=36), nullable=False), + sa.Column("organization_unit_id", sa.String(length=36), nullable=False), + sa.Column("candidate_identity_id", sa.String(length=36), nullable=False), + sa.Column("candidate_account_id", sa.String(length=36), nullable=True), + sa.Column("initiator_account_id", sa.String(length=36), nullable=False), + sa.Column("initiator_identity_id", sa.String(length=36), nullable=True), + sa.Column("represented_assignment_id", sa.String(length=36), nullable=True), + sa.Column("justification", sa.Text(), nullable=False), + sa.Column("evidence", sa.JSON(), nullable=False), + sa.Column("requested_valid_from", sa.DateTime(timezone=True), nullable=True), + sa.Column("requested_valid_until", sa.DateTime(timezone=True), nullable=True), + sa.Column("applies_to_subunits", sa.Boolean(), nullable=False), + sa.Column("assignment_source", sa.String(length=50), nullable=False), + sa.Column("required_steps", sa.JSON(), nullable=False), + sa.Column("completed_steps", sa.JSON(), nullable=False), + sa.Column("policy_decision", sa.JSON(), nullable=False), + sa.Column("workflow_definition_id", sa.String(length=36), nullable=True), + sa.Column( + "workflow_definition_revision_id", sa.String(length=36), nullable=True + ), + sa.Column("workflow_definition_revision", sa.Integer(), nullable=True), + sa.Column("workflow_definition_hash", sa.String(length=64), nullable=True), + sa.Column("workflow_instance_id", sa.String(length=36), nullable=True), + sa.Column("workflow_current_step_id", sa.String(length=36), nullable=True), + sa.Column("resulting_assignment_id", sa.String(length=36), nullable=True), + sa.Column("idempotency_key", sa.String(length=255), nullable=False), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("outcome_reason", sa.Text(), nullable=True), + sa.Column("resource_revision", sa.Integer(), nullable=False), + sa.Column("metadata", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["candidate_identity_id"], + ["identity_identities.id"], + name=op.f( + "fk_idm_function_assignment_changes_candidate_identity_id_identity_identities" + ), + ondelete="RESTRICT", + ), + sa.ForeignKeyConstraint( + ["function_id"], + ["organizations_functions.id"], + name=op.f( + "fk_idm_function_assignment_changes_function_id_organizations_functions" + ), + ondelete="RESTRICT", + ), + sa.ForeignKeyConstraint( + ["represented_assignment_id"], + ["idm_organization_function_assignments.id"], + name=op.f( + "fk_idm_function_assignment_changes_represented_assignment_id_idm_assignments" + ), + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["resulting_assignment_id"], + ["idm_organization_function_assignments.id"], + name=op.f( + "fk_idm_function_assignment_changes_resulting_assignment_id_idm_assignments" + ), + ondelete="SET NULL", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_idm_function_assignment_changes")), + sa.UniqueConstraint( + "tenant_id", + "kind", + "initiator_account_id", + "idempotency_key", + name="uq_idm_function_assignment_change_idempotency", + ), + sa.UniqueConstraint( + "workflow_instance_id", + name=op.f("uq_idm_function_assignment_changes_workflow_instance_id"), + ), + ) + for name, columns in ( + ("ix_idm_function_assignment_changes_tenant_id", ["tenant_id"]), + ("ix_idm_function_assignment_changes_kind", ["kind"]), + ("ix_idm_function_assignment_changes_state", ["state"]), + ("ix_idm_function_assignment_changes_function_id", ["function_id"]), + ( + "ix_idm_function_assignment_changes_organization_unit_id", + ["organization_unit_id"], + ), + ( + "ix_idm_function_assignment_changes_candidate_identity_id", + ["candidate_identity_id"], + ), + ( + "ix_idm_function_assignment_changes_candidate_account_id", + ["candidate_account_id"], + ), + ( + "ix_idm_function_assignment_changes_initiator_account_id", + ["initiator_account_id"], + ), + ( + "ix_idm_function_assignment_changes_initiator_identity_id", + ["initiator_identity_id"], + ), + ( + "ix_idm_function_assignment_changes_represented_assignment_id", + ["represented_assignment_id"], + ), + ( + "ix_idm_function_assignment_changes_workflow_definition_id", + ["workflow_definition_id"], + ), + ( + "ix_idm_function_assignment_changes_resulting_assignment_id", + ["resulting_assignment_id"], + ), + ( + "ix_idm_function_assignment_changes_tenant_state", + ["tenant_id", "state", "updated_at"], + ), + ( + "ix_idm_function_assignment_changes_candidate", + ["tenant_id", "candidate_identity_id", "state"], + ), + ( + "ix_idm_function_assignment_changes_expiry", + ["state", "expires_at"], + ), + ): + op.create_index(name, "idm_function_assignment_changes", columns, unique=False) + + op.create_table( + "idm_function_assignment_change_events", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("change_id", sa.String(length=36), nullable=False), + sa.Column("sequence", sa.Integer(), nullable=False), + sa.Column("action", sa.String(length=50), nullable=False), + sa.Column("from_state", sa.String(length=40), nullable=True), + sa.Column("to_state", sa.String(length=40), nullable=False), + sa.Column("actor_account_id", sa.String(length=36), nullable=True), + sa.Column("actor_identity_id", sa.String(length=36), nullable=True), + sa.Column("actor_assignment_id", sa.String(length=36), nullable=True), + sa.Column("comment", sa.Text(), nullable=True), + sa.Column("evidence", sa.JSON(), nullable=False), + sa.Column("policy_decision", sa.JSON(), nullable=False), + sa.Column("workflow_step_id", sa.String(length=36), nullable=True), + sa.Column("details", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["change_id"], + ["idm_function_assignment_changes.id"], + name=op.f( + "fk_idm_function_assignment_change_events_change_id_idm_function_assignment_changes" + ), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint( + "id", name=op.f("pk_idm_function_assignment_change_events") + ), + sa.UniqueConstraint( + "change_id", + "sequence", + name="uq_idm_function_assignment_change_event_sequence", + ), + ) + op.create_index( + "ix_idm_function_assignment_change_events_tenant_id", + "idm_function_assignment_change_events", + ["tenant_id"], + unique=False, + ) + op.create_index( + "ix_idm_function_assignment_change_events_change_id", + "idm_function_assignment_change_events", + ["change_id"], + unique=False, + ) + op.create_index( + "ix_idm_function_assignment_change_events_action", + "idm_function_assignment_change_events", + ["action"], + unique=False, + ) + op.create_index( + "ix_idm_function_assignment_change_events_actor_account_id", + "idm_function_assignment_change_events", + ["actor_account_id"], + unique=False, + ) + op.create_index( + "ix_idm_function_assignment_change_events_tenant_change", + "idm_function_assignment_change_events", + ["tenant_id", "change_id", "sequence"], + unique=False, + ) + + +def downgrade() -> None: + op.drop_table("idm_function_assignment_change_events") + op.drop_table("idm_function_assignment_changes") diff --git a/src/govoplan_idm/backend/workflow_definitions.py b/src/govoplan_idm/backend/workflow_definitions.py new file mode 100644 index 0000000..137e2d5 --- /dev/null +++ b/src/govoplan_idm/backend/workflow_definitions.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +from govoplan_core.core.workflows import WorkflowDefinitionContribution + + +def function_assignment_workflow_definitions( + *, + module_version: str, +) -> tuple[WorkflowDefinitionContribution, ...]: + return ( + _contribution( + module_version=module_version, + definition_key="function-assignment-request", + name="Request an organization function", + description=( + "Governed holder, authority, and optional recipient decisions " + "for a self-requested organization function assignment." + ), + kind="request", + ), + _contribution( + module_version=module_version, + definition_key="function-assignment-grant", + name="Bestow an organization function", + description=( + "Governed holder, authority, and recipient decisions for an " + "organization function grant." + ), + kind="grant", + ), + ) + + +def _contribution( + *, + module_version: str, + definition_key: str, + name: str, + description: str, + kind: str, +) -> WorkflowDefinitionContribution: + return WorkflowDefinitionContribution( + origin_module_id="idm", + origin_module_version=module_version, + definition_key=definition_key, + name=name, + description=description, + graph=_graph(kind=kind), + scope_type="system", + inherit_to_lower_scopes=True, + allow_start=True, + allow_reuse=False, + allow_automation=False, + execution_mode="guided", + activate_on_install=True, + metadata={ + "domain": "idm.function_assignment_change", + "change_kind": kind, + "state_owner": "idm", + }, + policy_metadata={ + "governance_capability": ("policy.functionAssignmentGovernance"), + }, + ) + + +def _graph(*, kind: str) -> dict[str, object]: + return { + "schema_version": 1, + "nodes": [ + { + "id": "start", + "type": "workflow.start.manual", + "label": "Submitted", + "position": {"x": 20, "y": 120}, + "config": { + "input_schema_ref": (f"govoplan/idm/function-assignment-{kind}.v1"), + }, + }, + { + "id": "holder_review", + "type": "workflow.review", + "label": "Holder review", + "position": {"x": 230, "y": 120}, + "config": { + "title": "Holder review", + "reviewer": "effective-holder", + "required_evidence": [], + "view_surface_ids": [ + "idm.action.view-function-assignments", + ], + }, + }, + { + "id": "authority_review", + "type": "workflow.review", + "label": "Authority clearance", + "position": {"x": 470, "y": 120}, + "config": { + "title": "Authority clearance", + "reviewer": "designated-authority", + "required_evidence": [], + "view_surface_ids": [ + "idm.action.view-function-assignments", + ], + }, + }, + { + "id": "recipient_review", + "type": "workflow.review", + "label": "Recipient acceptance", + "position": {"x": 730, "y": 120}, + "config": { + "title": "Recipient acceptance", + "reviewer": "candidate", + "required_evidence": [], + "view_surface_ids": [ + "idm.action.view-function-assignments", + ], + }, + }, + { + "id": "completed", + "type": "workflow.end.completed", + "label": "Approved", + "position": {"x": 990, "y": 70}, + "config": {"output_mapping": {}}, + }, + { + "id": "rejected", + "type": "workflow.end.cancelled", + "label": "Rejected", + "position": {"x": 990, "y": 230}, + "config": {"reason": "Function assignment change rejected"}, + }, + ], + "edges": [ + { + "id": "start-holder", + "source": "start", + "target": "holder_review", + }, + { + "id": "holder-authority", + "source": "holder_review", + "source_port": "approved", + "target": "authority_review", + }, + { + "id": "holder-rejected", + "source": "holder_review", + "source_port": "rejected", + "target": "rejected", + }, + { + "id": "authority-recipient", + "source": "authority_review", + "source_port": "approved", + "target": "recipient_review", + }, + { + "id": "authority-rejected", + "source": "authority_review", + "source_port": "rejected", + "target": "rejected", + }, + { + "id": "recipient-completed", + "source": "recipient_review", + "source_port": "approved", + "target": "completed", + }, + { + "id": "recipient-rejected", + "source": "recipient_review", + "source_port": "rejected", + "target": "rejected", + }, + ], + "metadata": { + "notation": "govoplan.workflow.native", + "domain": "idm.function_assignment_change", + "change_kind": kind, + "optional_steps": [ + "holder_review", + "authority_review", + "recipient_review", + ], + }, + } + + +__all__ = ["function_assignment_workflow_definitions"] diff --git a/tests/test_assignment_expiry.py b/tests/test_assignment_expiry.py index c58afb8..33f1ed5 100644 --- a/tests/test_assignment_expiry.py +++ b/tests/test_assignment_expiry.py @@ -9,7 +9,11 @@ from govoplan_core.db.base import Base from govoplan_core.db.session import configure_database, reset_database from govoplan_identity.backend.db import models as identity_models # noqa: F401 from govoplan_idm.backend.assignment_lifecycle import SqlIdmAssignmentLifecycle -from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment +from govoplan_idm.backend.db.models import ( + IdmFunctionAssignmentChange, + IdmFunctionAssignmentChangeEvent, + IdmOrganizationFunctionAssignment, +) from govoplan_organizations.backend.db import models as organization_models # noqa: F401 @@ -20,6 +24,8 @@ class AssignmentExpiryTests(unittest.TestCase): self.database.engine, tables=[ IdmOrganizationFunctionAssignment.__table__, + IdmFunctionAssignmentChange.__table__, + IdmFunctionAssignmentChangeEvent.__table__, ChangeSequenceEntry.__table__, ], ) @@ -143,6 +149,61 @@ class AssignmentExpiryTests(unittest.TestCase): with self.subTest(limit=value), self.assertRaises(ValueError): self.lifecycle.process_expired(session, limit=value) + def test_sweep_expires_open_governed_changes_once(self) -> None: + boundary = datetime(2026, 7, 31, 12, tzinfo=timezone.utc) + with self.database.session() as session: + session.add( + IdmFunctionAssignmentChange( + id="change-due", + tenant_id="tenant-1", + kind="request", + state="awaiting_holder", + profile="holder_grant", + function_id="function-1", + organization_unit_id="unit-1", + candidate_identity_id="identity-1", + initiator_account_id="account-1", + justification="Need the function", + evidence=[], + assignment_source="governance", + required_steps=["holder"], + completed_steps=[], + policy_decision={}, + idempotency_key="request-1", + expires_at=boundary - timedelta(seconds=1), + metadata_={}, + ) + ) + session.commit() + + events: list[PlatformEvent] = [] + bus = EventBus() + bus.subscribe("idm.function_change.expired.v1", events.append) + with self.database.SessionLocal() as session, event_bus_context(bus): + result = self.lifecycle.process_expired( + session, + tenant_id="tenant-1", + effective_at=boundary, + ) + session.commit() + repeated = self.lifecycle.process_expired( + session, + tenant_id="tenant-1", + effective_at=boundary, + ) + session.commit() + + self.assertEqual(1, result["expired_changes"]) + self.assertEqual(["change-due"], result["change_ids"]) + self.assertEqual(0, repeated["expired_changes"]) + self.assertEqual(1, len(events)) + with self.database.session() as session: + change = session.get(IdmFunctionAssignmentChange, "change-due") + self.assertEqual("expired", change.state) + self.assertEqual(2, change.resource_revision) + history = session.query(IdmFunctionAssignmentChangeEvent).all() + self.assertEqual(["expired"], [item.action for item in history]) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_function_assignment_changes.py b/tests/test_function_assignment_changes.py new file mode 100644 index 0000000..cf5bb38 --- /dev/null +++ b/tests/test_function_assignment_changes.py @@ -0,0 +1,375 @@ +from __future__ import annotations + +from dataclasses import replace +import unittest +from unittest.mock import patch + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.change_sequence import ChangeSequenceEntry +from govoplan_core.core.concurrency import RevisionConflictError +from govoplan_core.core.organizations import OrganizationFunctionRef +from govoplan_core.core.policy import FunctionAssignmentGovernanceDecision +from govoplan_core.core.workflows import WorkflowInstanceRef +from govoplan_core.db.base import Base +from govoplan_core.db.session import configure_database, reset_database +from govoplan_identity.backend.db import models as identity_models # noqa: F401 +from govoplan_idm.backend.api.v1.schemas import FunctionAssignmentChangeCreateRequest +from govoplan_idm.backend.api.v1.function_changes import _change_item +from govoplan_idm.backend.db.models import ( + IdmFunctionAssignmentChange, + IdmFunctionAssignmentChangeEvent, + IdmOrganizationFunctionAssignment, + IdmTenantSettings, +) +from govoplan_idm.backend.function_assignment_changes import ( + create_function_assignment_change, + transition_function_assignment_change, +) +from govoplan_organizations.backend.db import models as organization_models # noqa: F401 + + +class _Policy: + def resolve_function_assignment_action(self, session=None, *, request): + del session + steps = ("holder", "authority") + context = request.context + allowed = { + "submit": bool(context.get("candidate_is_actor")), + "approve_holder": bool(context.get("actor_is_holder")), + "approve_authority": bool(context.get("actor_is_authority")), + "accept_recipient": bool(context.get("candidate_is_actor")), + "request_changes": bool( + context.get("actor_is_holder") or context.get("actor_is_authority") + ), + "respond": bool( + context.get("actor_is_initiator") or context.get("candidate_is_actor") + ), + "reject": bool( + context.get("actor_is_holder") or context.get("actor_is_authority") + ), + "withdraw": bool(context.get("actor_is_initiator")), + "recover": request.actor.account_id == "admin", + "apply": bool(context.get("approvals_complete")), + }.get(request.action, False) + return FunctionAssignmentGovernanceDecision( + allowed=allowed, + reason=None if allowed else "Not eligible for this action.", + profile="holder_with_authority_clearance", + required_steps=steps, + authority_function_id="authority-function", + separation_of_duties=False, + request_expiry_hours=24, + ) + + +class _Workflow: + nodes = ("holder_review", "authority_review", "recipient_review") + + def __init__(self) -> None: + self.instances: dict[str, WorkflowInstanceRef] = {} + + def start_standard(self, session, principal, *, request): + del session, principal + instance_id = f"workflow-{request.idempotency_key}" + existing = self.instances.get(instance_id) + if existing is not None: + return replace(existing, replayed=True) + reference = WorkflowInstanceRef( + id=instance_id, + tenant_id=request.tenant_id, + definition_id="definition-1", + definition_revision_id="revision-1", + definition_revision=1, + definition_hash="a" * 64, + status="waiting", + current_step_id="step-holder_review", + current_node_id="holder_review", + ) + self.instances[instance_id] = reference + return reference + + def resolve_current_step( + self, + session, + principal, + *, + tenant_id, + instance_id, + resolution, + ): + del session, principal, tenant_id + current = self.instances[instance_id] + if resolution.expected_step_id != current.current_step_id: + raise ValueError("Workflow current step changed.") + if resolution.action == "changes": + return current + if resolution.action in {"reject", "cancel"}: + result = replace( + current, + status="completed", + current_step_id=None, + current_node_id=None, + ) + else: + index = self.nodes.index(current.current_node_id or "") + 1 + if index >= len(self.nodes): + result = replace( + current, + status="completed", + current_step_id=None, + current_node_id=None, + ) + else: + node = self.nodes[index] + result = replace( + current, + current_step_id=f"step-{node}", + current_node_id=node, + ) + self.instances[instance_id] = result + return result + + def get_instance(self, session, *, tenant_id, instance_id): + del session, tenant_id + return self.instances[instance_id] + + +class _Registry: + def __init__(self) -> None: + self.policy = _Policy() + self.workflow = _Workflow() + + def has_capability(self, name: str) -> bool: + return name in { + "policy.functionAssignmentGovernance", + "workflow.orchestration", + } + + def capability(self, name: str): + if name == "policy.functionAssignmentGovernance": + return self.policy + if name == "workflow.orchestration": + return self.workflow + return None + + +def principal(account_id: str, identity_id: str) -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id=account_id, + membership_id=f"membership-{account_id}", + tenant_id="tenant-1", + identity_id=identity_id, + scopes=frozenset({"idm:function_change:decide"}), + ), + account=object(), + user=object(), + ) + + +def function() -> OrganizationFunctionRef: + return OrganizationFunctionRef( + id="target-function", + tenant_id="tenant-1", + organization_unit_id="unit-1", + slug="target", + name="Target function", + settings={ + "assignment_governance": { + "request_profile": "holder_with_authority_clearance", + "authority_function_id": "authority-function", + } + }, + ) + + +def payload() -> FunctionAssignmentChangeCreateRequest: + return FunctionAssignmentChangeCreateRequest( + kind="request", + function_id="target-function", + candidate_identity_id="candidate-identity", + candidate_account_id="candidate", + justification="The function is needed for the assigned work.", + idempotency_key="request-1", + ) + + +class FunctionAssignmentChangeTests(unittest.TestCase): + def setUp(self) -> None: + self.database = configure_database("sqlite:///:memory:") + Base.metadata.create_all( + self.database.engine, + tables=[ + IdmOrganizationFunctionAssignment.__table__, + IdmFunctionAssignmentChange.__table__, + IdmFunctionAssignmentChangeEvent.__table__, + IdmTenantSettings.__table__, + ChangeSequenceEntry.__table__, + ], + ) + self.registry = _Registry() + + def tearDown(self) -> None: + reset_database(dispose=True) + + def _add_reviewer_assignments(self, session) -> None: + session.add_all( + ( + IdmOrganizationFunctionAssignment( + id="holder-assignment", + tenant_id="tenant-1", + identity_id="holder-identity", + account_id="holder", + function_id="target-function", + organization_unit_id="unit-1", + source="direct", + is_active=True, + settings={}, + ), + IdmOrganizationFunctionAssignment( + id="authority-assignment", + tenant_id="tenant-1", + identity_id="authority-identity", + account_id="authority", + function_id="authority-function", + organization_unit_id="unit-1", + source="direct", + is_active=True, + settings={}, + ), + ) + ) + session.flush() + + def test_request_is_idempotent_and_applies_once_after_required_steps(self) -> None: + with self.database.session() as session: + self._add_reviewer_assignments(session) + change, replayed = create_function_assignment_change( + session, + principal=principal("candidate", "candidate-identity"), + registry=self.registry, + function=function(), + payload=payload(), + ) + session.flush() + same, replay = create_function_assignment_change( + session, + principal=principal("candidate", "candidate-identity"), + registry=self.registry, + function=function(), + payload=payload(), + ) + self.assertFalse(replayed) + self.assertTrue(replay) + self.assertEqual(change.id, same.id) + self.assertEqual("awaiting_holder", change.state) + + transition_function_assignment_change( + session, + principal=principal("holder", "holder-identity"), + registry=self.registry, + change=change, + function=function(), + action="approve", + base_revision=1, + comment="Holder approval", + evidence=(), + ) + self.assertEqual("awaiting_authority", change.state) + transition_function_assignment_change( + session, + principal=principal("authority", "authority-identity"), + registry=self.registry, + change=change, + function=function(), + action="approve", + base_revision=2, + comment="Authority approval", + evidence=(), + ) + session.commit() + + self.assertEqual("applied", change.state) + self.assertIsNotNone(change.resulting_assignment_id) + resulting = session.get( + IdmOrganizationFunctionAssignment, + change.resulting_assignment_id, + ) + self.assertEqual("candidate-identity", resulting.identity_id) + self.assertEqual("governance", resulting.source) + self.assertEqual( + change.id, + resulting.settings["governance"]["change_id"], + ) + self.assertEqual( + 1, + session.query(IdmOrganizationFunctionAssignment) + .filter( + IdmOrganizationFunctionAssignment.identity_id + == "candidate-identity" + ) + .count(), + ) + + def test_vacant_function_blocks_and_revision_claim_rejects_stale_action( + self, + ) -> None: + with self.database.session() as session: + change, _ = create_function_assignment_change( + session, + principal=principal("candidate", "candidate-identity"), + registry=self.registry, + function=function(), + payload=payload(), + ) + session.flush() + self.assertEqual("blocked", change.state) + self.assertIn("vacant", change.outcome_reason) + + self._add_reviewer_assignments(session) + change.state = "awaiting_holder" + change.resource_revision = 2 + session.flush() + with self.assertRaises(RevisionConflictError): + transition_function_assignment_change( + session, + principal=principal("holder", "holder-identity"), + registry=self.registry, + change=change, + function=function(), + action="approve", + base_revision=1, + comment=None, + evidence=(), + ) + + def test_historical_change_remains_readable_after_function_removal(self) -> None: + with self.database.session() as session: + change, _ = create_function_assignment_change( + session, + principal=principal("candidate", "candidate-identity"), + registry=self.registry, + function=function(), + payload=payload(), + ) + session.flush() + + with patch( + "govoplan_idm.backend.api.v1.function_changes._historical_function", + return_value=None, + ): + item = _change_item( + session, + principal("candidate", "candidate-identity"), + change, + include_events=False, + ) + + self.assertEqual(change.id, item.id) + self.assertEqual([], item.available_actions) + self.assertIn("no longer available", item.availability_reason) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/src/api/idm.ts b/webui/src/api/idm.ts index f673915..090eb22 100644 --- a/webui/src/api/idm.ts +++ b/webui/src/api/idm.ts @@ -99,10 +99,90 @@ export type OrganizationFunctionAssignmentPayload = { is_active?: boolean; settings?: Record; change_request_id?: string | null; + governance_override_reason?: string | null; + governance_override_evidence?: string[]; }; export type IdmSettingsPayload = Partial>; +export type FunctionAssignmentChangeKind = "request" | "grant"; +export type FunctionAssignmentChangeAction = "approve" | "reject" | "accept" | "request_changes" | "respond" | "withdraw" | "recover"; + +export type FunctionAssignmentChangeEvent = { + id: string; + sequence: number; + action: string; + from_state?: string | null; + to_state: string; + actor_account_id?: string | null; + actor_identity_id?: string | null; + comment?: string | null; + evidence: string[]; + policy_decision: Record; + workflow_step_id?: string | null; + details: Record; + created_at: string; +}; + +export type FunctionAssignmentChange = { + id: string; + tenant_id: string; + kind: FunctionAssignmentChangeKind; + state: string; + profile: string; + function_id: string; + organization_unit_id: string; + candidate_identity_id: string; + candidate_account_id?: string | null; + initiator_account_id: string; + initiator_identity_id?: string | null; + justification: string; + evidence: string[]; + requested_valid_from?: string | null; + requested_valid_until?: string | null; + required_steps: string[]; + completed_steps: string[]; + policy_decision: Record; + workflow_definition_revision?: number | null; + workflow_definition_hash?: string | null; + workflow_instance_id?: string | null; + resulting_assignment_id?: string | null; + expires_at?: string | null; + outcome_reason?: string | null; + resource_revision: number; + etag: string; + metadata: Record; + events: FunctionAssignmentChangeEvent[]; + available_actions: FunctionAssignmentChangeAction[]; + availability_reason?: string | null; + created_at: string; + updated_at: string; +}; + +export type FunctionAssignmentChangeList = { + changes: FunctionAssignmentChange[]; + total: number; + page: number; + page_size: number; + pages: number; +}; + +export type FunctionAssignmentChangePayload = { + kind: FunctionAssignmentChangeKind; + function_id: string; + candidate_identity_id: string; + candidate_account_id?: string | null; + justification: string; + evidence?: string[]; + requested_valid_from?: string | null; + requested_valid_until?: string | null; + applies_to_subunits?: boolean; + assignment_source?: "governance" | "delegated"; + represented_assignment_id?: string | null; + idempotency_key: string; + metadata?: Record; +}; + function post>(settings: ApiSettings, path: string, payload: P): Promise { return apiFetch(settings, path, { method: "POST", body: JSON.stringify(payload) }); } @@ -168,3 +248,40 @@ export function patchOrganizationFunctionAssignment( ): Promise { return patch(settings, `/api/v1/idm/organization-function-assignments/${encodeURIComponent(id)}`, payload); } + +export function getFunctionAssignmentChanges(settings: ApiSettings): Promise { + return apiFetch(settings, "/api/v1/idm/function-assignment-changes?page_size=200"); +} + +export function getFunctionAssignmentChange(settings: ApiSettings, id: string): Promise { + return apiFetch(settings, `/api/v1/idm/function-assignment-changes/${encodeURIComponent(id)}`); +} + +export function createFunctionAssignmentChange( + settings: ApiSettings, + payload: FunctionAssignmentChangePayload +): Promise { + return post(settings, "/api/v1/idm/function-assignment-changes", payload); +} + +export function actOnFunctionAssignmentChange( + settings: ApiSettings, + change: FunctionAssignmentChange, + action: FunctionAssignmentChangeAction, + comment?: string +): Promise { + return apiFetch( + settings, + `/api/v1/idm/function-assignment-changes/${encodeURIComponent(change.id)}/actions`, + { + method: "POST", + headers: { "If-Match": change.etag }, + body: JSON.stringify({ + action, + base_revision: change.resource_revision, + comment: comment?.trim() || null, + evidence: [] + }) + } + ); +} diff --git a/webui/src/features/FunctionAssignmentChangesPanel.tsx b/webui/src/features/FunctionAssignmentChangesPanel.tsx new file mode 100644 index 0000000..91eed58 --- /dev/null +++ b/webui/src/features/FunctionAssignmentChangesPanel.tsx @@ -0,0 +1,372 @@ +import { useCallback, useEffect, useMemo, useRef, useState, type FormEvent } from "react"; +import { Check, Eye, Plus, RotateCcw, Undo2, X } from "lucide-react"; +import { + AdminIconButton, + ApiError, + Button, + Card, + DataGrid, + Dialog, + DismissibleAlert, + FormField, + SegmentedControl, + StatusBadge, + TableActionGroup, + hasScope, + type ApiSettings, + type AuthInfo, + type DataGridColumn +} from "@govoplan/core-webui"; +import { + actOnFunctionAssignmentChange, + createFunctionAssignmentChange, + getFunctionAssignmentChange, + getFunctionAssignmentChanges, + type FunctionAssignmentChange, + type FunctionAssignmentChangeAction, + type FunctionAssignmentChangeKind, + type IdentityOption, + type OrganizationModel +} from "../api/idm"; + +type Props = { + settings: ApiSettings; + auth: AuthInfo; + model: OrganizationModel; + identities: IdentityOption[]; +}; + +type Draft = { + kind: FunctionAssignmentChangeKind; + functionId: string; + identityId: string; + accountId: string; + justification: string; + evidence: string; + validFrom: string; + validUntil: string; +}; + +function emptyDraft(auth: AuthInfo, kind: FunctionAssignmentChangeKind): Draft { + return { + kind, + functionId: "", + identityId: kind === "request" ? auth.principal?.identity_id ?? "" : "", + accountId: kind === "request" ? auth.principal?.account_id ?? "" : "", + justification: "", + evidence: "", + validFrom: "", + validUntil: "" + }; +} + +function errorMessage(error: unknown): string { + if (error instanceof ApiError) { + try { + const body = JSON.parse(error.body) as { detail?: string | { message?: string } }; + if (typeof body.detail === "string") return body.detail; + if (body.detail?.message) return body.detail.message; + } catch { + // Use the transport message. + } + return error.message; + } + return error instanceof Error ? error.message : String(error); +} + +function dateTimeValue(value: string): string | null { + return value ? new Date(value).toISOString() : null; +} + +function statusTone(state: string): string { + if (state === "applied") return "success"; + if (["rejected", "expired", "withdrawn", "cancelled"].includes(state)) return "inactive"; + if (["blocked", "failed_manual_review"].includes(state)) return "danger"; + return "warning"; +} + +function actionLabel(action: FunctionAssignmentChangeAction): string { + return { + approve: "Approve", + reject: "Reject", + accept: "Accept", + request_changes: "Request changes", + respond: "Respond", + withdraw: "Withdraw", + recover: "Recheck" + }[action]; +} + +function actionIcon(action: FunctionAssignmentChangeAction): JSX.Element { + const props = { size: 16, "aria-hidden": true as const }; + if (action === "approve" || action === "accept") return ; + if (action === "reject") return ; + if (action === "request_changes") return ; + if (action === "recover") return ; + return ; +} + +export default function FunctionAssignmentChangesPanel({ settings, auth, model, identities }: Props) { + const canRequest = hasScope(auth, "idm:function_request:create"); + const canGrant = hasScope(auth, "idm:function_grant:create") || hasScope(auth, "idm:organization_assignment:write"); + const visible = canRequest || canGrant || hasScope(auth, "idm:function_change:read") || hasScope(auth, "idm:function_change:decide") || hasScope(auth, "idm:function_change:admin"); + const initialKind: FunctionAssignmentChangeKind = canRequest ? "request" : "grant"; + const [changes, setChanges] = useState([]); + const [draft, setDraft] = useState(() => emptyDraft(auth, initialKind)); + const [selected, setSelected] = useState(null); + const [createOpen, setCreateOpen] = useState(false); + const [detailOpen, setDetailOpen] = useState(false); + const [comment, setComment] = useState(""); + const [loading, setLoading] = useState(false); + const [busy, setBusy] = useState(false); + const [error, setError] = useState(""); + const initialChangeId = useMemo(() => { + if (typeof window === "undefined") return ""; + return new URLSearchParams(window.location.search).get("change") ?? ""; + }, []); + const initialChangeOpened = useRef(false); + const functionById = useMemo(() => new Map(model.functions.map((item) => [item.id, item])), [model.functions]); + const identityById = useMemo(() => new Map(identities.map((item) => [item.id, item])), [identities]); + const selectedIdentity = identityById.get(draft.identityId); + + const load = useCallback(async () => { + if (!visible) return; + setLoading(true); + setError(""); + try { + const response = await getFunctionAssignmentChanges(settings); + setChanges(response.changes); + if (initialChangeId && !initialChangeOpened.current) { + initialChangeOpened.current = true; + const detail = await getFunctionAssignmentChange(settings, initialChangeId); + setSelected(detail); + setDetailOpen(true); + } + } catch (caught) { + setError(errorMessage(caught)); + } finally { + setLoading(false); + } + }, [initialChangeId, settings, visible]); + + useEffect(() => { + void load(); + }, [load]); + + if (!visible) return null; + + function setKind(kind: FunctionAssignmentChangeKind) { + setDraft(emptyDraft(auth, kind)); + } + + function setIdentity(identityId: string) { + const identity = identityById.get(identityId); + setDraft((current) => ({ + ...current, + identityId, + accountId: identity?.primary_account_id ?? identity?.account_ids[0] ?? "" + })); + } + + async function openDetail(item: FunctionAssignmentChange) { + setBusy(true); + setError(""); + try { + const detail = await getFunctionAssignmentChange(settings, item.id); + setSelected(detail); + setComment(""); + setDetailOpen(true); + } catch (caught) { + setError(errorMessage(caught)); + } finally { + setBusy(false); + } + } + + async function submit(event: FormEvent) { + event.preventDefault(); + setBusy(true); + setError(""); + try { + const created = await createFunctionAssignmentChange(settings, { + kind: draft.kind, + function_id: draft.functionId, + candidate_identity_id: draft.identityId, + candidate_account_id: draft.accountId || null, + justification: draft.justification.trim(), + evidence: draft.evidence.split(/\r?\n/).map((item) => item.trim()).filter(Boolean), + requested_valid_from: dateTimeValue(draft.validFrom), + requested_valid_until: dateTimeValue(draft.validUntil), + idempotency_key: crypto.randomUUID(), + metadata: {} + }); + setCreateOpen(false); + setDraft(emptyDraft(auth, initialKind)); + setSelected(created); + setDetailOpen(true); + await load(); + } catch (caught) { + setError(errorMessage(caught)); + } finally { + setBusy(false); + } + } + + async function performAction(action: FunctionAssignmentChangeAction) { + if (!selected) return; + setBusy(true); + setError(""); + try { + const updated = await actOnFunctionAssignmentChange(settings, selected, action, comment); + setSelected(updated); + setComment(""); + await load(); + } catch (caught) { + setError(errorMessage(caught)); + } finally { + setBusy(false); + } + } + + const columns: DataGridColumn[] = [ + { id: "kind", header: "Kind", width: 110, sortable: true, filterable: true, value: (row) => row.kind }, + { + id: "function", + header: "Function", + minWidth: 220, + sortable: true, + filterable: true, + value: (row) => functionById.get(row.function_id)?.name ?? row.function_id, + render: (row) => functionById.get(row.function_id)?.name ?? row.function_id + }, + { + id: "candidate", + header: "Candidate", + minWidth: 220, + sortable: true, + filterable: true, + value: (row) => identityById.get(row.candidate_identity_id)?.display_name ?? row.candidate_identity_id, + render: (row) => identityById.get(row.candidate_identity_id)?.display_name ?? row.candidate_identity_id + }, + { id: "state", header: "State", width: 170, sortable: true, filterable: true, value: (row) => row.state, render: (row) => }, + { id: "progress", header: "Decisions", width: 140, value: (row) => `${row.completed_steps.length}/${row.required_steps.length}`, render: (row) => `${row.completed_steps.length} / ${row.required_steps.length}` }, + { id: "updated", header: "Updated", width: 170, sortable: true, value: (row) => row.updated_at, render: (row) => new Date(row.updated_at).toLocaleString() }, + { id: "actions", header: "Actions", width: 72, sticky: "end", render: (row) =>