DataOps-platform/tests/security/test_trusted_delivery.py

from __future__ import annotations

from copy import deepcopy
from datetime import UTC, datetime, timedelta

import pytest

OWNER = "01900000-0000-7000-8000-000000060001"
APPROVER = "01900000-0000-7000-8000-000000060002"
USER = "01900000-0000-7000-8000-000000060003"
DOMAIN = "01900000-0000-7000-8000-000000060101"
ASSET = "01900000-0000-7000-8000-000000060201"


class MemoryTrustedDeliveryRepository:
    def __init__(self):
        self.users = {OWNER, APPROVER, USER}
        self.policies = {}
        self.grants = {}
        self.holds = {}
        self.events = []

    def users_available(self, values):
        return set(values) & self.users

    def create_policy_version(self, record):
        self.policies[record["uid"]] = deepcopy(record)
        return deepcopy(record)

    def active_policies(self):
        return [deepcopy(value) for value in self.policies.values() if value["status"] == "active"]

    def create_grant(self, record):
        self.grants[record["uid"]] = deepcopy(record)
        return deepcopy(record)

    def grants_due_for_revoke(self, instant):
        return [deepcopy(value) for value in self.grants.values() if value["status"] == "active"]

    def active_hold_for_asset(self, asset_uid):
        return next((deepcopy(value) for value in self.holds.values() if value["asset_uid"] == asset_uid and value["status"] == "active"), None)

    def update_grant_status(self, uid, expected_version, status, reason_code):
        saved = deepcopy(self.grants[uid])
        if saved["current_version"] != expected_version:
            raise RuntimeError("trusted delivery version conflict")
        saved.update(status=status, reason_code=reason_code, current_version=expected_version + 1)
        self.grants[uid] = saved
        return deepcopy(saved)

    def create_legal_hold(self, record):
        self.holds[record["uid"]] = deepcopy(record)
        return deepcopy(record)

    def add_evidence(self, record):
        self.events.append(deepcopy(record))


class FakeProvider:
    def __init__(self):
        self.calls = []

    def apply(self, envelope):
        self.calls.append(deepcopy(envelope))
        return {"status": "applied", "receipt_code": "fake-accepted", "response_digest": "a" * 64}


@pytest.fixture()
def trusted():
    from app.core.system.trusted_delivery import (
        ClosedProviderRegistry,
        TrustedDeliveryService,
    )

    now = datetime(2026, 8, 11, 8, 0, tzinfo=UTC)
    ids = iter(f"01900000-0000-7000-8000-{value:012d}" for value in range(6300, 6500))
    provider = FakeProvider()
    registry = ClosedProviderRegistry.for_tests({"database": provider})
    repository = MemoryTrustedDeliveryRepository()
    return (
        TrustedDeliveryService(
            repository, provider_registry=registry, uid_factory=lambda: next(ids), now_factory=lambda: now
        ),
        repository,
        provider,
        now,
    )


def _policy_payload(now):
    return {
        "code": "MATERIAL_TEST_READ",
        "version": "1.0.0",
        "selector": {
            "subjects": [USER], "roles": ["editor"], "business_domains": [DOMAIN],
            "assets": [ASSET], "purposes": ["quality_review"], "environments": ["test"],
            "actions": ["read"], "classifications": ["sensitive"],
            "expires_at": (now + timedelta(days=3)).isoformat(),
        },
        "resource_rules": {
            "row_filter": {"op": "eq", "field": "region_code", "value": "CN-31"},
            "field_rules": [
                {"field": "material_code", "effect": "allow", "mask_ref": None},
                {"field": "supplier_phone", "effect": "allow", "mask_ref": "display-phone-v1"},
            ],
        },
    }


