DataOps-platform/tests/security/test_trusted_delivery_remediation.py

from __future__ import annotations

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

import pytest

OWNER = "01900000-0000-7000-8000-000000096001"
APPROVER = "01900000-0000-7000-8000-000000096002"
USER = "01900000-0000-7000-8000-000000096003"
DOMAIN = "01900000-0000-7000-8000-000000096101"
ASSET = "01900000-0000-7000-8000-000000096201"


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

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

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

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

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

    def active_hold_for_asset(self, asset_uid):
        return None

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

    def enqueue_grant(self, value):
        existing = next((item for item in self.grants.values() if item["idempotency_key"] == value["idempotency_key"]), None)
        if existing:
            if existing["request_digest"] != value["request_digest"]:
                raise RuntimeError("trusted delivery idempotency conflict")
            return {"uid": existing["uid"]}
        self.grants[value["uid"]] = deepcopy(value)
        return {"uid": value["uid"]}

    def get_grant(self, uid):
        value = self.grants.get(uid)
        return deepcopy(value) if value else None

    def get_provision_receipt(self, grant_uid, key):
        return deepcopy(self.receipts.get((grant_uid, key)))

    def record_provision_receipt(self, grant_uid, key, _request_digest, receipt):
        self.receipts[(grant_uid, key)] = deepcopy(receipt)
        return deepcopy(receipt)


class Provider:
    def __init__(self):
        self.envelopes = []

    def apply(self, envelope):
        self.envelopes.append(deepcopy(envelope))
        return {"status": "applied", "receipt_code": "TEST", "response_digest": "a" * 64}


def policy(now):
    return {
        "code": "REMEDIATION_READ",
        "version": "1.0.0",
        "selector": {
            "subjects": [USER], "roles": ["operator"], "business_domains": [DOMAIN],
            "assets": [ASSET], "purposes": ["quality_review"], "environments": ["test"],
            "actions": ["read"], "classifications": ["sensitive"],
            "expires_at": (now + timedelta(days=1)).isoformat(),
        },
        "resource_rules": {
            "row_filter": {"op": "eq", "field": "region", "value": "CN-31"},
            "field_rules": [{"field": "code", "effect": "allow", "mask_ref": "display-code-v1"}],
        },
    }


def service():
    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(9600, 9800))
    provider = Provider()
    return (
        TrustedDeliveryService(MemoryRepository(), provider_registry=ClosedProviderRegistry.for_tests({"database": provider}), uid_factory=lambda: next(ids), now_factory=lambda: now),
        provider,
        now,
    )


def test_first_policy_stays_draft_until_approved_activation():
    trusted, _provider, now = service()
    created = trusted.create_policy_version(policy(now), actor_uid=OWNER)
    assert created["status"] == "draft"
    assert trusted.evaluate_gateway({
        "subject_uid": USER, "roles": ["operator"], "business_domain_uid": DOMAIN, "asset_uid": ASSET,
        "purpose": "quality_review", "environment": "test", "action": "read", "classification": "sensitive", "requested_fields": ["code"],
    }) == {"decision": "deny", "reason_code": "default_deny"}


def test_provision_recomputes_policy_projection_and_rejects_caller_scope_expansion():
    trusted, provider, now = service()
    created = trusted.create_policy_version(policy(now), actor_uid=OWNER)
    trusted.activate_policy_version({"code": created["code"], "version": created["version"], "approval_ref": "approval:activate", "approval_digest": "b" * 64, "idempotency_key": "activate-1"}, actor_uid=APPROVER)
    base = {
        "policy_uid": created["uid"], "approval_ref": "approval:provision", "approval_digest": "c" * 64,
        "subject_uid": USER, "asset_uid": ASSET, "business_domain_uid": DOMAIN, "purpose": "quality_review", "environment": "test", "action": "read", "classification": "sensitive", "requested_fields": ["code"],
        "expires_at": (now + timedelta(minutes=5)).isoformat(), "target_capability": {"actions": ["read"], "fields": ["code"]}, "provider": "database", "idempotency_key": "provision-1",
    }
    with pytest.raises(PermissionError):
        trusted.provision({**base, "target_capability": {"actions": ["export"], "fields": ["code"]}}, actor_uid=OWNER)
    result = trusted.provision(base, actor_uid=OWNER)
    assert result["status"] == "applied"
    envelope = provider.envelopes[-1]
    assert envelope["projection"] == {"fields": ["code"], "row_predicate": {"op": "eq", "field": "region", "value": "CN-31"}, "masking": {"code": "display-code-v1"}}
    assert "approval_ref" not in envelope
    assert "target_capability" not in envelope


def test_hold_persists_two_distinct_approval_columns_for_sql_repository():
    trusted, _provider, _now = service()
    hold = trusted.create_legal_hold({
        "asset_uid": ASSET, "operation": "freeze", "approver_refs": ["approval:one", "approval:two"], "evidence_digest": "d" * 64,
    }, actor_uid=OWNER)
    assert hold["approver_ref_one"] == "approval:one"
    assert hold["approver_ref_two"] == "approval:two"