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

feat: govern campaign archive encryption

Browse Source
This commit is contained in:
zemion
2026-08-20 12:12:14 +02:00
parent 8fcc12dbb2
commit be5e3a7d72
11 changed files with 1070 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_policy/backend/api/v1/routes.py
+151
View File
@@ -34,6 +34,11 @@ from govoplan_policy.backend.definition_policy_service import (
save_definition_policy,
)
from govoplan_policy.backend.policy_overrides import PolicyOverrideError
from govoplan_policy.backend.campaign_archive_encryption import (
CampaignArchiveEncryptionPolicyError,
campaign_archive_encryption_policy_state,
save_campaign_archive_encryption_policy,
)
from govoplan_policy.backend.view_policy_service import (
ViewPolicyError,
remove_view_policy,
@@ -43,6 +48,8 @@ from govoplan_policy.backend.view_policy_service import (
)
from .schemas import (
CampaignArchiveEncryptionPolicyScopeRequest,
CampaignArchiveEncryptionPolicyScopeResponse,
DefinitionPolicyScopeRequest,
DefinitionPolicyScopeResponse,
PrivacyRetentionPolicyExplainResponse,
@@ -87,6 +94,150 @@ def _configuration_control_http_error(exc: ConfigurationControlError) -> HTTPExc
)
def _archive_encryption_policy_response(
*,
scope_type: str,
scope_id: str | None,
state,
) -> CampaignArchiveEncryptionPolicyScopeResponse:
row = state.row
return CampaignArchiveEncryptionPolicyScopeResponse(
scope_type=scope_type,
scope_id=scope_id,
id=row.id if row else None,
revision=row.revision if row else None,
policy=dict(row.policy) if row and isinstance(row.policy, dict) else {},
effective_policy=state.effective.to_dict(),
parent_policy=state.parent.to_dict(),
)
@router.get(
"/campaign-archive-encryption/policies/{scope_type}",
response_model=CampaignArchiveEncryptionPolicyScopeResponse,
)
def read_campaign_archive_encryption_policy(
scope_type: str,
scope_id: str | None = Query(default=None),
owner_type: str | None = Query(default=None),
owner_id: str | None = Query(default=None),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
):
_require_permission(principal, "admin:policies:read")
clean_scope = scope_type.strip().casefold()
try:
state = campaign_archive_encryption_policy_state(
session,
tenant_id=principal.tenant_id,
scope_type=clean_scope,
scope_id=scope_id,
owner_type=owner_type,
owner_id=owner_id,
)
return _archive_encryption_policy_response(
scope_type=clean_scope,
scope_id=scope_id,
state=state,
)
except (CampaignArchiveEncryptionPolicyError, PolicyOverrideError) as exc:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail=str(exc),
) from exc
@router.put(
"/campaign-archive-encryption/policies/{scope_type}",
response_model=CampaignArchiveEncryptionPolicyScopeResponse,
)
def write_campaign_archive_encryption_policy(
scope_type: str,
payload: CampaignArchiveEncryptionPolicyScopeRequest,
scope_id: str | None = Query(default=None),
owner_type: str | None = Query(default=None),
owner_id: str | None = Query(default=None),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
):
_require_permission(principal, "admin:policies:write")
clean_scope = scope_type.strip().casefold()
policy_value = payload.policy.model_dump(mode="json", exclude_none=True)
try:
before = campaign_archive_encryption_policy_state(
session,
tenant_id=principal.tenant_id,
scope_type=clean_scope,
scope_id=scope_id,
owner_type=owner_type,
owner_id=owner_id,
)
if clean_scope == "system":
approval = ensure_configuration_change_allowed(
session,
key="campaign_archive_encryption_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_campaign_archive_encryption_policy(
session,
tenant_id=principal.tenant_id,
scope_type=clean_scope,
scope_id=scope_id,
owner_type=owner_type,
owner_id=owner_id,
policy=policy_value,
actor_id=principal.user.id,
)
if clean_scope == "system":
record_configuration_change_applied(
session,
key="campaign_archive_encryption_policy",
before_value=(dict(before.row.policy) if before.row else {}),
after_value=policy_value,
actor_user_id=principal.user.id,
approval=approval,
target={"scope_type": clean_scope},
audit_event="campaign_archive_encryption_policy.updated",
)
audit_from_principal(
session,
principal,
action="campaign_archive_encryption_policy.updated",
scope="system" if clean_scope == "system" else "tenant",
object_type="campaign_archive_encryption_policy",
object_id=f"{clean_scope}:{scope_id or ''}",
details={
"scope_type": clean_scope,
"scope_id": scope_id,
"owner_type": owner_type,
"owner_id": owner_id,
"policy_hash": state.effective.policy_hash,
"fields": sorted(policy_value),
},
)
session.commit()
return _archive_encryption_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 (CampaignArchiveEncryptionPolicyError, PolicyOverrideError) as exc:
session.rollback()
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail=str(exc),
) from exc
def _definition_policy_response(
*,
module_id: str,
src/govoplan_policy/backend/api/v1/schemas.py
+30
View File
@@ -46,6 +46,36 @@ class PolicyDecisionItem(BaseModel):
details: dict[str, Any] = Field(default_factory=dict)
class CampaignArchiveEncryptionPolicyItem(BaseModel):
model_config = ConfigDict(extra="forbid")
allowed_password_encryption_methods: list[
Literal["aes", "zip_standard"]
] | None = None
allowed_password_delivery_channels: list[
Literal["separate_mail", "sms", "letter", "phone", "in_person"]
] | None = None
class CampaignArchiveEncryptionPolicyScopeRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
policy: CampaignArchiveEncryptionPolicyItem = Field(
default_factory=CampaignArchiveEncryptionPolicyItem
)
change_request_id: str | None = None
class CampaignArchiveEncryptionPolicyScopeResponse(BaseModel):
scope_type: Literal["system", "tenant", "group", "user", "campaign"]
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]
parent_policy: dict[str, Any]
class DefinitionPolicyItem(BaseModel):
model_config = ConfigDict(extra="forbid")
src/govoplan_policy/backend/campaign_archive_encryption.py
+353
View File
@@ -0,0 +1,353 @@
from __future__ import annotations
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
import hashlib
import json
from typing import Any
from sqlalchemy.orm import Session
from govoplan_core.core.policy import (
CampaignArchiveEncryptionDecision,
CampaignArchiveEncryptionRequest,
PolicySourceStep,
)
from govoplan_policy.backend.policy_overrides import (
PolicyOverride,
get_policy_override,
resolution_policy_overrides,
set_policy_override,
)
POLICY_FAMILY = "campaign_archive_encryption"
POLICY_TARGET = "*"
ENCRYPTION_METHODS = frozenset({"aes", "zip_standard"})
PASSWORD_DELIVERY_CHANNELS = frozenset(
{"separate_mail", "sms", "letter", "phone", "in_person"}
)
DEFAULT_METHODS = frozenset({"aes"})
DEFAULT_DELIVERY_CHANNELS = PASSWORD_DELIVERY_CHANNELS
class CampaignArchiveEncryptionPolicyError(ValueError):
pass
@dataclass(frozen=True, slots=True)
class CampaignArchiveEncryptionPolicyState:
row: PolicyOverride | None
effective: CampaignArchiveEncryptionDecision
parent: CampaignArchiveEncryptionDecision
class CampaignArchiveEncryptionPolicyProvider:
def resolve_campaign_archive_encryption(
self,
session: object | None = None,
*,
request: CampaignArchiveEncryptionRequest,
) -> CampaignArchiveEncryptionDecision:
if not isinstance(session, Session):
return _decision((), reason="Policy storage is unavailable; only AES is safe by default.")
rows = resolution_policy_overrides(
session,
policy_family=POLICY_FAMILY,
target_keys=(POLICY_TARGET,),
tenant_id=request.tenant_id,
group_ids=(request.owner_id,)
if request.owner_type == "group" and request.owner_id
else (),
user_ids=(request.owner_id,)
if request.owner_type == "user" and request.owner_id
else (),
campaign_ids=(request.campaign_id,),
)
return resolve_campaign_archive_encryption_rows(rows)
def validate_campaign_archive_encryption_policy(
value: object,
) -> dict[str, tuple[str, ...]]:
if not isinstance(value, Mapping):
raise CampaignArchiveEncryptionPolicyError("Archive-encryption policy must be an object")
supported = {
"allowed_password_encryption_methods": ENCRYPTION_METHODS,
"allowed_password_delivery_channels": PASSWORD_DELIVERY_CHANNELS,
}
unknown = sorted(str(key) for key in value if str(key) not in supported)
if unknown:
raise CampaignArchiveEncryptionPolicyError(
f"Unsupported archive-encryption policy fields: {', '.join(unknown)}"
)
result: dict[str, tuple[str, ...]] = {}
for key, raw in value.items():
if not isinstance(raw, Sequence) or isinstance(raw, (str, bytes)):
raise CampaignArchiveEncryptionPolicyError(f"{key} must be a list")
normalized = tuple(dict.fromkeys(str(item).strip() for item in raw))
invalid = sorted(set(normalized).difference(supported[str(key)]))
if invalid:
raise CampaignArchiveEncryptionPolicyError(
f"Unsupported values for {key}: {', '.join(invalid)}"
)
result[str(key)] = normalized
return result
def resolve_campaign_archive_encryption_rows(
rows: Sequence[PolicyOverride],
) -> CampaignArchiveEncryptionDecision:
methods = set(DEFAULT_METHODS)
channels = set(DEFAULT_DELIVERY_CHANNELS)
sources: list[PolicySourceStep] = [
PolicySourceStep(
scope_type="system",
label="Secure archive-encryption baseline",
applied_fields=(
"allowed_password_encryption_methods",
"allowed_password_delivery_channels",
),
policy={
"allowed_password_encryption_methods": sorted(DEFAULT_METHODS),
"allowed_password_delivery_channels": sorted(DEFAULT_DELIVERY_CHANNELS),
"implicit": True,
},
)
]
diagnostics: list[Mapping[str, Any]] = []
explicit_system = False
for row in rows:
try:
policy = validate_campaign_archive_encryption_policy(row.policy)
except CampaignArchiveEncryptionPolicyError as exc:
methods.clear()
channels.clear()
diagnostics.append(
{
"code": "campaign_archive_encryption.invalid_fail_closed",
"scope": row.scope_key,
"message": str(exc),
}
)
sources.append(
PolicySourceStep(
scope_type=row.scope_type, # type: ignore[arg-type]
scope_id=row.scope_id,
label=f"{row.scope_type.capitalize()} archive-encryption policy",
applied_fields=("configuration_status",),
policy={"configuration_status": "invalid_fail_closed"},
)
)
continue
configured_methods = policy.get("allowed_password_encryption_methods")
configured_channels = policy.get("allowed_password_delivery_channels")
if row.scope_type == "system" and not explicit_system:
explicit_system = True
if configured_methods is not None:
methods = set(configured_methods)
if configured_channels is not None:
channels = set(configured_channels)
sources[0] = PolicySourceStep(
scope_type="system",
label="System archive-encryption policy",
applied_fields=tuple(sorted(policy)),
policy={key: list(items) for key, items in policy.items()},
)
continue
if configured_methods is not None:
methods.intersection_update(configured_methods)
if configured_channels is not None:
channels.intersection_update(configured_channels)
sources.append(
PolicySourceStep(
scope_type=row.scope_type, # type: ignore[arg-type]
scope_id=row.scope_id,
label=f"{row.scope_type.capitalize()} archive-encryption policy",
applied_fields=tuple(sorted(policy)),
policy={key: list(items) for key, items in policy.items()},
)
)
return _decision(tuple(sources), methods=methods, channels=channels, diagnostics=diagnostics)
def campaign_archive_encryption_policy_state(
session: Session,
*,
tenant_id: str,
scope_type: str,
scope_id: str | None,
owner_type: str | None = None,
owner_id: str | None = None,
) -> CampaignArchiveEncryptionPolicyState:
clean_scope = scope_type.strip().casefold()
if clean_scope not in {"system", "tenant", "group", "user", "campaign"}:
raise CampaignArchiveEncryptionPolicyError("Unsupported policy scope")
effective_rows = _rows_for_scope(
session,
tenant_id=tenant_id,
scope_type=clean_scope,
scope_id=scope_id,
owner_type=owner_type,
owner_id=owner_id,
)
parent_rows = tuple(row for row in effective_rows if row.scope_type != clean_scope)
row = get_policy_override(
session,
policy_family=POLICY_FAMILY,
target_key=POLICY_TARGET,
tenant_id=tenant_id,
scope_type=clean_scope,
scope_id=scope_id,
)
return CampaignArchiveEncryptionPolicyState(
row=row,
effective=resolve_campaign_archive_encryption_rows(effective_rows),
parent=resolve_campaign_archive_encryption_rows(parent_rows),
)
def save_campaign_archive_encryption_policy(
session: Session,
*,
tenant_id: str,
scope_type: str,
scope_id: str | None,
owner_type: str | None,
owner_id: str | None,
policy: object,
actor_id: str | None,
) -> CampaignArchiveEncryptionPolicyState:
clean_policy = validate_campaign_archive_encryption_policy(policy)
before = campaign_archive_encryption_policy_state(
session,
tenant_id=tenant_id,
scope_type=scope_type,
scope_id=scope_id,
owner_type=owner_type,
owner_id=owner_id,
)
if scope_type.strip().casefold() != "system":
requested_methods = set(
clean_policy.get(
"allowed_password_encryption_methods",
tuple(before.parent.allowed_password_encryption_methods),
)
)
requested_channels = set(
clean_policy.get(
"allowed_password_delivery_channels",
tuple(before.parent.allowed_password_delivery_channels),
)
)
if not requested_methods.issubset(before.parent.allowed_password_encryption_methods):
raise CampaignArchiveEncryptionPolicyError(
"A child scope cannot enable an archive-encryption method blocked by its parent"
)
if not requested_channels.issubset(before.parent.allowed_password_delivery_channels):
raise CampaignArchiveEncryptionPolicyError(
"A child scope cannot enable a password-delivery channel blocked by its parent"
)
set_policy_override(
session,
policy_family=POLICY_FAMILY,
target_key=POLICY_TARGET,
tenant_id=tenant_id,
scope_type=scope_type,
scope_id=scope_id,
policy={key: list(items) for key, items in clean_policy.items()},
actor_id=actor_id,
)
return campaign_archive_encryption_policy_state(
session,
tenant_id=tenant_id,
scope_type=scope_type,
scope_id=scope_id,
owner_type=owner_type,
owner_id=owner_id,
)
def _rows_for_scope(
session: Session,
*,
tenant_id: str,
scope_type: str,
scope_id: str | None,
owner_type: str | None,
owner_id: str | None,
) -> tuple[PolicyOverride, ...]:
group_ids: tuple[str, ...] = ()
user_ids: tuple[str, ...] = ()
campaign_ids: tuple[str, ...] = ()
if scope_type == "group" and scope_id:
group_ids = (scope_id,)
elif scope_type == "user" and scope_id:
user_ids = (scope_id,)
elif scope_type == "campaign":
if not scope_id:
raise CampaignArchiveEncryptionPolicyError("Campaign scope requires scope_id")
campaign_ids = (scope_id,)
if owner_type == "group" and owner_id:
group_ids = (owner_id,)
elif owner_type == "user" and owner_id:
user_ids = (owner_id,)
elif owner_type or owner_id:
raise CampaignArchiveEncryptionPolicyError("Campaign owner context is invalid")
rows = resolution_policy_overrides(
session,
policy_family=POLICY_FAMILY,
target_keys=(POLICY_TARGET,),
tenant_id=tenant_id,
group_ids=group_ids,
user_ids=user_ids,
campaign_ids=campaign_ids,
)
maximum = {"system": 0, "tenant": 1, "group": 2, "user": 2, "campaign": 3}[scope_type]
rank = {"system": 0, "tenant": 1, "group": 2, "user": 2, "campaign": 3}
return tuple(row for row in rows if rank.get(row.scope_type, 99) <= maximum)
def _decision(
source_path: tuple[PolicySourceStep, ...],
*,
methods: set[str] | frozenset[str] = DEFAULT_METHODS,
channels: set[str] | frozenset[str] = DEFAULT_DELIVERY_CHANNELS,
reason: str | None = None,
diagnostics: Sequence[Mapping[str, Any]] = (),
) -> CampaignArchiveEncryptionDecision:
payload = {
"allowed_password_encryption_methods": sorted(methods),
"allowed_password_delivery_channels": sorted(channels),
"source_path": [step.to_dict() for step in source_path],
"diagnostics": [dict(item) for item in diagnostics],
}
digest = hashlib.sha256(
json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode()
).hexdigest()
if reason is None:
reason = (
"Legacy ZipCrypto is permitted by the effective policy."
if "zip_standard" in methods
else "Legacy ZipCrypto is blocked by the effective archive-encryption policy."
)
return CampaignArchiveEncryptionDecision(
allowed_password_encryption_methods=frozenset(methods), # type: ignore[arg-type]
allowed_password_delivery_channels=frozenset(channels), # type: ignore[arg-type]
policy_hash=digest,
source_path=source_path,
reason=reason,
diagnostics=tuple(diagnostics),
)
__all__ = [
"CampaignArchiveEncryptionPolicyError",
"CampaignArchiveEncryptionPolicyProvider",
"CampaignArchiveEncryptionPolicyState",
"campaign_archive_encryption_policy_state",
"resolve_campaign_archive_encryption_rows",
"save_campaign_archive_encryption_policy",
"validate_campaign_archive_encryption_policy",
]
src/govoplan_policy/backend/manifest.py
+64
View File
@@ -15,6 +15,7 @@ from govoplan_core.core.module_guards import (
persistent_table_uninstall_guard,
)
from govoplan_core.core.policy import (
CAPABILITY_POLICY_CAMPAIGN_ARCHIVE_ENCRYPTION,
CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE,
CAPABILITY_POLICY_PRIVACY_RETENTION,
@@ -115,6 +116,15 @@ def _access_explanation_subject_policy(context: ModuleContext) -> object:
return AccessExplanationSubjectPolicyProvider()
def _campaign_archive_encryption_policy(context: ModuleContext) -> object:
del context
from govoplan_policy.backend.campaign_archive_encryption import (
CampaignArchiveEncryptionPolicyProvider,
)
return CampaignArchiveEncryptionPolicyProvider()
ACCESS_EXPLANATION_SUBJECT_SCOPE = "policy:access_explanation:select_user"
@@ -223,6 +233,24 @@ manifest = ModuleManifest(
audience=("user", "tenant_admin", "policy_admin"),
metadata={"kind": "reference"},
),
DocumentationTopic(
id="policy.campaign-archive-encryption",
title="Govern Campaign archive encryption",
summary="Restrict password-protected Campaign ZIP formats and password-delivery channels through an explainable hierarchy.",
body=(
"The secure baseline permits AES only. An authorized policy administrator may explicitly permit legacy ZipCrypto at system scope, after which tenant, owner group or user, and campaign rules may only narrow the inherited methods. The same intersection controls the separate channel used to convey a password. Policy records the complete source path and a stable policy hash; malformed configuration fails closed. Policy changes never rewrite old build evidence, while Campaign rejects a queued or sent build whose effective policy is now more restrictive."
),
documentation_types=("admin", "user"),
audience=("system_admin", "tenant_admin", "policy_admin", "campaign_manager"),
related_modules=("campaign", "audit", "access"),
metadata={
"kind": "reference",
"help_contexts": [
"policy.campaign-archive-encryption",
"campaign.archive-encryption",
],
},
),
DocumentationTopic(
id="policy.hierarchy-overrides-and-retention",
title="Administer policy hierarchy and overrides",
@@ -389,6 +417,34 @@ manifest = ModuleManifest(
label="User View policy",
order=70,
),
ViewSurface(
id="policy.admin.system-campaign-archive-encryption",
module_id="policy",
kind="section",
label="System Campaign archive encryption",
order=75,
),
ViewSurface(
id="policy.admin.tenant-campaign-archive-encryption",
module_id="policy",
kind="section",
label="Tenant Campaign archive encryption",
order=75,
),
ViewSurface(
id="policy.admin.group-campaign-archive-encryption",
module_id="policy",
kind="section",
label="Group Campaign archive encryption",
order=75,
),
ViewSurface(
id="policy.admin.user-campaign-archive-encryption",
module_id="policy",
kind="section",
label="User Campaign archive encryption",
order=75,
),
ViewSurface(
id="policy.admin.system-retention",
module_id="policy",
@@ -432,6 +488,7 @@ manifest = ModuleManifest(
),
CAPABILITY_POLICY_PRIVACY_RETENTION: _privacy_retention_service,
CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY: _scheduling_participant_privacy_policy,
CAPABILITY_POLICY_CAMPAIGN_ARCHIVE_ENCRYPTION: _campaign_archive_encryption_policy,
},
capability_documentation={
CAPABILITY_POLICY_ACCESS_EXPLANATION_SUBJECTS: CapabilityDocumentation(
@@ -454,6 +511,13 @@ manifest = ModuleManifest(
documentation_types=("admin",),
audience=("policy_admin", "privacy_officer", "system_admin"),
),
CAPABILITY_POLICY_CAMPAIGN_ARCHIVE_ENCRYPTION: CapabilityDocumentation(
label="Campaign archive-encryption policy",
summary="Returns the restrictive format and password-delivery ceiling with complete provenance and a stable hash.",
contract_version="1.0",
documentation_types=("admin", "user"),
audience=("policy_admin", "campaign_manager"),
),
},
architecture=declared_module_architecture(
layer="governance_accountability",
src/govoplan_policy/backend/policy_overrides.py
+28 -5
View File
@@ -9,8 +9,10 @@ from sqlalchemy.orm import Session
from govoplan_policy.backend.db.models import PolicyOverride
POLICY_SCOPE_TYPES = frozenset({"system", "tenant", "group", "user"})
POLICY_FAMILIES = frozenset({"definition", "distribution_channels", "view"})
POLICY_SCOPE_TYPES = frozenset({"system", "tenant", "group", "user", "campaign"})
POLICY_FAMILIES = frozenset(
{"campaign_archive_encryption", "definition", "distribution_channels", "view"}
)
class PolicyOverrideError(ValueError):
@@ -22,7 +24,7 @@ def normalize_policy_target(policy_family: str, target_key: str) -> tuple[str, s
target = target_key.strip().casefold()
if family not in POLICY_FAMILIES:
raise PolicyOverrideError(
"Policy family must be definition, distribution_channels, or view"
"Policy family must be campaign_archive_encryption, definition, distribution_channels, or view"
)
if not target or len(target) > 120:
raise PolicyOverrideError("Policy target must contain 1 to 120 characters")
@@ -40,7 +42,9 @@ def normalize_policy_scope(
clean_scope = scope_type.strip().casefold()
clean_id = str(scope_id or "").strip() or None
if clean_scope not in POLICY_SCOPE_TYPES:
raise PolicyOverrideError("Policy scope must be system, tenant, group, or user")
raise PolicyOverrideError(
"Policy scope must be system, tenant, group, user, or campaign"
)
if clean_scope == "system":
if clean_id is not None:
raise PolicyOverrideError("System policy cannot declare a scope ID")
@@ -142,6 +146,7 @@ def resolution_policy_overrides(
tenant_id: str,
group_ids: Iterable[str] = (),
user_ids: Iterable[str] = (),
campaign_ids: Iterable[str] = (),
) -> tuple[PolicyOverride, ...]:
family = policy_family.strip().casefold()
targets = tuple(
@@ -151,6 +156,9 @@ def resolution_policy_overrides(
)
groups = tuple(sorted({str(value) for value in group_ids if str(value)}))
users = tuple(sorted({str(value) for value in user_ids if str(value)}))
campaigns = tuple(
sorted({str(value) for value in campaign_ids if str(value)})
)
scope_filters = [
PolicyOverride.scope_key == "system",
PolicyOverride.scope_key == f"tenant:{tenant_id}",
@@ -167,6 +175,15 @@ def resolution_policy_overrides(
[f"user:{tenant_id}:{user_id}" for user_id in users]
)
)
if campaigns:
scope_filters.append(
PolicyOverride.scope_key.in_(
[
f"campaign:{tenant_id}:{campaign_id}"
for campaign_id in campaigns
]
)
)
rows = (
session.query(PolicyOverride)
.filter(
@@ -177,7 +194,13 @@ def resolution_policy_overrides(
.all()
)
target_order = {target: index for index, target in enumerate(targets)}
scope_order = {"system": 0, "tenant": 1, "group": 2, "user": 3}
scope_order = {
"system": 0,
"tenant": 1,
"group": 2,
"user": 2,
"campaign": 3,
}
return tuple(
sorted(
rows,