GovOPlaN/govoplan-policy
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

chore: sync GovOPlaN module split state

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-10 12:51:22 +02:00
parent 3a3676e289
commit b2903e6adf
25 changed files with 218 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
src/govoplan_policy/backend/api/v1/routes.py
View File

@@ -1,13 +1,22 @@
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from govoplan_access.backend.auth.dependencies import ApiPrincipal, get_api_principal, has_scope, require_scope
from govoplan_access.auth import ApiPrincipal, get_api_principal, has_scope, require_scope
from govoplan_core.audit.logging import audit_from_principal
from govoplan_core.core.configuration_control import (
ConfigurationControlError,
ensure_configuration_change_allowed,
record_configuration_change_applied,
)
from govoplan_core.core.policy import PolicyDecision, PolicySourceStep
from govoplan_core.db.session import get_session
from govoplan_core.privacy.retention import (
PrivacyPolicyError,
RETENTION_POLICY_FIELD_KEYS,
apply_retention_policy,
effective_privacy_policy,
effective_privacy_policy_sources,
@@ -18,6 +27,7 @@ from govoplan_core.privacy.retention import (
)
from .schemas import (
PrivacyRetentionPolicyExplainResponse,
PrivacyRetentionPolicyItem,
PrivacyRetentionPolicyScopeRequest,
PrivacyRetentionPolicyScopeResponse,
@@ -47,6 +57,13 @@ def _require_privacy_policy_write(principal: ApiPrincipal, scope_type: str) -> N
_require_permission(principal, "admin:policies:write")
def _configuration_control_http_error(exc: ConfigurationControlError) -> HTTPException:
return HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail={"code": exc.code, "message": str(exc), "plan": exc.plan},
)
@router.get("/privacy-retention/policies/{scope_type}", response_model=PrivacyRetentionPolicyScopeResponse)
def read_privacy_retention_policy(
scope_type: str,
@@ -83,13 +100,51 @@ def write_privacy_retention_policy(
):
clean_scope = scope_type.strip().casefold()
_require_privacy_policy_write(principal, clean_scope)
policy_value = payload.policy.model_dump(mode="json", exclude_none=True)
before_value: dict[str, Any] | None = None
if clean_scope == "system":
before_value = get_privacy_policy_for_scope(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
try:
approval = ensure_configuration_change_allowed(
session,
key="privacy_retention_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, "scope_id": scope_id},
)
except ConfigurationControlError as exc:
raise _configuration_control_http_error(exc) from exc
else:
approval = None
try:
policy = set_privacy_policy_for_scope(
session,
tenant_id=principal.tenant_id,
scope_type=clean_scope,
scope_id=scope_id,
policy=payload.policy.model_dump(mode="json", exclude_none=True),
policy=policy_value,
)
if clean_scope == "system":
record_configuration_change_applied(
session,
key="privacy_retention_policy",
before_value=before_value,
after_value=policy,
actor_user_id=principal.user.id,
approval=approval,
target={"scope_type": clean_scope, "scope_id": scope_id},
audit_event="privacy_retention.policy_updated",
)
audit_from_principal(
session,
principal,
action="privacy_retention.policy_updated",
scope="system" if clean_scope == "system" else "tenant",
object_type="privacy_retention_policy",
object_id=clean_scope if scope_id is None else f"{clean_scope}:{scope_id}",
details={"scope_type": clean_scope, "scope_id": scope_id},
)
session.commit()
effective = _effective_privacy_policy_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
@@ -105,7 +160,52 @@ def write_privacy_retention_policy(
)
except PrivacyPolicyError as exc:
session.rollback()
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
@router.get("/privacy-retention/policies/{scope_type}/explain", response_model=PrivacyRetentionPolicyExplainResponse)
def explain_privacy_retention_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:
effective = _effective_privacy_policy_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
parent = _parent_privacy_policy_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
effective_sources = _effective_privacy_policy_sources_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
parent_sources = _parent_privacy_policy_sources_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
blocked_fields = _blocked_privacy_retention_fields(parent)
decision_sources = parent_sources or effective_sources
decision = PolicyDecision(
allowed=not blocked_fields,
reason="Parent retention policy locks lower-level changes." if blocked_fields else None,
source_path=tuple(PolicySourceStep.from_mapping(source) for source in decision_sources),
requirements=tuple(blocked_fields),
details={"blocked_fields": blocked_fields},
)
return PrivacyRetentionPolicyExplainResponse(
scope_type=clean_scope,
scope_id=scope_id,
decision=decision.to_dict(),
effective_policy=PrivacyRetentionPolicyItem.model_validate(effective.model_dump(mode="json")),
parent_policy=PrivacyRetentionPolicyItem.model_validate(parent.model_dump(mode="json")) if parent else None,
effective_policy_sources=effective_sources,
parent_policy_sources=parent_sources,
blocked_fields=blocked_fields,
)
except PrivacyPolicyError as exc:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
def _blocked_privacy_retention_fields(parent) -> list[str]:
if parent is None:
return []
payload = parent.model_dump(mode="json")
allow_lower_level_limits = payload.get("allow_lower_level_limits") or {}
return [key for key in RETENTION_POLICY_FIELD_KEYS if allow_lower_level_limits.get(key) is False]
def _parent_privacy_policy_sources_for_response(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None):