427 lines
15 KiB
Python
427 lines
15 KiB
Python
|
"""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
|