216 lines
6.9 KiB
Python
216 lines
6.9 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.policy import DefinitionGovernanceRequest, DefinitionScopeRef
|
|
from govoplan_policy.backend.db.models import PolicyOverride
|
|
from govoplan_policy.backend.definition_governance import (
|
|
DEFINITION_POLICY_FIELDS,
|
|
DefinitionGovernancePolicyProvider,
|
|
)
|
|
from govoplan_policy.backend.definition_policy_service import (
|
|
DefinitionPolicyError,
|
|
definition_policy_response_payload,
|
|
definition_policy_state,
|
|
save_definition_policy,
|
|
)
|
|
|
|
|
|
class DefinitionPolicyOverrideTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
PolicyOverride.__table__.create(self.engine)
|
|
self.session_factory = sessionmaker(
|
|
bind=self.engine,
|
|
expire_on_commit=False,
|
|
)
|
|
self.session: Session = self.session_factory()
|
|
self.provider = DefinitionGovernancePolicyProvider()
|
|
self.actor = PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="user-1",
|
|
tenant_id="tenant-1",
|
|
group_ids=frozenset({"group-1"}),
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def _save(
|
|
self,
|
|
*,
|
|
scope_type: str,
|
|
policy: object,
|
|
scope_id: str | None = None,
|
|
module_id: str = "dataflow",
|
|
):
|
|
return save_definition_policy(
|
|
self.session,
|
|
module_id=module_id,
|
|
tenant_id="tenant-1",
|
|
scope_type=scope_type,
|
|
scope_id=scope_id,
|
|
policy=policy,
|
|
actor_id="account-1",
|
|
)
|
|
|
|
def _resolve(self, *, action: str, definition_scope: DefinitionScopeRef):
|
|
return self.provider.resolve_definition_action(
|
|
self.session,
|
|
request=DefinitionGovernanceRequest(
|
|
module_id="dataflow",
|
|
definition_ref="pipeline:1",
|
|
tenant_id="tenant-1",
|
|
definition_scope=definition_scope,
|
|
target_scope=DefinitionScopeRef("tenant", "tenant-1"),
|
|
definition_kind="flow",
|
|
action=action, # type: ignore[arg-type]
|
|
actor=self.actor,
|
|
status="active",
|
|
inherit_to_lower_scopes=True,
|
|
allow_run=True,
|
|
allow_reuse=True,
|
|
allow_automation=True,
|
|
),
|
|
)
|
|
|
|
def test_persists_and_revises_an_explicit_policy(self) -> None:
|
|
first = self._save(scope_type="tenant", policy={"allow_run": False})
|
|
second = self._save(
|
|
scope_type="tenant",
|
|
policy={"allow_run": False, "allow_reuse": False},
|
|
)
|
|
|
|
self.assertEqual(first.row.id, second.row.id)
|
|
self.assertEqual(2, second.row.revision)
|
|
self.assertEqual(
|
|
{"allow_run": False, "allow_reuse": False},
|
|
second.local_policy,
|
|
)
|
|
|
|
def test_system_and_tenant_limits_apply_to_group_and_user_targets(self) -> None:
|
|
self._save(
|
|
scope_type="system",
|
|
module_id="*",
|
|
policy={"allow_automation": False},
|
|
)
|
|
self._save(
|
|
scope_type="tenant",
|
|
policy={"allow_reuse": False},
|
|
)
|
|
|
|
group = definition_policy_state(
|
|
self.session,
|
|
module_id="dataflow",
|
|
tenant_id="tenant-1",
|
|
scope_type="group",
|
|
scope_id="group-1",
|
|
)
|
|
user = definition_policy_state(
|
|
self.session,
|
|
module_id="dataflow",
|
|
tenant_id="tenant-1",
|
|
scope_type="user",
|
|
scope_id="user-1",
|
|
)
|
|
|
|
for state in (group, user):
|
|
self.assertFalse(state.effective.limits["allow_automation"])
|
|
self.assertFalse(state.effective.limits["allow_reuse"])
|
|
self.assertEqual(
|
|
["system", "tenant:tenant-1"],
|
|
[step.path for step in state.effective.source_path],
|
|
)
|
|
|
|
def test_lower_scopes_cannot_broaden_parent_restrictions(self) -> None:
|
|
self._save(scope_type="tenant", policy={"allow_run": False})
|
|
|
|
with self.assertRaisesRegex(
|
|
DefinitionPolicyError,
|
|
"cannot broaden parent restrictions: allow_run",
|
|
):
|
|
self._save(
|
|
scope_type="group",
|
|
scope_id="group-1",
|
|
policy={"allow_run": True},
|
|
)
|
|
|
|
def test_explicit_policy_restricts_provider_actions(self) -> None:
|
|
self._save(
|
|
scope_type="tenant",
|
|
policy={"allow_edit": False, "allow_reuse": False},
|
|
)
|
|
|
|
edit = self._resolve(
|
|
action="edit",
|
|
definition_scope=DefinitionScopeRef("tenant", "tenant-1"),
|
|
)
|
|
reuse = self._resolve(
|
|
action="reuse",
|
|
definition_scope=DefinitionScopeRef("group", "group-1"),
|
|
)
|
|
|
|
self.assertFalse(edit.allowed)
|
|
self.assertFalse(reuse.allowed)
|
|
self.assertEqual(
|
|
"Editing is disabled by explicit Policy restrictions.",
|
|
edit.reason,
|
|
)
|
|
self.assertEqual("tenant:tenant-1", edit.source_path[0].path)
|
|
|
|
def test_malformed_persisted_policy_fails_closed_and_is_redacted(self) -> None:
|
|
row = PolicyOverride(
|
|
policy_family="definition",
|
|
target_key="dataflow",
|
|
tenant_id="tenant-1",
|
|
scope_type="tenant",
|
|
scope_id="tenant-1",
|
|
scope_key="tenant:tenant-1",
|
|
policy={"allow_run": "definitely", "secret": "do-not-echo"},
|
|
)
|
|
self.session.add(row)
|
|
self.session.flush()
|
|
|
|
decision = self._resolve(
|
|
action="run",
|
|
definition_scope=DefinitionScopeRef("tenant", "tenant-1"),
|
|
)
|
|
state = definition_policy_state(
|
|
self.session,
|
|
module_id="dataflow",
|
|
tenant_id="tenant-1",
|
|
scope_type="tenant",
|
|
)
|
|
payload = definition_policy_response_payload(state)
|
|
|
|
self.assertFalse(decision.allowed)
|
|
self.assertEqual(
|
|
{field: False for field in DEFINITION_POLICY_FIELDS},
|
|
decision.details["effective_limits"],
|
|
)
|
|
self.assertEqual(
|
|
[
|
|
{
|
|
"code": "definition_policy.invalid",
|
|
"scope": "tenant:tenant-1",
|
|
"target_key": "dataflow",
|
|
}
|
|
],
|
|
decision.details["policy_diagnostics"],
|
|
)
|
|
self.assertEqual(
|
|
{"configuration_status": "invalid_fail_closed"},
|
|
payload["policy"],
|
|
)
|
|
self.assertNotIn("secret", repr(decision.to_dict()))
|
|
self.assertNotIn("do-not-echo", repr(payload))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|