def test_gateway_matches_structured_policy_and_returns_safe_projection(trusted):
    service, _repository, _provider, now = trusted
    policy = service.create_policy_version(_policy_payload(now), actor_uid=OWNER)
    service.activate_policy_version({"code": policy["code"], "version": policy["version"], "approval_ref": "approval:gateway", "approval_digest": "a" * 64, "idempotency_key": "gateway-activate"}, actor_uid=APPROVER)

    decision = service.evaluate_gateway({
        "subject_uid": USER, "roles": ["editor"], "business_domain_uid": DOMAIN,
        "asset_uid": ASSET, "purpose": "quality_review", "environment": "test",
        "action": "read", "classification": "sensitive",
        "requested_fields": ["material_code", "supplier_phone"],
    })

    assert decision == {
        "decision": "allow", "reason_code": "policy_allowed",
        "policy_code": "MATERIAL_TEST_READ", "policy_version": "1.0.0",
        "projected_fields": ["material_code", "supplier_phone"],
        "row_filter": {"op": "eq", "field": "region_code", "value": "CN-31"},
        "field_masks": {"supplier_phone": "display-phone-v1"},
    }


def test_gateway_rejects_unsafe_dsl_and_high_sensitivity_export(trusted):
    service, _repository, _provider, now = trusted
    malformed = _policy_payload(now)
    malformed["resource_rules"]["row_filter"] = {"op": "sql", "value": "select 1"}
    with pytest.raises(ValueError, match="unsupported row filter"):
        service.create_policy_version(malformed, actor_uid=OWNER)

    service.create_policy_version(_policy_payload(now), actor_uid=OWNER)
    denied = service.evaluate_gateway({
        "subject_uid": USER, "roles": ["editor"], "business_domain_uid": DOMAIN,
        "asset_uid": ASSET, "purpose": "quality_review", "environment": "test",
        "action": "export", "classification": "highly_sensitive", "requested_fields": ["material_code"],
    })
    assert denied == {"decision": "deny", "reason_code": "highly_sensitive_export_denied"}


def test_grant_is_approval_bound_provider_disabled_fails_closed_and_expiry_recovery_obeys_hold(trusted):
    service, repository, provider, now = trusted
    policy = service.create_policy_version(_policy_payload(now), actor_uid=OWNER)
    service.activate_policy_version({"code": policy["code"], "version": policy["version"], "approval_ref": "approval:grant", "approval_digest": "a" * 64, "idempotency_key": "grant-activate"}, actor_uid=APPROVER)
    decision = service.evaluate_gateway({
        "subject_uid": USER, "roles": ["editor"], "business_domain_uid": DOMAIN,
        "asset_uid": ASSET, "purpose": "quality_review", "environment": "test",
        "action": "read", "classification": "sensitive", "requested_fields": ["material_code"],
    })
    grant = service.create_grant({
        "policy_uid": policy["uid"], "approval_ref": "approval:wp06-test:1",
        "approval_digest": "b" * 64, "subject_uid": USER, "asset_uid": ASSET,
        "provider": "database", "expires_at": (now + timedelta(minutes=1)).isoformat(),
        "decision": decision,
    }, actor_uid=OWNER)
    receipt = service.dispatch_grant(grant["uid"], actor_uid=OWNER)
    assert receipt["status"] == "applied"
    assert provider.calls[0]["grant_uid"] == grant["uid"]
    assert "approval_ref" not in provider.calls[0]
    assert service.provider_registry.is_enterprise_ready is False

    with pytest.raises(PermissionError, match="disabled"):
        service.create_grant({
            "policy_uid": policy["uid"], "approval_ref": "approval:wp06-test:2", "approval_digest": "c" * 64,
            "subject_uid": USER, "asset_uid": ASSET, "provider": "iam",
            "expires_at": (now + timedelta(minutes=1)).isoformat(), "decision": decision,
        }, actor_uid=OWNER)

    hold = service.create_legal_hold({
        "asset_uid": ASSET, "operation": "freeze", "approver_refs": ["approval:one", "approval:two"],
        "evidence_digest": "d" * 64,
    }, actor_uid=OWNER)
    assert hold["status"] == "active"
    assert service.revoke_expired(as_of=now + timedelta(minutes=2), actor_uid=OWNER) == []
    assert repository.grants[grant["uid"]]["status"] == "active"


