diff --git a/src/govoplan_policy/backend/api/v1/routes.py b/src/govoplan_policy/backend/api/v1/routes.py index e2a7c1a..7ed7d2f 100644 --- a/src/govoplan_policy/backend/api/v1/routes.py +++ b/src/govoplan_policy/backend/api/v1/routes.py @@ -34,6 +34,13 @@ from govoplan_policy.backend.definition_policy_service import ( save_definition_policy, ) from govoplan_policy.backend.policy_overrides import PolicyOverrideError +from govoplan_policy.backend.view_policy_service import ( + ViewPolicyError, + remove_view_policy, + save_view_policy, + view_policy_response_payload, + view_policy_state, +) from .schemas import ( DefinitionPolicyScopeRequest, @@ -45,6 +52,8 @@ from .schemas import ( PrivacyRetentionPolicySimulationResponse, RetentionRunRequest, RetentionRunResponse, + ViewPolicyScopeRequest, + ViewPolicyScopeResponse, ) router = APIRouter(prefix="/admin", tags=["admin"]) @@ -305,6 +314,214 @@ def delete_definition_policy_route( ) from exc +def _view_policy_response( + *, + scope_type: str, + scope_id: str | None, + state, +) -> ViewPolicyScopeResponse: + return ViewPolicyScopeResponse( + scope_type=scope_type, + scope_id=scope_id, + **view_policy_response_payload(state), + ) + + +@router.get( + "/view-policies/{scope_type}", + response_model=ViewPolicyScopeResponse, +) +def read_view_policy( + scope_type: str, + scope_id: str | None = Query(default=None), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +): + clean_scope = scope_type.strip().casefold() + _require_privacy_policy_read(principal, clean_scope) + try: + state = view_policy_state( + session, + tenant_id=principal.tenant_id, + scope_type=clean_scope, + scope_id=scope_id, + ) + return _view_policy_response( + scope_type=clean_scope, + scope_id=scope_id, + state=state, + ) + except (ViewPolicyError, PolicyOverrideError) as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=str(exc), + ) from exc + + +@router.put( + "/view-policies/{scope_type}", + response_model=ViewPolicyScopeResponse, +) +def write_view_policy( + scope_type: str, + payload: ViewPolicyScopeRequest, + scope_id: str | None = Query(default=None), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +): + clean_scope = scope_type.strip().casefold() + _require_privacy_policy_write(principal, clean_scope) + policy_value = payload.policy.model_dump(mode="json", exclude_none=True) + try: + before = view_policy_state( + session, + tenant_id=principal.tenant_id, + scope_type=clean_scope, + scope_id=scope_id, + ) + if clean_scope == "system": + approval = ensure_configuration_change_allowed( + session, + key="view_policy", + value=policy_value, + actor_user_id=principal.user.id, + actor_scopes=tuple(principal.scopes), + change_request_id=payload.change_request_id, + target={"scope_type": clean_scope}, + ) + else: + approval = None + state = save_view_policy( + session, + tenant_id=principal.tenant_id, + scope_type=clean_scope, + scope_id=scope_id, + policy=policy_value, + actor_id=principal.user.id, + ) + if clean_scope == "system": + record_configuration_change_applied( + session, + key="view_policy", + before_value=view_policy_response_payload(before)["policy"], + after_value=policy_value, + actor_user_id=principal.user.id, + approval=approval, + target={"scope_type": clean_scope}, + audit_event="view_policy.updated", + ) + audit_from_principal( + session, + principal, + action="view_policy.updated", + scope="system" if clean_scope == "system" else "tenant", + object_type="view_policy", + object_id=f"{clean_scope}:{scope_id or ''}", + details={ + "scope_type": clean_scope, + "scope_id": scope_id, + "fields": sorted(policy_value), + }, + ) + session.commit() + return _view_policy_response( + scope_type=clean_scope, + scope_id=scope_id, + state=state, + ) + except ConfigurationControlError as exc: + session.rollback() + raise _configuration_control_http_error(exc) from exc + except (ViewPolicyError, PolicyOverrideError) as exc: + session.rollback() + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=str(exc), + ) from exc + + +@router.delete( + "/view-policies/{scope_type}", + response_model=ViewPolicyScopeResponse, +) +def delete_view_policy_route( + scope_type: str, + scope_id: str | None = Query(default=None), + change_request_id: str | None = Query(default=None), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +): + clean_scope = scope_type.strip().casefold() + _require_privacy_policy_write(principal, clean_scope) + try: + before = view_policy_state( + session, + tenant_id=principal.tenant_id, + scope_type=clean_scope, + scope_id=scope_id, + ) + if clean_scope == "system": + approval = ensure_configuration_change_allowed( + session, + key="view_policy", + value={}, + actor_user_id=principal.user.id, + actor_scopes=tuple(principal.scopes), + change_request_id=change_request_id, + target={"scope_type": clean_scope}, + ) + else: + approval = None + removed = remove_view_policy( + session, + tenant_id=principal.tenant_id, + scope_type=clean_scope, + scope_id=scope_id, + ) + if removed: + if clean_scope == "system": + record_configuration_change_applied( + session, + key="view_policy", + before_value=view_policy_response_payload(before)["policy"], + after_value={}, + actor_user_id=principal.user.id, + approval=approval, + target={"scope_type": clean_scope}, + audit_event="view_policy.removed", + ) + audit_from_principal( + session, + principal, + action="view_policy.removed", + scope="system" if clean_scope == "system" else "tenant", + object_type="view_policy", + object_id=f"{clean_scope}:{scope_id or ''}", + details={"scope_type": clean_scope, "scope_id": scope_id}, + ) + session.commit() + state = view_policy_state( + session, + tenant_id=principal.tenant_id, + scope_type=clean_scope, + scope_id=scope_id, + ) + return _view_policy_response( + scope_type=clean_scope, + scope_id=scope_id, + state=state, + ) + except ConfigurationControlError as exc: + session.rollback() + raise _configuration_control_http_error(exc) from exc + except (ViewPolicyError, PolicyOverrideError) as exc: + session.rollback() + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=str(exc), + ) from exc + + @router.get( "/privacy-retention/policies/{scope_type}", response_model=PrivacyRetentionPolicyScopeResponse, diff --git a/src/govoplan_policy/backend/api/v1/schemas.py b/src/govoplan_policy/backend/api/v1/schemas.py index 2ba5e88..bb8627a 100644 --- a/src/govoplan_policy/backend/api/v1/schemas.py +++ b/src/govoplan_policy/backend/api/v1/schemas.py @@ -77,6 +77,38 @@ class DefinitionPolicyScopeResponse(BaseModel): diagnostics: list[dict[str, str]] = Field(default_factory=list) +class ViewPolicyItem(BaseModel): + model_config = ConfigDict(extra="forbid") + + allow_view: bool | None = None + allow_select: bool | None = None + allow_assign: bool | None = None + allow_edit: bool | None = None + allow_derive: bool | None = None + allow_workflow_activate: bool | None = None + allowed_view_ids: list[str] | None = Field(default=None, max_length=1000) + visible_surface_ids: list[str] | None = Field(default=None, max_length=2000) + + +class ViewPolicyScopeRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + policy: ViewPolicyItem = Field(default_factory=ViewPolicyItem) + change_request_id: str | None = None + + +class ViewPolicyScopeResponse(BaseModel): + scope_type: Literal["system", "tenant", "group", "user"] + scope_id: str | None = None + id: str | None = None + revision: int | None = None + policy: dict[str, Any] = Field(default_factory=dict) + effective_policy: dict[str, Any] = Field(default_factory=dict) + parent_policy: dict[str, Any] = Field(default_factory=dict) + source_path: list[PolicySourceStepItem] = Field(default_factory=list) + diagnostics: list[dict[str, Any]] = Field(default_factory=list) + + class PrivacyRetentionPolicyExplainResponse(BaseModel): scope_type: Literal["system", "tenant", "user", "group", "campaign"] scope_id: str | None = None diff --git a/src/govoplan_policy/backend/manifest.py b/src/govoplan_policy/backend/manifest.py index c1906ca..cbe86cb 100644 --- a/src/govoplan_policy/backend/manifest.py +++ b/src/govoplan_policy/backend/manifest.py @@ -14,6 +14,7 @@ from govoplan_core.core.policy import ( CAPABILITY_POLICY_DEFINITION_GOVERNANCE, CAPABILITY_POLICY_PRIVACY_RETENTION, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, + CAPABILITY_POLICY_VIEW_GOVERNANCE, ) from govoplan_core.core.modules import ( FrontendModule, @@ -59,6 +60,15 @@ def _definition_governance_policy(context: ModuleContext) -> object: return DefinitionGovernancePolicyProvider() +def _view_governance_policy(context: ModuleContext) -> object: + del context + from govoplan_policy.backend.view_governance import ( + ViewGovernancePolicyProvider, + ) + + return ViewGovernancePolicyProvider() + + manifest = ModuleManifest( id="policy", name="Policy", @@ -72,6 +82,10 @@ manifest = ModuleManifest( name="policy.definition_governance", version="0.1.0", ), + ModuleInterfaceProvider( + name="policy.view_governance", + version="0.1.0", + ), ), route_factory=_route_factory, migration_spec=MigrationSpec( @@ -130,6 +144,7 @@ manifest = ModuleManifest( ), capability_factories={ CAPABILITY_POLICY_DEFINITION_GOVERNANCE: _definition_governance_policy, + CAPABILITY_POLICY_VIEW_GOVERNANCE: _view_governance_policy, CAPABILITY_POLICY_PRIVACY_RETENTION: _privacy_retention_service, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY: _scheduling_participant_privacy_policy, }, diff --git a/src/govoplan_policy/backend/view_governance.py b/src/govoplan_policy/backend/view_governance.py new file mode 100644 index 0000000..b6cd832 --- /dev/null +++ b/src/govoplan_policy/backend/view_governance.py @@ -0,0 +1,319 @@ +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +from sqlalchemy.orm import Session + +from govoplan_core.core.policy import ( + PolicySourceStep, + ViewGovernanceDecision, + ViewGovernanceRequest, +) +from govoplan_core.db.session import get_database +from govoplan_policy.backend.policy_overrides import resolution_policy_overrides + + +VIEW_POLICY_BOOLEAN_FIELDS = ( + "allow_view", + "allow_select", + "allow_assign", + "allow_edit", + "allow_derive", + "allow_workflow_activate", +) +VIEW_POLICY_SET_FIELDS = ("allowed_view_ids", "visible_surface_ids") +VIEW_POLICY_FIELDS = (*VIEW_POLICY_BOOLEAN_FIELDS, *VIEW_POLICY_SET_FIELDS) + + +@dataclass(frozen=True, slots=True) +class ViewPolicyResolution: + limits: Mapping[str, bool] + allowed_view_ids: frozenset[str] | None = None + visible_surface_ids: frozenset[str] | None = None + source_path: tuple[PolicySourceStep, ...] = () + diagnostics: tuple[Mapping[str, Any], ...] = () + + +class ViewGovernancePolicyProvider: + def resolve_view_action( + self, + session: object | None = None, + *, + request: ViewGovernanceRequest, + ) -> ViewGovernanceDecision: + resolution = _explicit_view_policy_resolution(session, request) + action_field = f"allow_{request.action}" + allowed = resolution.limits["allow_view"] and resolution.limits.get( + action_field, + False, + ) + reason = None + if not allowed: + reason = f"View action '{request.action}' is disabled by Policy." + + allowed_view_ids = _bounded_candidates( + request.candidate_view_ids, + resolution.allowed_view_ids, + ) + visible_surface_ids = _bounded_candidates( + request.candidate_surface_ids, + resolution.visible_surface_ids, + ) + unavailable_view_ids = sorted( + set(request.candidate_view_ids) - set(allowed_view_ids or ()) + if resolution.allowed_view_ids is not None + else () + ) + hidden_surface_ids = sorted( + set(request.candidate_surface_ids) - set(visible_surface_ids or ()) + if resolution.visible_surface_ids is not None + else () + ) + if ( + request.view_id is not None + and resolution.allowed_view_ids is not None + and request.view_id not in resolution.allowed_view_ids + ): + allowed = False + reason = "The requested View is outside the effective Policy ceiling." + + requested_outside_ceiling = sorted( + set(request.requested_surface_ids) - set(visible_surface_ids or ()) + if resolution.visible_surface_ids is not None + else () + ) + if requested_outside_ceiling and request.action in {"edit", "derive"}: + allowed = False + reason = ( + "The requested View surfaces are outside the effective Policy ceiling." + ) + diagnostics = list(resolution.diagnostics) + if requested_outside_ceiling: + diagnostics.append( + { + "code": "view_policy.workflow_surface_bounded", + "severity": "warning", + "surface_ids": requested_outside_ceiling, + "message": ( + "Workflow surfaces outside the effective Policy ceiling " + "remain hidden." + ), + } + ) + return ViewGovernanceDecision( + allowed=allowed, + reason=reason, + allowed_view_ids=allowed_view_ids, + visible_surface_ids=visible_surface_ids, + source_path=resolution.source_path, + requirements=(() if allowed else (f"policy.view.{request.action}",)), + diagnostics=tuple(diagnostics), + details={ + "action": request.action, + "target_scope": request.target_scope.path, + "unavailable_view_ids": unavailable_view_ids, + "hidden_surface_ids": hidden_surface_ids, + "requested_surfaces_outside_ceiling": requested_outside_ceiling, + "view_provenance": _excluded_candidate_provenance( + request.candidate_view_ids, + resolution.source_path, + field="allowed_view_ids", + ), + "surface_provenance": _excluded_candidate_provenance( + request.candidate_surface_ids, + resolution.source_path, + field="visible_surface_ids", + ), + }, + ) + + +def _bounded_candidates( + candidates: tuple[str, ...], + ceiling: frozenset[str] | None, +) -> frozenset[str] | None: + if ceiling is None: + return None + return frozenset(candidates).intersection(ceiling) + + +def _excluded_candidate_provenance( + candidates: tuple[str, ...], + source_path: tuple[PolicySourceStep, ...], + *, + field: str, +) -> list[dict[str, Any]]: + result: list[dict[str, Any]] = [] + for candidate in candidates: + sources = [ + step.path + for step in source_path + if isinstance(step.policy.get(field), list) + and candidate not in step.policy[field] + ] + if sources: + result.append({"id": candidate, "sources": sources}) + return result + + +def _explicit_view_policy_resolution( + session: object | None, + request: ViewGovernanceRequest, +) -> ViewPolicyResolution: + if not isinstance(session, Session): + try: + with get_database().SessionLocal() as policy_session: + return _explicit_view_policy_resolution(policy_session, request) + except RuntimeError: + return ViewPolicyResolution( + limits={field: True for field in VIEW_POLICY_BOOLEAN_FIELDS} + ) + cache_key = ( + "view", + request.tenant_id, + request.target_scope.path, + tuple(sorted(_target_group_ids(request))), + tuple(sorted(_target_user_ids(request))), + ) + cache = session.info.setdefault("govoplan_policy_override_resolution", {}) + if isinstance(cache, dict) and cache_key in cache: + cached = cache[cache_key] + if isinstance(cached, ViewPolicyResolution): + return cached + rows = resolution_policy_overrides( + session, + policy_family="view", + target_keys=("*",), + tenant_id=request.tenant_id, + group_ids=_target_group_ids(request), + user_ids=_target_user_ids(request), + ) + result = resolve_view_policy_rows(rows) + if isinstance(cache, dict): + cache[cache_key] = result + return result + + +def _target_group_ids(request: ViewGovernanceRequest) -> tuple[str, ...]: + target = request.target_scope + if target.scope_type == "group" and target.scope_id: + return (target.scope_id,) + if target.scope_type == "user" and target.scope_id in { + request.actor.account_id, + request.actor.membership_id, + }: + return tuple(sorted(request.actor.group_ids)) + return () + + +def _target_user_ids(request: ViewGovernanceRequest) -> tuple[str, ...]: + target = request.target_scope + if target.scope_type == "user" and target.scope_id: + return (target.scope_id,) + return () + + +def resolve_view_policy_rows(rows: object) -> ViewPolicyResolution: + limits = {field: True for field in VIEW_POLICY_BOOLEAN_FIELDS} + allowed_view_ids: frozenset[str] | None = None + visible_surface_ids: frozenset[str] | None = None + source_path: list[PolicySourceStep] = [] + diagnostics: list[Mapping[str, Any]] = [] + for row in rows if isinstance(rows, (list, tuple)) else (): + policy, malformed = validate_view_policy(row.policy) + if malformed: + limits.update({field: False for field in VIEW_POLICY_BOOLEAN_FIELDS}) + allowed_view_ids = frozenset() + visible_surface_ids = frozenset() + applied_fields = VIEW_POLICY_FIELDS + source_policy: Mapping[str, Any] = { + "configuration_status": "invalid_fail_closed", + "target_key": row.target_key, + } + diagnostics.append( + { + "code": "view_policy.invalid", + "severity": "error", + "scope": row.scope_key, + "message": "A malformed View policy record was ignored safely.", + } + ) + else: + for field in VIEW_POLICY_BOOLEAN_FIELDS: + value = policy.get(field) + if isinstance(value, bool): + limits[field] = limits[field] and value + allowed = policy.get("allowed_view_ids") + if isinstance(allowed, tuple): + candidate = frozenset(allowed) + allowed_view_ids = ( + candidate + if allowed_view_ids is None + else allowed_view_ids.intersection(candidate) + ) + visible = policy.get("visible_surface_ids") + if isinstance(visible, tuple): + candidate = frozenset(visible) + visible_surface_ids = ( + candidate + if visible_surface_ids is None + else visible_surface_ids.intersection(candidate) + ) + applied_fields = tuple(sorted(policy)) + source_policy = { + key: list(value) if isinstance(value, tuple) else value + for key, value in policy.items() + } + source_path.append( + PolicySourceStep( + scope_type=row.scope_type, # type: ignore[arg-type] + scope_id=row.scope_id, + label=f"{row.scope_type.capitalize()} View policy", + applied_fields=tuple(applied_fields), + policy=source_policy, + ) + ) + return ViewPolicyResolution( + limits=limits, + allowed_view_ids=allowed_view_ids, + visible_surface_ids=visible_surface_ids, + source_path=tuple(source_path), + diagnostics=tuple(diagnostics), + ) + + +def validate_view_policy( + value: object, +) -> tuple[dict[str, bool | tuple[str, ...]], bool]: + if not isinstance(value, Mapping): + return {}, True + if any(str(key) not in VIEW_POLICY_FIELDS for key in value): + return {}, True + result: dict[str, bool | tuple[str, ...]] = {} + for raw_key, raw_value in value.items(): + key = str(raw_key) + if key in VIEW_POLICY_BOOLEAN_FIELDS: + if not isinstance(raw_value, bool): + return {}, True + result[key] = raw_value + continue + if not isinstance(raw_value, (list, tuple)): + return {}, True + values = tuple(dict.fromkeys(str(item).strip() for item in raw_value)) + if any(not item or len(item) > 160 for item in values): + return {}, True + result[key] = values + return result, False + + +__all__ = [ + "VIEW_POLICY_BOOLEAN_FIELDS", + "VIEW_POLICY_FIELDS", + "VIEW_POLICY_SET_FIELDS", + "ViewGovernancePolicyProvider", + "ViewPolicyResolution", + "resolve_view_policy_rows", + "validate_view_policy", +] diff --git a/src/govoplan_policy/backend/view_policy_service.py b/src/govoplan_policy/backend/view_policy_service.py new file mode 100644 index 0000000..e652d90 --- /dev/null +++ b/src/govoplan_policy/backend/view_policy_service.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +from sqlalchemy.orm import Session + +from govoplan_policy.backend.db.models import PolicyOverride +from govoplan_policy.backend.policy_overrides import ( + delete_policy_override, + get_policy_override, + normalize_policy_scope, + resolution_policy_overrides, + set_policy_override, +) +from govoplan_policy.backend.view_governance import ( + VIEW_POLICY_BOOLEAN_FIELDS, + ViewPolicyResolution, + resolve_view_policy_rows, + validate_view_policy, +) + + +class ViewPolicyError(ValueError): + pass + + +@dataclass(frozen=True, slots=True) +class ViewPolicyState: + row: PolicyOverride | None + local_policy: Mapping[str, bool | tuple[str, ...]] + effective: ViewPolicyResolution + parent: ViewPolicyResolution + + +def view_policy_state( + session: Session, + *, + tenant_id: str, + scope_type: str, + scope_id: str | None = None, +) -> ViewPolicyState: + _, clean_scope_id, _ = normalize_policy_scope( + tenant_id=tenant_id, + scope_type=scope_type, + scope_id=scope_id, + ) + row = get_policy_override( + session, + policy_family="view", + target_key="*", + tenant_id=tenant_id, + scope_type=scope_type, + scope_id=clean_scope_id, + ) + local_policy, malformed = validate_view_policy( + row.policy if row is not None else {} + ) + if malformed: + local_policy = {} + rows = _rows_for_scope( + session, + tenant_id=tenant_id, + scope_type=scope_type, + scope_id=clean_scope_id, + ) + rank = _scope_rank(scope_type) + return ViewPolicyState( + row=row, + local_policy=local_policy, + effective=resolve_view_policy_rows(rows), + parent=resolve_view_policy_rows( + tuple(item for item in rows if _scope_rank(item.scope_type) < rank) + ), + ) + + +def save_view_policy( + session: Session, + *, + tenant_id: str, + scope_type: str, + scope_id: str | None, + policy: object, + actor_id: str | None, +) -> ViewPolicyState: + clean_policy, malformed = validate_view_policy(policy) + if malformed: + raise ViewPolicyError("View policy fields have invalid names or values") + before = view_policy_state( + session, + tenant_id=tenant_id, + scope_type=scope_type, + scope_id=scope_id, + ) + broadened = sorted( + field + for field in VIEW_POLICY_BOOLEAN_FIELDS + if clean_policy.get(field) is True and before.parent.limits[field] is False + ) + for field, parent_values in ( + ("allowed_view_ids", before.parent.allowed_view_ids), + ("visible_surface_ids", before.parent.visible_surface_ids), + ): + local_values = clean_policy.get(field) + if ( + isinstance(local_values, tuple) + and parent_values is not None + and not set(local_values).issubset(parent_values) + ): + broadened.append(field) + if broadened: + raise ViewPolicyError( + "Lower-scope View policy cannot broaden parent restrictions: " + + ", ".join(broadened) + ) + set_policy_override( + session, + policy_family="view", + target_key="*", + tenant_id=tenant_id, + scope_type=scope_type, + scope_id=scope_id, + policy={ + key: list(value) if isinstance(value, tuple) else value + for key, value in clean_policy.items() + }, + actor_id=actor_id, + ) + _clear_resolution_cache(session) + return view_policy_state( + session, + tenant_id=tenant_id, + scope_type=scope_type, + scope_id=scope_id, + ) + + +def remove_view_policy( + session: Session, + *, + tenant_id: str, + scope_type: str, + scope_id: str | None, +) -> bool: + row = get_policy_override( + session, + policy_family="view", + target_key="*", + tenant_id=tenant_id, + scope_type=scope_type, + scope_id=scope_id, + ) + if row is None: + return False + delete_policy_override(session, row) + _clear_resolution_cache(session) + return True + + +def view_policy_response_payload(state: ViewPolicyState) -> dict[str, Any]: + local_policy: Mapping[str, Any] = state.local_policy + if state.row is not None: + _, malformed = validate_view_policy(state.row.policy) + if malformed: + local_policy = {"configuration_status": "invalid_fail_closed"} + return { + "id": state.row.id if state.row is not None else None, + "revision": state.row.revision if state.row is not None else None, + "policy": _json_policy(local_policy), + "effective_policy": _resolution_payload(state.effective), + "parent_policy": _resolution_payload(state.parent), + "source_path": [step.to_dict() for step in state.effective.source_path], + "diagnostics": [dict(item) for item in state.effective.diagnostics], + } + + +def _resolution_payload(resolution: ViewPolicyResolution) -> dict[str, Any]: + return { + **dict(resolution.limits), + "allowed_view_ids": ( + sorted(resolution.allowed_view_ids) + if resolution.allowed_view_ids is not None + else None + ), + "visible_surface_ids": ( + sorted(resolution.visible_surface_ids) + if resolution.visible_surface_ids is not None + else None + ), + } + + +def _json_policy(policy: Mapping[str, Any]) -> dict[str, Any]: + return { + key: list(value) if isinstance(value, tuple) else value + for key, value in policy.items() + } + + +def _rows_for_scope( + session: Session, + *, + tenant_id: str, + scope_type: str, + scope_id: str | None, +) -> tuple[PolicyOverride, ...]: + clean_scope = scope_type.strip().casefold() + rows = resolution_policy_overrides( + session, + policy_family="view", + target_keys=("*",), + tenant_id=tenant_id, + group_ids=(scope_id,) if clean_scope == "group" and scope_id else (), + user_ids=(scope_id,) if clean_scope == "user" and scope_id else (), + ) + maximum_rank = _scope_rank(clean_scope) + return tuple(row for row in rows if _scope_rank(row.scope_type) <= maximum_rank) + + +def _scope_rank(scope_type: str) -> int: + try: + return ("system", "tenant", "group", "user").index( + scope_type.strip().casefold() + ) + except ValueError as exc: + raise ViewPolicyError( + "View policy scope must be system, tenant, group, or user" + ) from exc + + +def _clear_resolution_cache(session: Session) -> None: + session.info.pop("govoplan_policy_override_resolution", None) + + +__all__ = [ + "ViewPolicyError", + "ViewPolicyState", + "remove_view_policy", + "save_view_policy", + "view_policy_response_payload", + "view_policy_state", +] diff --git a/tests/test_policy_module_contract.py b/tests/test_policy_module_contract.py index 30a6dac..8da8913 100644 --- a/tests/test_policy_module_contract.py +++ b/tests/test_policy_module_contract.py @@ -8,6 +8,7 @@ from govoplan_core.core.policy import ( CAPABILITY_POLICY_DEFINITION_GOVERNANCE, CAPABILITY_POLICY_PRIVACY_RETENTION, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, + CAPABILITY_POLICY_VIEW_GOVERNANCE, ) from govoplan_policy.backend.manifest import manifest @@ -38,6 +39,7 @@ class PolicyModuleContractTests(unittest.TestCase): CAPABILITY_POLICY_DEFINITION_GOVERNANCE, CAPABILITY_POLICY_PRIVACY_RETENTION, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, + CAPABILITY_POLICY_VIEW_GOVERNANCE, }, set(manifest.capability_factories), ) diff --git a/tests/test_view_governance.py b/tests/test_view_governance.py new file mode 100644 index 0000000..a08e366 --- /dev/null +++ b/tests/test_view_governance.py @@ -0,0 +1,227 @@ +from __future__ import annotations + +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, sessionmaker + +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.policy import DefinitionScopeRef, ViewGovernanceRequest +from govoplan_policy.backend.db.models import PolicyOverride +from govoplan_policy.backend.view_governance import ViewGovernancePolicyProvider +from govoplan_policy.backend.view_policy_service import ( + ViewPolicyError, + save_view_policy, + view_policy_response_payload, + view_policy_state, +) + + +class ViewGovernanceTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite+pysqlite:///:memory:") + PolicyOverride.__table__.create(self.engine) + self.session: Session = sessionmaker( + bind=self.engine, + expire_on_commit=False, + )() + self.provider = ViewGovernancePolicyProvider() + self.actor = PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id="tenant-1", + group_ids=frozenset({"group-1"}), + ) + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def _save( + self, + scope_type: str, + policy: object, + scope_id: str | None = None, + ): + return save_view_policy( + self.session, + tenant_id="tenant-1", + scope_type=scope_type, + scope_id=scope_id, + policy=policy, + actor_id="account-1", + ) + + def _resolve( + self, + *, + action: str = "view", + view_id: str | None = None, + requested_surface_ids: tuple[str, ...] = (), + ): + return self.provider.resolve_view_action( + self.session, + request=ViewGovernanceRequest( + tenant_id="tenant-1", + action=action, # type: ignore[arg-type] + actor=self.actor, + target_scope=DefinitionScopeRef("user", "account-1"), + view_id=view_id, + candidate_view_ids=("view-1", "view-2", "view-3"), + candidate_surface_ids=("surface.a", "surface.b", "surface.c"), + requested_surface_ids=requested_surface_ids, + ), + ) + + def test_hierarchy_intersects_view_and_surface_ceilings(self) -> None: + self._save( + "system", + { + "allowed_view_ids": ["view-1", "view-2"], + "visible_surface_ids": ["surface.a", "surface.b", "surface.c"], + }, + ) + self._save( + "tenant", + { + "allowed_view_ids": ["view-2"], + "visible_surface_ids": ["surface.a", "surface.b"], + }, + ) + + decision = self._resolve( + action="workflow_activate", + view_id="view-2", + requested_surface_ids=("surface.b", "surface.c"), + ) + + self.assertTrue(decision.allowed) + self.assertEqual(frozenset({"view-2"}), decision.allowed_view_ids) + self.assertEqual( + frozenset({"surface.a", "surface.b"}), + decision.visible_surface_ids, + ) + self.assertEqual( + ["system", "tenant:tenant-1"], + [step.path for step in decision.source_path], + ) + self.assertEqual( + ["surface.c"], + decision.details["requested_surfaces_outside_ceiling"], + ) + + def test_lower_scope_cannot_broaden_boolean_or_set_ceiling(self) -> None: + self._save( + "tenant", + { + "allow_assign": False, + "allowed_view_ids": ["view-1"], + }, + ) + + with self.assertRaisesRegex( + ViewPolicyError, + "allow_assign, allowed_view_ids", + ): + self._save( + "group", + { + "allow_assign": True, + "allowed_view_ids": ["view-1", "view-2"], + }, + "group-1", + ) + + def test_action_and_requested_view_are_bounded_independently(self) -> None: + self._save( + "tenant", + { + "allow_assign": False, + "allowed_view_ids": ["view-1"], + }, + ) + + assignment = self._resolve(action="assign", view_id="view-1") + selection = self._resolve(action="select", view_id="view-2") + + self.assertFalse(assignment.allowed) + self.assertIn("disabled by Policy", assignment.reason or "") + self.assertFalse(selection.allowed) + self.assertIn("outside the effective Policy ceiling", selection.reason or "") + + def test_edit_cannot_store_surfaces_outside_the_policy_ceiling(self) -> None: + self._save( + "tenant", + {"visible_surface_ids": ["surface.a", "surface.b"]}, + ) + + decision = self._resolve( + action="edit", + requested_surface_ids=("surface.a", "surface.c"), + ) + + self.assertFalse(decision.allowed) + self.assertIn("surfaces are outside", decision.reason or "") + self.assertEqual( + [{"id": "surface.c", "sources": ["tenant:tenant-1"]}], + decision.details["surface_provenance"], + ) + + def test_assignment_uses_the_target_group_policy(self) -> None: + self._save( + "group", + {"allow_assign": False}, + "group-2", + ) + + decision = self.provider.resolve_view_action( + self.session, + request=ViewGovernanceRequest( + tenant_id="tenant-1", + action="assign", + actor=self.actor, + target_scope=DefinitionScopeRef("group", "group-2"), + view_id="view-1", + candidate_view_ids=("view-1",), + ), + ) + + self.assertFalse(decision.allowed) + self.assertEqual("group:group-2", decision.source_path[-1].path) + + def test_malformed_policy_fails_closed_without_echoing_record(self) -> None: + self.session.add( + PolicyOverride( + policy_family="view", + target_key="*", + tenant_id="tenant-1", + scope_type="tenant", + scope_id="tenant-1", + scope_key="tenant:tenant-1", + policy={"visible_surface_ids": "secret-value"}, + ) + ) + self.session.flush() + + decision = self._resolve(action="select", view_id="view-1") + payload = view_policy_response_payload( + view_policy_state( + self.session, + tenant_id="tenant-1", + scope_type="tenant", + ) + ) + + self.assertFalse(decision.allowed) + self.assertEqual(frozenset(), decision.allowed_view_ids) + self.assertEqual(frozenset(), decision.visible_surface_ids) + self.assertEqual( + {"configuration_status": "invalid_fail_closed"}, + payload["policy"], + ) + self.assertNotIn("secret-value", repr(decision.to_dict())) + self.assertNotIn("secret-value", repr(payload)) + + +if __name__ == "__main__": + unittest.main()