264 lines
8.8 KiB
Python
264 lines
8.8 KiB
Python
"""Hierarchical privacy and export policy for cross-module reports."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.admin.settings import get_system_settings
|
|
from govoplan_core.core.reporting import (
|
|
ReportingGovernanceDecision,
|
|
ReportingGovernanceRequest,
|
|
)
|
|
from govoplan_core.tenancy.scope import Tenant
|
|
from govoplan_policy.backend.retention import effective_privacy_policy
|
|
|
|
|
|
REPORTING_POLICY_SETTINGS_KEY = "reporting_governance_policy"
|
|
_ALLOWED_KEYS = {
|
|
"allow_exports",
|
|
"allowed_export_formats",
|
|
"allow_high_reidentification_risk",
|
|
"max_retention_days",
|
|
"required_privacy_transforms",
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _Policy:
|
|
allow_exports: bool = True
|
|
allowed_export_formats: tuple[str, ...] = ("json", "csv")
|
|
allow_high_reidentification_risk: bool = False
|
|
max_retention_days: int | None = 30
|
|
required_privacy_transforms: tuple[str, ...] = ()
|
|
|
|
|
|
class ReportingGovernancePolicyProvider:
|
|
def decide_reporting_action(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
request: ReportingGovernanceRequest,
|
|
) -> ReportingGovernanceDecision:
|
|
if not isinstance(session, Session):
|
|
raise TypeError("Reporting governance requires a SQLAlchemy Session")
|
|
try:
|
|
policy, sources = _effective_policy(
|
|
session,
|
|
tenant_id=request.tenant_id,
|
|
)
|
|
retention = effective_privacy_policy(
|
|
session,
|
|
tenant_id=request.tenant_id,
|
|
).stored_report_detail_retention_days
|
|
except (LookupError, TypeError, ValueError) as exc:
|
|
return ReportingGovernanceDecision(
|
|
allowed=False,
|
|
reason="Reporting governance Policy is malformed or unavailable.",
|
|
retention_days=0,
|
|
export_formats=(),
|
|
required_privacy_transforms=request.declared_privacy_transforms,
|
|
provenance={
|
|
"provider": "policy.reporting_governance",
|
|
"version": "1",
|
|
"decision": "fail_closed",
|
|
"error_type": type(exc).__name__,
|
|
},
|
|
)
|
|
del principal
|
|
retention_days = _minimum_optional(
|
|
policy.max_retention_days,
|
|
retention,
|
|
)
|
|
allowed = True
|
|
reason = None
|
|
if (
|
|
request.reidentification_risk == "high"
|
|
and not policy.allow_high_reidentification_risk
|
|
):
|
|
allowed = False
|
|
reason = "Policy blocks reports with high re-identification risk."
|
|
if request.action == "export":
|
|
if not policy.allow_exports:
|
|
allowed = False
|
|
reason = "Policy disables cross-module report exports."
|
|
elif request.export_format not in policy.allowed_export_formats:
|
|
allowed = False
|
|
reason = "Policy does not allow the requested report export format."
|
|
required = tuple(
|
|
sorted(
|
|
set(request.declared_privacy_transforms)
|
|
| set(policy.required_privacy_transforms)
|
|
)
|
|
)
|
|
return ReportingGovernanceDecision(
|
|
allowed=allowed,
|
|
reason=reason,
|
|
retention_days=retention_days,
|
|
export_formats=(
|
|
policy.allowed_export_formats if policy.allow_exports else ()
|
|
),
|
|
required_privacy_transforms=required,
|
|
provenance={
|
|
"provider": "policy.reporting_governance",
|
|
"version": "1",
|
|
"sources": sources,
|
|
"effective": {
|
|
"allow_exports": policy.allow_exports,
|
|
"allowed_export_formats": list(policy.allowed_export_formats),
|
|
"allow_high_reidentification_risk": (
|
|
policy.allow_high_reidentification_risk
|
|
),
|
|
"max_retention_days": policy.max_retention_days,
|
|
"privacy_retention_days": retention,
|
|
"required_privacy_transforms": list(required),
|
|
},
|
|
},
|
|
)
|
|
|
|
|
|
def _effective_policy(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
) -> tuple[_Policy, list[dict[str, object]]]:
|
|
system_settings = get_system_settings(session).settings or {}
|
|
system_patch = _patch(system_settings, "system")
|
|
policy = _apply_patch(_Policy(), system_patch)
|
|
sources = [
|
|
{
|
|
"scope_type": "system",
|
|
"scope_id": None,
|
|
"applied_fields": sorted(system_patch),
|
|
}
|
|
]
|
|
tenant = session.get(Tenant, tenant_id)
|
|
if tenant is None:
|
|
raise LookupError("Tenant not found for Reporting Policy")
|
|
tenant_patch = _patch(tenant.settings or {}, "tenant")
|
|
policy = _tighten(policy, tenant_patch)
|
|
sources.append(
|
|
{
|
|
"scope_type": "tenant",
|
|
"scope_id": tenant_id,
|
|
"applied_fields": sorted(tenant_patch),
|
|
}
|
|
)
|
|
return policy, sources
|
|
|
|
|
|
def _patch(settings: Mapping[str, object], source: str) -> dict[str, object]:
|
|
raw = settings.get(REPORTING_POLICY_SETTINGS_KEY)
|
|
if raw in (None, ""):
|
|
return {}
|
|
if not isinstance(raw, Mapping):
|
|
raise ValueError(f"{source} Reporting Policy must be an object")
|
|
unknown = set(raw) - _ALLOWED_KEYS
|
|
if unknown:
|
|
raise ValueError(
|
|
f"{source} Reporting Policy has unknown fields: "
|
|
+ ", ".join(sorted(str(item) for item in unknown))
|
|
)
|
|
return dict(raw)
|
|
|
|
|
|
def _apply_patch(parent: _Policy, patch: Mapping[str, object]) -> _Policy:
|
|
return _Policy(
|
|
allow_exports=_boolean(patch, "allow_exports", parent.allow_exports),
|
|
allowed_export_formats=_formats(
|
|
patch.get("allowed_export_formats"),
|
|
default=parent.allowed_export_formats,
|
|
),
|
|
allow_high_reidentification_risk=_boolean(
|
|
patch,
|
|
"allow_high_reidentification_risk",
|
|
parent.allow_high_reidentification_risk,
|
|
),
|
|
max_retention_days=_days(
|
|
patch.get("max_retention_days"),
|
|
default=parent.max_retention_days,
|
|
),
|
|
required_privacy_transforms=_transforms(
|
|
patch.get("required_privacy_transforms"),
|
|
default=parent.required_privacy_transforms,
|
|
),
|
|
)
|
|
|
|
|
|
def _tighten(parent: _Policy, patch: Mapping[str, object]) -> _Policy:
|
|
child = _apply_patch(parent, patch)
|
|
return _Policy(
|
|
allow_exports=parent.allow_exports and child.allow_exports,
|
|
allowed_export_formats=tuple(
|
|
item
|
|
for item in parent.allowed_export_formats
|
|
if item in child.allowed_export_formats
|
|
),
|
|
allow_high_reidentification_risk=(
|
|
parent.allow_high_reidentification_risk
|
|
and child.allow_high_reidentification_risk
|
|
),
|
|
max_retention_days=_minimum_optional(
|
|
parent.max_retention_days,
|
|
child.max_retention_days,
|
|
),
|
|
required_privacy_transforms=tuple(
|
|
sorted(
|
|
set(parent.required_privacy_transforms)
|
|
| set(child.required_privacy_transforms)
|
|
)
|
|
),
|
|
)
|
|
|
|
|
|
def _boolean(
|
|
value: Mapping[str, object],
|
|
key: str,
|
|
default: bool,
|
|
) -> bool:
|
|
raw = value.get(key)
|
|
if raw is None:
|
|
return default
|
|
if not isinstance(raw, bool):
|
|
raise ValueError(f"Reporting Policy {key} must be boolean")
|
|
return raw
|
|
|
|
|
|
def _formats(value: object, *, default: tuple[str, ...]) -> tuple[str, ...]:
|
|
if value is None:
|
|
return default
|
|
if not isinstance(value, list) or any(
|
|
item not in {"json", "csv"} for item in value
|
|
):
|
|
raise ValueError("Reporting Policy export formats must contain json or csv")
|
|
return tuple(dict.fromkeys(str(item) for item in value))
|
|
|
|
|
|
def _transforms(value: object, *, default: tuple[str, ...]) -> tuple[str, ...]:
|
|
if value is None:
|
|
return default
|
|
if not isinstance(value, list) or any(
|
|
not isinstance(item, str) or not item.strip() for item in value
|
|
):
|
|
raise ValueError("Reporting Policy privacy transforms must be strings")
|
|
return tuple(sorted(set(item.strip() for item in value)))
|
|
|
|
|
|
def _days(value: object, *, default: int | None) -> int | None:
|
|
if value is None:
|
|
return default
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
|
raise ValueError("Reporting Policy retention days must be a positive integer")
|
|
return value
|
|
|
|
|
|
def _minimum_optional(left: int | None, right: int | None) -> int | None:
|
|
values = [item for item in (left, right) if item is not None]
|
|
return min(values) if values else None
|
|
|
|
|
|
__all__ = ["ReportingGovernancePolicyProvider"]
|