897 lines
30 KiB
Python
897 lines
30 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.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.schemas import RETENTION_POLICY_FIELD_KEYS
|
|
from govoplan_policy.backend.retention import (
|
|
PrivacyPolicyError,
|
|
apply_retention_policy,
|
|
effective_privacy_policy,
|
|
effective_privacy_policy_sources,
|
|
get_privacy_policy_for_scope,
|
|
parent_privacy_policy,
|
|
parent_privacy_policy_sources,
|
|
set_privacy_policy_for_scope,
|
|
simulate_privacy_policy_change,
|
|
)
|
|
from govoplan_policy.backend.definition_policy_service import (
|
|
DefinitionPolicyError,
|
|
definition_policy_response_payload,
|
|
definition_policy_state,
|
|
remove_definition_policy,
|
|
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,
|
|
DefinitionPolicyScopeResponse,
|
|
PrivacyRetentionPolicyExplainResponse,
|
|
PrivacyRetentionPolicyItem,
|
|
PrivacyRetentionPolicyScopeRequest,
|
|
PrivacyRetentionPolicyScopeResponse,
|
|
PrivacyRetentionPolicySimulationResponse,
|
|
RetentionRunRequest,
|
|
RetentionRunResponse,
|
|
ViewPolicyScopeRequest,
|
|
ViewPolicyScopeResponse,
|
|
)
|
|
|
|
router = APIRouter(prefix="/admin", tags=["admin"])
|
|
|
|
|
|
def _require_permission(principal: ApiPrincipal, scope: str) -> None:
|
|
if not has_scope(principal, scope):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {scope}"
|
|
)
|
|
|
|
|
|
def _require_privacy_policy_read(principal: ApiPrincipal, scope_type: str) -> None:
|
|
if scope_type == "system":
|
|
_require_permission(principal, "system:settings:read")
|
|
else:
|
|
_require_permission(principal, "admin:policies:read")
|
|
|
|
|
|
def _require_privacy_policy_write(principal: ApiPrincipal, scope_type: str) -> None:
|
|
if scope_type == "system":
|
|
_require_permission(principal, "system:settings:write")
|
|
else:
|
|
_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},
|
|
)
|
|
|
|
|
|
def _definition_policy_response(
|
|
*,
|
|
module_id: str,
|
|
scope_type: str,
|
|
scope_id: str | None,
|
|
state,
|
|
) -> DefinitionPolicyScopeResponse:
|
|
return DefinitionPolicyScopeResponse(
|
|
module_id=module_id,
|
|
scope_type=scope_type,
|
|
scope_id=scope_id,
|
|
**definition_policy_response_payload(state),
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/definition-policies/{module_id}/{scope_type}",
|
|
response_model=DefinitionPolicyScopeResponse,
|
|
)
|
|
def read_definition_policy(
|
|
module_id: str,
|
|
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 = definition_policy_state(
|
|
session,
|
|
module_id=module_id,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
)
|
|
return _definition_policy_response(
|
|
module_id=module_id,
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
state=state,
|
|
)
|
|
except (DefinitionPolicyError, PolicyOverrideError) as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
|
detail=str(exc),
|
|
) from exc
|
|
|
|
|
|
@router.put(
|
|
"/definition-policies/{module_id}/{scope_type}",
|
|
response_model=DefinitionPolicyScopeResponse,
|
|
)
|
|
def write_definition_policy(
|
|
module_id: str,
|
|
scope_type: str,
|
|
payload: DefinitionPolicyScopeRequest,
|
|
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 = definition_policy_state(
|
|
session,
|
|
module_id=module_id,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
)
|
|
if clean_scope == "system":
|
|
approval = ensure_configuration_change_allowed(
|
|
session,
|
|
key="definition_policy",
|
|
value=policy_value,
|
|
actor_user_id=principal.user.id,
|
|
actor_scopes=tuple(principal.scopes),
|
|
change_request_id=payload.change_request_id,
|
|
target={"module_id": module_id, "scope_type": clean_scope},
|
|
)
|
|
else:
|
|
approval = None
|
|
state = save_definition_policy(
|
|
session,
|
|
module_id=module_id,
|
|
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="definition_policy",
|
|
before_value=definition_policy_response_payload(before)["policy"],
|
|
after_value=policy_value,
|
|
actor_user_id=principal.user.id,
|
|
approval=approval,
|
|
target={"module_id": module_id, "scope_type": clean_scope},
|
|
audit_event="definition_policy.updated",
|
|
)
|
|
audit_from_principal(
|
|
session,
|
|
principal,
|
|
action="definition_policy.updated",
|
|
scope="system" if clean_scope == "system" else "tenant",
|
|
object_type="definition_policy",
|
|
object_id=f"{module_id}:{clean_scope}:{scope_id or ''}",
|
|
details={
|
|
"module_id": module_id,
|
|
"scope_type": clean_scope,
|
|
"scope_id": scope_id,
|
|
"fields": sorted(policy_value),
|
|
},
|
|
)
|
|
session.commit()
|
|
return _definition_policy_response(
|
|
module_id=module_id,
|
|
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 (DefinitionPolicyError, PolicyOverrideError) as exc:
|
|
session.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
|
detail=str(exc),
|
|
) from exc
|
|
|
|
|
|
@router.delete(
|
|
"/definition-policies/{module_id}/{scope_type}",
|
|
response_model=DefinitionPolicyScopeResponse,
|
|
)
|
|
def delete_definition_policy_route(
|
|
module_id: str,
|
|
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 = definition_policy_state(
|
|
session,
|
|
module_id=module_id,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
)
|
|
if clean_scope == "system":
|
|
approval = ensure_configuration_change_allowed(
|
|
session,
|
|
key="definition_policy",
|
|
value={},
|
|
actor_user_id=principal.user.id,
|
|
actor_scopes=tuple(principal.scopes),
|
|
change_request_id=change_request_id,
|
|
target={"module_id": module_id, "scope_type": clean_scope},
|
|
)
|
|
else:
|
|
approval = None
|
|
removed = remove_definition_policy(
|
|
session,
|
|
module_id=module_id,
|
|
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="definition_policy",
|
|
before_value=definition_policy_response_payload(before)["policy"],
|
|
after_value={},
|
|
actor_user_id=principal.user.id,
|
|
approval=approval,
|
|
target={"module_id": module_id, "scope_type": clean_scope},
|
|
audit_event="definition_policy.removed",
|
|
)
|
|
audit_from_principal(
|
|
session,
|
|
principal,
|
|
action="definition_policy.removed",
|
|
scope="system" if clean_scope == "system" else "tenant",
|
|
object_type="definition_policy",
|
|
object_id=f"{module_id}:{clean_scope}:{scope_id or ''}",
|
|
details={
|
|
"module_id": module_id,
|
|
"scope_type": clean_scope,
|
|
"scope_id": scope_id,
|
|
},
|
|
)
|
|
session.commit()
|
|
state = definition_policy_state(
|
|
session,
|
|
module_id=module_id,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
)
|
|
return _definition_policy_response(
|
|
module_id=module_id,
|
|
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 (DefinitionPolicyError, PolicyOverrideError) as exc:
|
|
session.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
|
detail=str(exc),
|
|
) 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,
|
|
)
|
|
def read_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:
|
|
policy = get_privacy_policy_for_scope(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
)
|
|
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,
|
|
)
|
|
return PrivacyRetentionPolicyScopeResponse(
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
policy=policy,
|
|
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_privacy_policy_sources_for_response(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
),
|
|
parent_policy_sources=_parent_privacy_policy_sources_for_response(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
),
|
|
)
|
|
except PrivacyPolicyError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
|
|
) from exc
|
|
|
|
|
|
@router.put(
|
|
"/privacy-retention/policies/{scope_type}",
|
|
response_model=PrivacyRetentionPolicyScopeResponse,
|
|
)
|
|
def write_privacy_retention_policy(
|
|
scope_type: str,
|
|
payload: PrivacyRetentionPolicyScopeRequest,
|
|
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)
|
|
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=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,
|
|
)
|
|
parent = _parent_privacy_policy_for_response(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
)
|
|
return PrivacyRetentionPolicyScopeResponse(
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
policy=policy,
|
|
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_privacy_policy_sources_for_response(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
),
|
|
parent_policy_sources=_parent_privacy_policy_sources_for_response(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
),
|
|
)
|
|
except PrivacyPolicyError 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}/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
|
|
|
|
|
|
@router.post(
|
|
"/privacy-retention/policies/{scope_type}/simulate",
|
|
response_model=PrivacyRetentionPolicySimulationResponse,
|
|
)
|
|
def simulate_privacy_retention_policy(
|
|
scope_type: str,
|
|
payload: PrivacyRetentionPolicyScopeRequest,
|
|
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)
|
|
try:
|
|
return PrivacyRetentionPolicySimulationResponse(
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
simulation=simulate_privacy_policy_change(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=clean_scope,
|
|
scope_id=scope_id,
|
|
policy=payload.policy.model_dump(mode="json", exclude_none=True),
|
|
),
|
|
)
|
|
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
|
|
):
|
|
if scope_type == "system":
|
|
return []
|
|
return parent_privacy_policy_sources(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
scope_type=scope_type,
|
|
scope_id=scope_id or (tenant_id if scope_type == "tenant" else None),
|
|
)
|
|
|
|
|
|
def _effective_privacy_policy_sources_for_response(
|
|
session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None
|
|
):
|
|
if scope_type == "system":
|
|
return effective_privacy_policy_sources(session)
|
|
if scope_type == "tenant":
|
|
return effective_privacy_policy_sources(
|
|
session, tenant_id=scope_id or tenant_id
|
|
)
|
|
if scope_type == "campaign" and scope_id:
|
|
return effective_privacy_policy_sources(session, campaign_id=scope_id)
|
|
if scope_type == "user" and scope_id:
|
|
return effective_privacy_policy_sources(
|
|
session, tenant_id=tenant_id, owner_user_id=scope_id
|
|
)
|
|
if scope_type == "group" and scope_id:
|
|
return effective_privacy_policy_sources(
|
|
session, tenant_id=tenant_id, owner_group_id=scope_id
|
|
)
|
|
return effective_privacy_policy_sources(session, tenant_id=tenant_id)
|
|
|
|
|
|
def _parent_privacy_policy_for_response(
|
|
session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None
|
|
):
|
|
if scope_type == "system":
|
|
return None
|
|
return parent_privacy_policy(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
scope_type=scope_type,
|
|
scope_id=scope_id or (tenant_id if scope_type == "tenant" else None),
|
|
)
|
|
|
|
|
|
def _effective_privacy_policy_for_response(
|
|
session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None
|
|
):
|
|
if scope_type == "system":
|
|
return effective_privacy_policy(session)
|
|
if scope_type == "tenant":
|
|
return effective_privacy_policy(session, tenant_id=scope_id or tenant_id)
|
|
if scope_type == "campaign" and scope_id:
|
|
return effective_privacy_policy(session, campaign_id=scope_id)
|
|
if scope_type == "user" and scope_id:
|
|
return effective_privacy_policy(
|
|
session, tenant_id=tenant_id, owner_user_id=scope_id
|
|
)
|
|
if scope_type == "group" and scope_id:
|
|
return effective_privacy_policy(
|
|
session, tenant_id=tenant_id, owner_group_id=scope_id
|
|
)
|
|
return effective_privacy_policy(session, tenant_id=tenant_id)
|
|
|
|
|
|
@router.post("/system/retention/run", response_model=RetentionRunResponse)
|
|
def run_retention_policy(
|
|
payload: RetentionRunRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_scope("system:settings:write")),
|
|
):
|
|
result = apply_retention_policy(session, dry_run=payload.dry_run)
|
|
audit_from_principal(
|
|
session,
|
|
principal,
|
|
action="retention_policy.run",
|
|
scope="system",
|
|
object_type="retention_policy",
|
|
object_id="global",
|
|
details={"dry_run": payload.dry_run, "counts": result.get("counts")},
|
|
)
|
|
session.commit()
|
|
return RetentionRunResponse(result=result)
|