193 lines
9.4 KiB
Python
193 lines
9.4 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from govoplan_core.admin.models import SystemSettings
|
|
from govoplan_core.auth import ApiPrincipal, get_api_principal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.change_sequence import ChangeSequenceEntry
|
|
from govoplan_core.core.configuration_control import (
|
|
configuration_control_snapshot,
|
|
create_configuration_change_request,
|
|
)
|
|
from govoplan_core.core.configuration_safety import (
|
|
classify_configuration_field,
|
|
plan_configuration_change,
|
|
)
|
|
from govoplan_core.db.session import get_session
|
|
from govoplan_policy.backend.api.v1.routes import router
|
|
from govoplan_policy.backend.db.models import PolicyOverride
|
|
|
|
|
|
class CampaignArchiveEncryptionApiTests(unittest.TestCase):
|
|
"""Exercise the real HTTP route, safety catalog, persistence, and history."""
|
|
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine(
|
|
"sqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
self.addCleanup(self.engine.dispose)
|
|
for table in (SystemSettings.__table__, PolicyOverride.__table__, ChangeSequenceEntry.__table__):
|
|
table.create(self.engine)
|
|
self.principal = self._principal("admin:policies:read", "admin:policies:write", "system:settings:write")
|
|
app = FastAPI()
|
|
app.include_router(router, prefix="/api/v1")
|
|
|
|
def session_dependency():
|
|
with Session(self.engine) as session:
|
|
yield session
|
|
|
|
app.dependency_overrides[get_session] = session_dependency
|
|
app.dependency_overrides[get_api_principal] = lambda: self.principal
|
|
self.client = TestClient(app)
|
|
self.addCleanup(self.client.close)
|
|
|
|
@staticmethod
|
|
def _principal(*scopes: str) -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="admin-account", membership_id="admin-user", tenant_id="tenant-1",
|
|
scopes=frozenset(scopes),
|
|
),
|
|
account=SimpleNamespace(id="admin-account"),
|
|
user=SimpleNamespace(id="admin-user"),
|
|
)
|
|
|
|
def test_system_legacy_opt_in_passes_real_catalog_and_retains_history(self) -> None:
|
|
path = "/api/v1/admin/campaign-archive-encryption/policies/system"
|
|
field = classify_configuration_field("campaign_archive_encryption_policy")
|
|
self.assertIsNotNone(field)
|
|
self.assertEqual("policy", field.owner_module)
|
|
self.assertTrue(field.validation_required)
|
|
self.assertTrue(field.rollback_history_required)
|
|
self.assertEqual({}, self.client.get(path).json()["policy"])
|
|
policy = {
|
|
"allowed_password_encryption_methods": ["aes", "zip_standard"],
|
|
"allowed_password_delivery_channels": ["phone", "letter"],
|
|
}
|
|
response = self.client.put(path, json={"policy": policy})
|
|
self.assertEqual(200, response.status_code, response.text)
|
|
self.assertEqual(policy["allowed_password_encryption_methods"], response.json()["effective_policy"]["allowed_password_encryption_methods"])
|
|
loaded = self.client.get(path)
|
|
self.assertEqual(200, loaded.status_code)
|
|
self.assertEqual(policy, loaded.json()["policy"])
|
|
with Session(self.engine) as session:
|
|
history = configuration_control_snapshot(session)["history"]
|
|
self.assertEqual(1, len(history))
|
|
self.assertEqual("campaign_archive_encryption_policy", history[0]["key"])
|
|
self.assertEqual("campaign_archive_encryption_policy.updated", history[0]["audit_event"])
|
|
self.assertEqual({}, history[0]["before"])
|
|
self.assertEqual(policy, history[0]["after"])
|
|
self.assertTrue(history[0]["plan"]["allowed"])
|
|
self.assertEqual([], history[0]["plan"]["blockers"])
|
|
audit_changes = session.query(ChangeSequenceEntry).filter(
|
|
ChangeSequenceEntry.module_id == "audit"
|
|
).all()
|
|
self.assertEqual(1, len(audit_changes))
|
|
self.assertEqual("campaign_archive_encryption_policy.updated", audit_changes[0].payload["action"])
|
|
|
|
narrowed_policy = {"allowed_password_encryption_methods": ["aes"]}
|
|
narrowed = self.client.put(path, json={"policy": narrowed_policy})
|
|
self.assertEqual(200, narrowed.status_code, narrowed.text)
|
|
with Session(self.engine) as session:
|
|
history = configuration_control_snapshot(session)["history"]
|
|
self.assertEqual(2, len(history))
|
|
self.assertEqual(policy, history[0]["before"])
|
|
self.assertEqual(policy, history[0]["rollback_value"])
|
|
self.assertEqual(narrowed_policy, history[0]["after"])
|
|
|
|
def test_read_only_actor_cannot_change_system_policy(self) -> None:
|
|
self.principal = self._principal("admin:policies:read")
|
|
policy = {"allowed_password_encryption_methods": ["aes", "zip_standard"]}
|
|
response = self.client.put(
|
|
"/api/v1/admin/campaign-archive-encryption/policies/system", json={"policy": policy}
|
|
)
|
|
self.assertEqual(403, response.status_code)
|
|
plan = plan_configuration_change("campaign_archive_encryption_policy", actor_scopes=tuple(self.principal.scopes), value=policy)
|
|
self.assertFalse(plan.allowed)
|
|
self.assertEqual(("system:settings:write", "admin:policies:write"), plan.missing_scopes)
|
|
with Session(self.engine) as session:
|
|
self.assertEqual(0, session.query(PolicyOverride).count())
|
|
|
|
def test_tenant_policy_writer_cannot_loosen_global_system_ceiling(self) -> None:
|
|
self.principal = self._principal("admin:policies:read", "admin:policies:write")
|
|
policy = {"allowed_password_encryption_methods": ["aes", "zip_standard"]}
|
|
response = self.client.put(
|
|
"/api/v1/admin/campaign-archive-encryption/policies/system", json={"policy": policy}
|
|
)
|
|
self.assertIn(response.status_code, (403, 409))
|
|
self.assertIn("system:settings:write", response.text)
|
|
with Session(self.engine) as session:
|
|
self.assertEqual(0, session.query(PolicyOverride).count())
|
|
self.assertEqual(0, session.query(ChangeSequenceEntry).count())
|
|
|
|
narrowed = self.client.put(
|
|
"/api/v1/admin/campaign-archive-encryption/policies/tenant",
|
|
json={"policy": {"allowed_password_encryption_methods": ["aes"]}},
|
|
)
|
|
self.assertEqual(200, narrowed.status_code, narrowed.text)
|
|
|
|
def test_invalid_method_and_child_ceiling_still_fail_closed(self) -> None:
|
|
path = "/api/v1/admin/campaign-archive-encryption/policies"
|
|
invalid = self.client.put(f"{path}/system", json={"policy": {"allowed_password_encryption_methods": ["plaintext"]}})
|
|
self.assertEqual(422, invalid.status_code)
|
|
child = self.client.put(f"{path}/tenant", json={"policy": {"allowed_password_encryption_methods": ["aes", "zip_standard"]}})
|
|
self.assertEqual(422, child.status_code)
|
|
self.assertIn("parent", child.text)
|
|
with Session(self.engine) as session:
|
|
self.assertEqual(0, session.query(PolicyOverride).count())
|
|
|
|
def test_configuration_preview_preserves_only_known_non_secret_enum_lists(self) -> None:
|
|
unsafe = {
|
|
"allowed_password_encryption_methods": ["aes", "literal-secret"],
|
|
"allowed_password_delivery_channels": {"password": "nested-secret"},
|
|
"password": "actual-secret",
|
|
"arbitrary_field": ["unknown-secret"],
|
|
}
|
|
with Session(self.engine) as session:
|
|
request = create_configuration_change_request(
|
|
session,
|
|
key="campaign_archive_encryption_policy",
|
|
value=unsafe,
|
|
actor_user_id="admin-user",
|
|
actor_scopes=tuple(self.principal.scopes),
|
|
dry_run=False,
|
|
target={"scope_type": "system"},
|
|
)
|
|
self.assertEqual({key: "<redacted>" for key in unsafe}, request["value_preview"])
|
|
self.assertNotIn("literal-secret", str(configuration_control_snapshot(session)))
|
|
self.assertNotIn("actual-secret", str(configuration_control_snapshot(session)))
|
|
for malformed in ("scalar-secret", ["list-secret"], None, 7):
|
|
with self.subTest(malformed=type(malformed).__name__):
|
|
malformed_request = create_configuration_change_request(
|
|
session,
|
|
key="campaign_archive_encryption_policy",
|
|
value=malformed,
|
|
actor_user_id="admin-user",
|
|
actor_scopes=tuple(self.principal.scopes),
|
|
dry_run=False,
|
|
target={"scope_type": "system"},
|
|
)
|
|
self.assertEqual("<redacted>", malformed_request["value_preview"])
|
|
snapshot = str(configuration_control_snapshot(session))
|
|
self.assertNotIn("scalar-secret", snapshot)
|
|
self.assertNotIn("list-secret", snapshot)
|
|
invalid = self.client.put(
|
|
"/api/v1/admin/campaign-archive-encryption/policies/system",
|
|
json={"policy": unsafe},
|
|
)
|
|
self.assertEqual(422, invalid.status_code)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|