feat(B-SENS): Sensitive Data Boundary + AI/Data Exposure Policy + AIProvider Compliance

B-SENS: app/core/sensitive_data.py (NEU) — zentrale Sensitive-Field-Verwaltung - SENSITIVE_FIELDS dict für contact/user/mail_account/system_settings - is_sensitive(), sanitize_dict(), register_sensitive_fields() - Integration: errors.py (Log-Redaction), audit.py (Audit-Masking), export_service.py (Export-Filter), embedding.py (Index-Filter) B-DATA-POL: AI/Data Exposure Policy - DATA_EXPOSURE_POLICY: pro Entity+Field welche Systeme erlaubt (llm_context/search/embeddings/rag/agent_memory/export) - filter_for_llm_context/search/embeddings/export/rag/agent_memory() B-AIPROV-COMP: AIProvider Compliance Metadata - Migration 0119: 7 neue Spalten an ai_providers (region, hosting_type, dpa_status, retention_policy, training_on_customer_data, transfer_notice, allowed_data_classes) - llm_client.py: get_provider_compliance() + check_data_class_allowed() B-PRIV-TEST: 76 Tests in test_sensitive_data.py — alle grün - Sensitive Fields, Exposure Policy, Provider Compliance, Secrets-always-blocked - Keine Regression: 39 LLM-Client Tests grün
2026-08-13 20:39:32 +02:00
parent bb36378494
commit b231c2d0d3
10 changed files with 939 additions and 15 deletions
@@ -0,0 +1,426 @@
+"""Tests for sensitive-data boundary and AI provider compliance (B-PRIV-TEST)."""
+
+from __future__ import annotations
+
+import pytest
+
+from app.core.sensitive_data import (
+    DATA_EXPOSURE_POLICY,
+    KNOWN_DATA_CLASSES,
+    SENSITIVE_FIELDS,
+    check_provider_compliance,
+    filter_for_embeddings,
+    filter_for_export,
+    filter_for_llm_context,
+    filter_for_search,
+    get_data_class_for_field,
+    get_exposure_policy,
+    get_sensitive_fields,
+    is_sensitive,
+    register_sensitive_fields,
+    sanitize_dict,
+)
+
+
+# ── is_sensitive ──
+
+
+class TestIsSensitive:
+    def test_known_sensitive_contact_field(self):
+        assert is_sensitive("contact", "password_hash") is True
+
+    def test_known_sensitive_mail_account_field(self):
+        assert is_sensitive("mail_account", "smtp_password") is True
+
+    def test_known_sensitive_user_field(self):
+        assert is_sensitive("user", "api_key") is True
+
+    def test_non_sensitive_contact_field(self):
+        assert is_sensitive("contact", "firstname") is False
+
+    def test_unknown_entity_type(self):
+        assert is_sensitive("nonexistent", "password_hash") is False
+
+    def test_unknown_field(self):
+        assert is_sensitive("contact", "unknown_field") is False
+
+
+# ── get_sensitive_fields ──
+
+
+class TestGetSensitiveFields:
+    def test_contact_fields(self):
+        fields = get_sensitive_fields("contact")
+        assert "password_hash" in fields
+        assert "smtp_password" in fields
+        assert "api_key" in fields
+
+    def test_user_fields(self):
+        fields = get_sensitive_fields("user")
+        assert "password_hash" in fields
+        assert "session_token" in fields
+
+    def test_unknown_entity_returns_empty(self):
+        assert get_sensitive_fields("nonexistent") == set()
+
+    def test_returns_copy(self):
+        fields = get_sensitive_fields("contact")
+        fields.add("temp_field")
+        # Original should not be modified
+        assert "temp_field" not in SENSITIVE_FIELDS["contact"]
+
+
+# ── sanitize_dict ──
+
+
+class TestSanitizeDict:
+    def test_redacts_sensitive_fields(self):
+        data = {"firstname": "John", "password_hash": "secret123", "email_1": "john@example.com"}
+        result = sanitize_dict(data, "contact")
+        assert result["firstname"] == "John"
+        assert result["password_hash"] == "***REDACTED***"
+        assert result["email_1"] == "john@example.com"
+
+    def test_preserves_non_sensitive_fields(self):
+        data = {"firstname": "Jane", "surname": "Doe"}
+        result = sanitize_dict(data, "contact")
+        assert result == data
+
+    def test_does_not_mutate_original(self):
+        data = {"password_hash": "secret"}
+        sanitize_dict(data, "contact")
+        assert data["password_hash"] == "secret"
+
+    def test_unknown_entity_type_preserves_all(self):
+        data = {"password_hash": "secret", "name": "test"}
+        result = sanitize_dict(data, "nonexistent")
+        assert result == data
+
+    def test_nested_dict_redaction(self):
+        data = {"meta": {"password_hash": "secret", "info": "ok"}}
+        result = sanitize_dict(data, "contact")
+        assert result["meta"]["password_hash"] == "***REDACTED***"
+        assert result["meta"]["info"] == "ok"
+
+    def test_user_entity_redaction(self):
+        data = {"name": "admin", "password_hash": "hashed", "api_key": "key123"}
+        result = sanitize_dict(data, "user")
+        assert result["name"] == "admin"
+        assert result["password_hash"] == "***REDACTED***"
+        assert result["api_key"] == "***REDACTED***"
+
+
+# ── register_sensitive_fields ──
+
+
+class TestRegisterSensitiveFields:
+    def test_register_new_entity_type(self):
+        register_sensitive_fields("custom_plugin_entity", {"secret_field"})
+        assert is_sensitive("custom_plugin_entity", "secret_field") is True
+        # Cleanup
+        SENSITIVE_FIELDS.pop("custom_plugin_entity", None)
+
+    def test_register_merges_with_existing(self):
+        original = set(SENSITIVE_FIELDS.get("contact", set()))
+        register_sensitive_fields("contact", {"new_secret_field"})
+        assert is_sensitive("contact", "new_secret_field") is True
+        assert is_sensitive("contact", "password_hash") is True
+        # Cleanup
+        SENSITIVE_FIELDS["contact"] = original
+
+    def test_register_multiple_fields(self):
+        register_sensitive_fields("test_multi", {"field1", "field2", "field3"})
+        fields = get_sensitive_fields("test_multi")
+        assert fields == {"field1", "field2", "field3"}
+        # Cleanup
+        SENSITIVE_FIELDS.pop("test_multi", None)
+
+
+# ── get_exposure_policy ──
+
+
+class TestGetExposurePolicy:
+    def test_sensitive_field_always_blocked(self):
+        policy = get_exposure_policy("contact", "password_hash")
+        assert all(v is False for v in policy.values())
+
+    def test_normal_field_all_allowed(self):
+        policy = get_exposure_policy("contact", "firstname")
+        assert all(v is True for v in policy.values())
+
+    def test_sensitive_financial_field_export_only(self):
+        policy = get_exposure_policy("contact", "vat_code")
+        assert policy["export"] is True
+        assert policy["llm_context"] is False
+        assert policy["embeddings"] is False
+        assert policy["search"] is False
+
+    def test_unknown_field_defaults_to_all_allowed(self):
+        policy = get_exposure_policy("contact", "totally_unknown_field")
+        assert all(v is True for v in policy.values())
+
+    def test_policy_has_all_systems(self):
+        policy = get_exposure_policy("contact", "firstname")
+        for system in ("llm_context", "search", "embeddings", "rag", "agent_memory", "export"):
+            assert system in policy
+
+
+# ── filter_for_llm_context ──
+
+
+class TestFilterForLlmContext:
+    def test_removes_sensitive_fields(self):
+        data = {"firstname": "John", "password_hash": "secret", "email_1": "john@example.com"}
+        result = filter_for_llm_context(data, "contact")
+        assert "password_hash" not in result
+        assert result["firstname"] == "John"
+
+    def test_removes_export_only_fields(self):
+        data = {"firstname": "John", "vat_code": "DE123", "notes": "private"}
+        result = filter_for_llm_context(data, "contact")
+        assert "vat_code" not in result
+        assert "notes" not in result
+        assert result["firstname"] == "John"
+
+    def test_keeps_normal_fields(self):
+        data = {"firstname": "John", "surname": "Doe", "email_1": "john@example.com"}
+        result = filter_for_llm_context(data, "contact")
+        assert result == data
+
+
+# ── filter_for_search ──
+
+
+class TestFilterForSearch:
+    def test_removes_sensitive_fields(self):
+        data = {"firstname": "John", "password_hash": "secret"}
+        result = filter_for_search(data, "contact")
+        assert "password_hash" not in result
+        assert result["firstname"] == "John"
+
+    def test_removes_export_only_fields(self):
+        data = {"firstname": "John", "vat_code": "DE123"}
+        result = filter_for_search(data, "contact")
+        assert "vat_code" not in result
+        assert result["firstname"] == "John"
+
+
+# ── filter_for_export ──
+
+
+class TestFilterForExport:
+    def test_removes_sensitive_fields(self):
+        data = {"firstname": "John", "password_hash": "secret", "vat_code": "DE123"}
+        result = filter_for_export(data, "contact")
+        assert "password_hash" not in result
+        assert result["firstname"] == "John"
+        assert result["vat_code"] == "DE123"
+
+    def test_keeps_export_allowed_fields(self):
+        data = {"firstname": "John", "vat_code": "DE123", "bic": "ABCDEF"}
+        result = filter_for_export(data, "contact")
+        assert result["firstname"] == "John"
+        assert result["vat_code"] == "DE123"
+        assert result["bic"] == "ABCDEF"
+
+
+# ── filter_for_embeddings ──
+
+
+class TestFilterForEmbeddings:
+    def test_removes_sensitive_fields(self):
+        data = {"firstname": "John", "password_hash": "secret"}
+        result = filter_for_embeddings(data, "contact")
+        assert "password_hash" not in result
+        assert result["firstname"] == "John"
+
+    def test_removes_export_only_fields(self):
+        data = {"firstname": "John", "vat_code": "DE123"}
+        result = filter_for_embeddings(data, "contact")
+        assert "vat_code" not in result
+        assert result["firstname"] == "John"
+
+
+# ── Secrets always blocked ──
+
+
+class TestSecretsAlwaysBlocked:
+    """Ensure that known secret fields are blocked in all filters."""
+
+    @pytest.mark.parametrize("field_name", [
+        "password_hash",
+        "smtp_password",
+        "imap_password",
+        "api_key",
+        "oauth_token",
+        "session_token",
+        "encryption_key",
+    ])
+    def test_secret_blocked_in_llm_context(self, field_name):
+        data = {field_name: "secret_value", "firstname": "John"}
+        result = filter_for_llm_context(data, "contact")
+        assert field_name not in result
+
+    @pytest.mark.parametrize("field_name", [
+        "password_hash",
+        "smtp_password",
+        "imap_password",
+        "api_key",
+        "oauth_token",
+        "session_token",
+        "encryption_key",
+    ])
+    def test_secret_blocked_in_search(self, field_name):
+        data = {field_name: "secret_value", "firstname": "John"}
+        result = filter_for_search(data, "contact")
+        assert field_name not in result
+
+    @pytest.mark.parametrize("field_name", [
+        "password_hash",
+        "smtp_password",
+        "imap_password",
+        "api_key",
+        "oauth_token",
+        "session_token",
+        "encryption_key",
+    ])
+    def test_secret_blocked_in_embeddings(self, field_name):
+        data = {field_name: "secret_value", "firstname": "John"}
+        result = filter_for_embeddings(data, "contact")
+        assert field_name not in result
+
+    @pytest.mark.parametrize("field_name", [
+        "password_hash",
+        "smtp_password",
+        "imap_password",
+        "api_key",
+        "oauth_token",
+        "session_token",
+        "encryption_key",
+    ])
+    def test_secret_blocked_in_export(self, field_name):
+        data = {field_name: "secret_value", "firstname": "John"}
+        result = filter_for_export(data, "contact")
+        assert field_name not in result
+
+
+# ── Exposure policy enforcement ──
+
+
+class TestExposurePolicyEnforcement:
+    def test_sensitive_field_not_in_llm_context(self):
+        data = {"firstname": "John", "vat_code": "DE123"}
+        result = filter_for_llm_context(data, "contact")
+        assert "vat_code" not in result
+        assert "firstname" in result
+
+    def test_sensitive_field_not_in_embeddings(self):
+        data = {"firstname": "John", "notes": "private notes"}
+        result = filter_for_embeddings(data, "contact")
+        assert "notes" not in result
+        assert "firstname" in result
+
+    def test_sensitive_field_in_export(self):
+        data = {"firstname": "John", "vat_code": "DE123"}
+        result = filter_for_export(data, "contact")
+        assert "vat_code" in result
+        assert result["vat_code"] == "DE123"
+
+
+# ── AI Provider compliance ──
+
+
+class TestProviderCompliance:
+    def test_allowed_data_class(self):
+        assert check_provider_compliance(["public", "internal"], "public") is True
+
+    def test_disallowed_data_class(self):
+        assert check_provider_compliance(["public"], "sensitive") is False
+
+    def test_none_allowed_classes_fails_open(self):
+        assert check_provider_compliance(None, "sensitive") is True
+
+    def test_empty_allowed_classes_fails_open(self):
+        assert check_provider_compliance([], "sensitive") is True
+
+    def test_critical_data_class_blocked(self):
+        assert check_provider_compliance(["public", "sensitive"], "critical") is False
+
+    def test_all_known_data_classes_exist(self):
+        for dc in ("public", "internal", "sensitive", "critical"):
+            assert dc in KNOWN_DATA_CLASSES
+
+
+# ── get_data_class_for_field ──
+
+
+class TestGetDataClassForField:
+    def test_sensitive_field_is_critical(self):
+        assert get_data_class_for_field("contact", "password_hash") == "critical"
+
+    def test_normal_field_is_public(self):
+        assert get_data_class_for_field("contact", "firstname") == "public"
+
+    def test_export_only_field_is_sensitive(self):
+        assert get_data_class_for_field("contact", "vat_code") == "sensitive"
+
+
+# ── Integration: non-approved provider receives no sensitive data ──
+
+
+class TestProviderDataFiltering:
+    """Verify that a non-approved provider does not receive sensitive data."""
+
+    def test_filter_removes_data_before_provider_check(self):
+        """Simulate the flow: filter data → check provider compliance."""
+        data = {
+            "firstname": "John",
+            "password_hash": "secret",
+            "vat_code": "DE123",
+            "notes": "private",
+        }
+
+        # Step 1: Filter for LLM context
+        filtered = filter_for_llm_context(data, "contact")
+
+        # Step 2: Check provider compliance for remaining fields
+        for field_name in filtered:
+            data_class = get_data_class_for_field("contact", field_name)
+            # Provider only allows public data
+            assert check_provider_compliance(["public"], data_class) is True, \
+                f"Field {field_name} with data_class={data_class} should be allowed"
+
+        # Sensitive fields should have been removed
+        assert "password_hash" not in filtered
+        assert "vat_code" not in filtered
+        assert "notes" not in filtered
+
+    def test_provider_without_sensitive_data_approval(self):
+        """A provider that only allows 'public' should not receive 'sensitive' data."""
+        data = {"firstname": "John", "vat_code": "DE123"}
+        filtered = filter_for_llm_context(data, "contact")
+
+        # vat_code should be filtered out (export-only → not in llm_context)
+        assert "vat_code" not in filtered
+
+        # Even if it weren't filtered, compliance check would block it
+        data_class = get_data_class_for_field("contact", "vat_code")
+        assert check_provider_compliance(["public"], data_class) is False
+
+    def test_provider_with_full_approval(self):
+        """A provider that allows all data classes should receive all non-sensitive data."""
+        data = {"firstname": "John", "vat_code": "DE123", "password_hash": "secret"}
+        filtered = filter_for_llm_context(data, "contact")
+
+        # password_hash always blocked
+        assert "password_hash" not in filtered
+
+        # vat_code is export-only, not allowed in llm_context
+        assert "vat_code" not in filtered
+
+        # firstname is normal, always allowed
+        assert "firstname" in filtered
+
+        # Compliance check for firstname should pass
+        data_class = get_data_class_for_field("contact", "firstname")
+        assert check_provider_compliance(["public", "sensitive", "critical"], data_class) is True