def test_mask_preview_is_deterministic_and_never_echoes_raw_input(trusted):
    service, _repository, _provider, _now = trusted
    first = service.mask_preview({"mode": "display", "mask_ref": "display-phone-v1", "value": "13800138000"})
    second = service.mask_preview({"mode": "display", "mask_ref": "display-phone-v1", "value": "13800138000"})
    assert first == second
    assert "13800138000" not in repr(first)
    assert first["masked_value"] == "***-****"


def test_lifecycle_methods_are_available_for_versioned_policy_and_reclaim(trusted):
    """P3-WP06 B1 closes the durable lifecycle rather than only previewing it."""
    service, _repository, _provider, _now = trusted

    assert callable(service.activate_policy_version)
    assert callable(service.rollback_policy_version)
    assert callable(service.release_legal_hold)
    assert callable(service.execute_reclaim)


def test_policy_activation_and_hold_release_require_independent_replay_safe_approval(trusted):
    service, repository, _provider, now = trusted
    first = service.create_policy_version(_policy_payload(now), actor_uid=OWNER)
    second_payload = _policy_payload(now)
    second_payload["version"] = "2.0.0"
    second = service.create_policy_version(second_payload, actor_uid=OWNER)
    assert second["status"] == "draft"

    request = {
        "code": first["code"], "version": "2.0.0", "approval_ref": "approval:activate:2",
        "approval_digest": "1" * 64, "idempotency_key": "policy-activate-2",
    }
    activated = service.activate_policy_version(request, actor_uid=APPROVER)
    assert activated["status"] == "active"
    assert repository.policies[first["uid"]]["status"] == "draft"
    assert service.activate_policy_version(request, actor_uid=APPROVER) == activated
    with pytest.raises(RuntimeError, match="idempotency conflict"):
        service.activate_policy_version({**request, "version": "1.0.0"}, actor_uid=APPROVER)

    hold = service.create_legal_hold({
        "asset_uid": ASSET, "operation": "freeze", "approver_refs": ["approval:hold:one", "approval:hold:two"],
        "evidence_digest": "2" * 64,
    }, actor_uid=OWNER)
    release = {"hold_uid": hold["uid"], "approval_ref": "approval:release:one", "approval_digest": "3" * 64, "idempotency_key": "hold-release-1"}
    with pytest.raises(PermissionError, match="independent"):
        service.release_legal_hold(release, actor_uid=OWNER)
    released = service.release_legal_hold(release, actor_uid=APPROVER)
    assert released["status"] == "released"
    assert service.release_legal_hold(release, actor_uid=APPROVER) == released


def test_active_hold_denies_export_gateway_decision(trusted):
    service, _repository, _provider, now = trusted
    payload = _policy_payload(now)
    payload["selector"]["actions"] = ["export"]
    policy = service.create_policy_version(payload, actor_uid=OWNER)
    service.activate_policy_version({"code": policy["code"], "version": policy["version"], "approval_ref": "approval:export", "approval_digest": "a" * 64, "idempotency_key": "export-activate"}, actor_uid=APPROVER)
    service.create_legal_hold({
        "asset_uid": ASSET, "operation": "export_review", "approver_refs": ["approval:export:one", "approval:export:two"],
        "evidence_digest": "4" * 64,
    }, actor_uid=OWNER)

    assert service.evaluate_gateway({
        "subject_uid": USER, "roles": ["editor"], "business_domain_uid": DOMAIN,
        "asset_uid": ASSET, "purpose": "quality_review", "environment": "test",
        "action": "export", "classification": "sensitive", "requested_fields": ["material_code"],
    }) == {"decision": "deny", "reason_code": "active_legal_hold"}


def test_closed_contract_rejects_bare_secret_material_without_echo(trusted):
    service, _repository, _provider, _now = trusted
    with pytest.raises(ValueError, match="unsafe"):
        service.mask_preview({"mode": "display", "mask_ref": "display-phone-v1", "value": "password"})