GovOPlaN/govoplan-core
2
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: define policy impact subject contract

Browse Source
This commit is contained in:
zemion
2026-08-20 20:27:17 +02:00
parent 8e687c4420
commit 4f4007aff1
4 changed files with 377 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/POLICY_CONTRACTS.md
+24
View File
@@ -126,6 +126,27 @@ When the capability is absent, modules must not silently emulate cross-scope
inheritance. Their conservative fallback is limited to local tenant
definitions and disables reuse, derivation, and automation.
## Bounded Impact-Subject Providers
Policy impact previews discover optional subject providers through capability
names beginning with `policy.impactSubjects.`. The suffix is the stable
provider ID; for example, Views contributes `policy.impactSubjects.views`.
Providers implement `PolicyImpactSubjectProvider` and receive a
`PolicyImpactPopulationRequest` containing the active tenant, policy family,
an explicit selector, actor scopes, detail-disclosure decision, and a limit of
at most 500. They return `PolicyImpactSubjectBatch` with unique opaque subject
references and an explicit `complete`, `sampled`, `truncated`, or `unavailable`
state. An unavailable batch must explain the gap, and a total may never be
smaller than the returned subject count.
Core does not scan module data or evaluate domain policy. The owning module
selects and permission-filters its candidates; Policy compares the current and
proposed decisions and controls response disclosure. A caller must select one
or more provider populations explicitly. This preserves optional-module
boundaries and prevents a seemingly harmless preview from becoming an
unbounded platform query. Providers must not include credentials, secrets, or
unfiltered cross-tenant labels in subject attributes.
## Frontend Contract
Policy UIs must:
@@ -138,6 +159,9 @@ Policy UIs must:
lower-level limit to `false`
- avoid sending locked fields or re-enable attempts in save payloads
- show inherited values separately from local overrides
- require a current impact preview before enabling a governed high-impact save,
preserve its proposal hash on commit, and explain incomplete population
coverage rather than presenting unavailable providers as zero impact
The core WebUI helper `privacyRetentionParentAllowsField()` centralizes the
field-lock decision used by the retention editor and its lightweight module
src/govoplan_core/core/policy.py
+147
View File
@@ -7,6 +7,12 @@ from urllib.parse import quote, unquote
from govoplan_core.core.access import PrincipalRef
PolicyScopeType = Literal["system", "tenant", "user", "group", "campaign"]
PolicyImpactPopulationState = Literal[
"complete",
"sampled",
"truncated",
"unavailable",
]
CampaignArchiveEncryptionMethod = Literal["aes", "zip_standard"]
CampaignArchivePasswordDeliveryChannel = Literal[
"separate_mail",
@@ -54,6 +60,7 @@ CAPABILITY_POLICY_DEFINITION_GOVERNANCE = "policy.definitionGovernance"
CAPABILITY_POLICY_VIEW_GOVERNANCE = "policy.viewGovernance"
CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE = "policy.functionAssignmentGovernance"
CAPABILITY_POLICY_CAMPAIGN_ARCHIVE_ENCRYPTION = "policy.campaignArchiveEncryption"
CAPABILITY_POLICY_IMPACT_SUBJECT_PREFIX = "policy.impactSubjects."
POLICY_SCOPE_TYPES: tuple[PolicyScopeType, ...] = (
"system",
@@ -191,6 +198,146 @@ class PolicyDecision:
}
@dataclass(frozen=True, slots=True)
class PolicyImpactPopulationRequest:
"""One explicit, bounded request to an optional impact-subject provider."""
tenant_id: str
policy_family: str
selector: Mapping[str, Any] = field(default_factory=dict)
limit: int = 200
actor_scopes: tuple[str, ...] = ()
allow_sensitive_details: bool = False
def __post_init__(self) -> None:
if not self.tenant_id.strip():
raise ValueError("Policy impact population requires a tenant ID")
if not self.policy_family.strip() or len(self.policy_family) > 120:
raise ValueError(
"Policy impact population family must contain 1 to 120 characters"
)
if self.limit < 1 or self.limit > 500:
raise ValueError("Policy impact population limit must be between 1 and 500")
@dataclass(frozen=True, slots=True)
class PolicyImpactSubject:
"""Provider-owned reference safe for Policy to compare without domain imports."""
module_id: str
resource_type: str
resource_id: str
action: str
label: str | None = None
scope_type: PolicyScopeType | None = None
scope_id: str | None = None
attributes: Mapping[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
for label, value, maximum in (
("module ID", self.module_id, 80),
("resource type", self.resource_type, 80),
("resource ID", self.resource_id, 240),
("action", self.action, 120),
):
if not value.strip() or len(value) > maximum:
raise ValueError(
f"Policy impact subject {label} must contain 1 to {maximum} characters"
)
@property
def key(self) -> tuple[str, str, str, str]:
return (
self.module_id,
self.resource_type,
self.resource_id,
self.action,
)
def to_dict(self) -> dict[str, Any]:
return {
"module_id": self.module_id,
"resource_type": self.resource_type,
"resource_id": self.resource_id,
"action": self.action,
"label": self.label,
"scope_type": self.scope_type,
"scope_id": self.scope_id,
"attributes": dict(self.attributes),
}
@dataclass(frozen=True, slots=True)
class PolicyImpactSubjectBatch:
provider_id: str
subjects: tuple[PolicyImpactSubject, ...] = ()
state: PolicyImpactPopulationState = "complete"
total_available: int | None = None
explanation: str | None = None
def __post_init__(self) -> None:
if not self.provider_id.strip() or len(self.provider_id) > 120:
raise ValueError(
"Policy impact provider ID must contain 1 to 120 characters"
)
if len(self.subjects) > 500:
raise ValueError("Policy impact providers may return at most 500 subjects")
if len({subject.key for subject in self.subjects}) != len(self.subjects):
raise ValueError("Policy impact provider returned duplicate subjects")
if self.total_available is not None and self.total_available < len(self.subjects):
raise ValueError(
"Policy impact population total cannot be smaller than its subjects"
)
if self.state == "unavailable" and not self.explanation:
raise ValueError("Unavailable policy impact populations need an explanation")
def to_dict(self, *, include_subjects: bool = True) -> dict[str, Any]:
return {
"provider_id": self.provider_id,
"state": self.state,
"returned": len(self.subjects),
"total_available": self.total_available,
"explanation": self.explanation,
"subjects": (
[subject.to_dict() for subject in self.subjects]
if include_subjects
else []
),
}
@runtime_checkable
class PolicyImpactSubjectProvider(Protocol):
provider_id: str
supported_policy_families: tuple[str, ...]
def collect_policy_impact_subjects(
self,
session: object | None = None,
*,
request: PolicyImpactPopulationRequest,
) -> PolicyImpactSubjectBatch: ...
def policy_impact_subject_provider(
registry: object | None,
provider_id: str,
) -> PolicyImpactSubjectProvider | None:
clean_provider_id = provider_id.strip()
if not clean_provider_id or registry is None:
return None
capability_name = f"{CAPABILITY_POLICY_IMPACT_SUBJECT_PREFIX}{clean_provider_id}"
if (
not hasattr(registry, "has_capability")
or not registry.has_capability(capability_name)
):
return None
capability = registry.capability(capability_name)
if not isinstance(capability, PolicyImpactSubjectProvider):
return None
return capability
@dataclass(frozen=True, slots=True)
class CampaignArchiveEncryptionRequest:
"""Context required to resolve one Campaign archive-encryption ceiling.
tests/test_api_smoke.py
+115
View File
@@ -6219,6 +6219,121 @@ class ApiSmokeTests(unittest.TestCase):
self.assertEqual(campaign_source["path"], f"campaign:{campaign_id}")
self.assertIn("allow_campaign_profiles", campaign_source["applied_fields"])
def test_policy_impact_preview_is_bounded_audited_and_linked_to_commit(self) -> None:
headers, _ = self._login()
proposed_policy = {"visible_surface_ids": []}
preview = self.client.post(
"/api/v1/admin/policy-impact/preview",
headers=headers,
json={
"policy_family": "view",
"scope_type": "tenant",
"proposed_policy": proposed_policy,
"populations": [
{
"provider_id": "views",
"selector": {
"include_views": False,
"include_surfaces": True,
},
"limit": 10,
}
],
"include_details": True,
},
)
self.assertEqual(200, preview.status_code, preview.text)
preview_payload = preview.json()
self.assertEqual(10, preview_payload["counts"]["newly_denied"])
self.assertEqual("truncated", preview_payload["populations"][0]["state"])
self.assertEqual(10, len(preview_payload["effects"]))
unchanged = self.client.get(
"/api/v1/admin/view-policies/tenant",
headers=headers,
)
self.assertEqual(200, unchanged.status_code, unchanged.text)
self.assertEqual({}, unchanged.json()["policy"])
committed = self.client.put(
"/api/v1/admin/view-policies/tenant",
headers=headers,
json={
"policy": proposed_policy,
"impact_preview_id": preview_payload["preview_id"],
"impact_proposal_hash": preview_payload["proposal_hash"],
},
)
self.assertEqual(200, committed.status_code, committed.text)
self.assertEqual([], committed.json()["policy"]["visible_surface_ids"])
from govoplan_audit.backend.db.models import AuditLog
with SessionLocal() as session:
audit_rows = (
session.query(AuditLog)
.filter(
AuditLog.action.in_(
("policy.impact_previewed", "view_policy.updated")
)
)
.all()
)
by_action = {row.action: row for row in audit_rows}
self.assertEqual(
preview_payload["proposal_hash"],
by_action["policy.impact_previewed"].details["proposal_hash"],
)
self.assertEqual(
preview_payload["preview_id"],
by_action["view_policy.updated"].details["impact_preview_id"],
)
stale = self.client.put(
"/api/v1/admin/view-policies/tenant",
headers=headers,
json={
"policy": {"allow_edit": False},
"impact_preview_id": preview_payload["preview_id"],
"impact_proposal_hash": preview_payload["proposal_hash"],
},
)
self.assertEqual(409, stale.status_code, stale.text)
self.assertEqual("policy_impact_preview_stale", stale.json()["detail"]["code"])
inherited_preview = self.client.post(
"/api/v1/admin/policy-impact/preview",
headers=headers,
json={
"policy_family": "view",
"scope_type": "tenant",
"proposed_policy": {},
"populations": [
{
"provider_id": "views",
"selector": {
"include_views": False,
"include_surfaces": True,
},
"limit": 10,
}
],
},
)
self.assertEqual(200, inherited_preview.status_code, inherited_preview.text)
inherited_payload = inherited_preview.json()
removed = self.client.delete(
"/api/v1/admin/view-policies/tenant",
headers=headers,
params={
"impact_preview_id": inherited_payload["preview_id"],
"impact_proposal_hash": inherited_payload["proposal_hash"],
},
)
self.assertEqual(200, removed.status_code, removed.text)
self.assertEqual({}, removed.json()["policy"])
def test_campaign_scoped_mail_profile_policy_is_enforced(self) -> None:
headers, _ = self._login()
created = self.client.post(
tests/test_policy_impact_contract.py
+91
View File
@@ -0,0 +1,91 @@
from __future__ import annotations
import unittest
from govoplan_core.core.policy import (
CAPABILITY_POLICY_IMPACT_SUBJECT_PREFIX,
PolicyImpactPopulationRequest,
PolicyImpactSubject,
PolicyImpactSubjectBatch,
PolicyImpactSubjectProvider,
policy_impact_subject_provider,
)
class _Provider:
provider_id = "example"
supported_policy_families = ("view",)
def collect_policy_impact_subjects(
self,
session: object | None = None,
*,
request: PolicyImpactPopulationRequest,
) -> PolicyImpactSubjectBatch:
del session, request
return PolicyImpactSubjectBatch(
provider_id=self.provider_id,
subjects=(
PolicyImpactSubject(
module_id="example",
resource_type="record",
resource_id="record-1",
action="view",
),
),
total_available=1,
)
class _Registry:
def __init__(self, capability: object) -> None:
self._capability = capability
def has_capability(self, name: str) -> bool:
return name == f"{CAPABILITY_POLICY_IMPACT_SUBJECT_PREFIX}example"
def capability(self, name: str) -> object:
del name
return self._capability
class PolicyImpactContractTests(unittest.TestCase):
def test_provider_contract_is_runtime_checkable_and_discoverable(self) -> None:
provider = _Provider()
self.assertIsInstance(provider, PolicyImpactSubjectProvider)
self.assertIs(
policy_impact_subject_provider(_Registry(provider), "example"),
provider,
)
def test_population_and_provider_batches_are_bounded(self) -> None:
with self.assertRaisesRegex(ValueError, "between 1 and 500"):
PolicyImpactPopulationRequest(
tenant_id="tenant-1",
policy_family="view",
limit=501,
)
with self.assertRaisesRegex(ValueError, "at most 500"):
PolicyImpactSubjectBatch(
provider_id="example",
subjects=tuple(
PolicyImpactSubject(
module_id="example",
resource_type="record",
resource_id=str(index),
action="view",
)
for index in range(501)
),
)
def test_unavailable_batches_explain_the_gap(self) -> None:
with self.assertRaisesRegex(ValueError, "need an explanation"):
PolicyImpactSubjectBatch(
provider_id="example",
state="unavailable",
)
if __name__ == "__main__":
unittest.main()