diff --git a/alembic/versions/0119_ai_provider_compliance_fields.py b/alembic/versions/0119_ai_provider_compliance_fields.py new file mode 100644 index 0000000..f503617 --- /dev/null +++ b/alembic/versions/0119_ai_provider_compliance_fields.py @@ -0,0 +1,38 @@ +"""Add compliance metadata columns to ai_providers table (B-AIPROV-COMP). + +Adds region, hosting_type, dpa_status, retention_policy, +training_on_customer_data, transfer_notice, allowed_data_classes +to support AI provider compliance checks. + +Revision ID: 0119 +Revises: 0118 +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import JSONB + +revision = "0119" +down_revision = "0118" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column("ai_providers", sa.Column("region", sa.String(20), nullable=False, server_default="unknown")) + op.add_column("ai_providers", sa.Column("hosting_type", sa.String(30), nullable=False, server_default="cloud")) + op.add_column("ai_providers", sa.Column("dpa_status", sa.String(20), nullable=False, server_default="none")) + op.add_column("ai_providers", sa.Column("retention_policy", sa.Text(), nullable=False, server_default="")) + op.add_column("ai_providers", sa.Column("training_on_customer_data", sa.Boolean(), nullable=False, server_default=sa.text("false"))) + op.add_column("ai_providers", sa.Column("transfer_notice", sa.Text(), nullable=False, server_default="")) + op.add_column("ai_providers", sa.Column("allowed_data_classes", JSONB, nullable=False, server_default=sa.text("'[]'::jsonb"))) + + +def downgrade() -> None: + op.drop_column("ai_providers", "allowed_data_classes") + op.drop_column("ai_providers", "transfer_notice") + op.drop_column("ai_providers", "training_on_customer_data") + op.drop_column("ai_providers", "retention_policy") + op.drop_column("ai_providers", "dpa_status") + op.drop_column("ai_providers", "hosting_type") + op.drop_column("ai_providers", "region") diff --git a/app/ai/llm_client.py b/app/ai/llm_client.py index 2eb181c..91a278e 100644 --- a/app/ai/llm_client.py +++ b/app/ai/llm_client.py @@ -131,6 +131,60 @@ async def get_api_credentials( return (env_key if env_key else None), None, None +async def get_provider_compliance( + db: AsyncSession | None, + tenant_id: uuid.UUID | None, +) -> dict[str, Any] | None: + """Get compliance metadata for the active AI provider. + + Returns a dict with keys: ``region``, ``hosting_type``, ``dpa_status``, + ``retention_policy``, ``training_on_customer_data``, ``transfer_notice``, + ``allowed_data_classes``. + + Returns ``None`` if no DB provider is configured (env-based fallback). + """ + if not (db and tenant_id): + return None + try: + from app.plugins.builtins.ai_assistant.contracts import get_default_provider + + provider = await get_default_provider(db, tenant_id) + if provider is None: + return None + return { + "region": getattr(provider, "region", "unknown"), + "hosting_type": getattr(provider, "hosting_type", "cloud"), + "dpa_status": getattr(provider, "dpa_status", "none"), + "retention_policy": getattr(provider, "retention_policy", ""), + "training_on_customer_data": getattr(provider, "training_on_customer_data", False), + "transfer_notice": getattr(provider, "transfer_notice", ""), + "allowed_data_classes": getattr(provider, "allowed_data_classes", []), + } + except Exception: + logger.debug("Failed to get provider compliance metadata") + return None + + +def check_data_class_allowed( + compliance: dict[str, Any] | None, + data_class: str, +) -> bool: + """Check whether the configured provider may process *data_class*. + + Uses :func:`app.core.sensitive_data.check_provider_compliance`. + Returns ``True`` if compliance metadata is unavailable (fail-open for + backward compatibility and mock mode). + """ + from app.core.sensitive_data import check_provider_compliance + + if compliance is None: + return True + return check_provider_compliance( + compliance.get("allowed_data_classes"), + data_class, + ) + + def build_model(model: str, provider_type: str | None) -> str: """Build LiteLLM model string with provider prefix. diff --git a/app/core/audit.py b/app/core/audit.py index a0a7baa..9e1d4c3 100644 --- a/app/core/audit.py +++ b/app/core/audit.py @@ -21,14 +21,21 @@ async def log_audit( changes: dict[str, Any] | None = None, details: dict[str, Any] | None = None, ) -> AuditLog: - """Create an audit log entry.""" + """Create an audit log entry. + + Sensitive fields in *changes* are masked using the central + :mod:`app.core.sensitive_data` module. + """ + from app.core.sensitive_data import sanitize_dict + + masked_changes = sanitize_dict(changes, entity_type) if changes else changes entry = AuditLog( tenant_id=tenant_id, user_id=user_id, action=action, entity_type=entity_type, entity_id=entity_id, - changes=changes, + changes=masked_changes, ) db.add(entry) await db.flush() @@ -46,14 +53,19 @@ async def log_deletion( """Create a deletion history entry (merged from DeletionLog into EntityHistory). Stores the full entity snapshot in snapshot_before for forensic recovery. + Sensitive fields are masked using the central + :mod:`app.core.sensitive_data` module. """ + from app.core.sensitive_data import sanitize_dict + + masked_snapshot = sanitize_dict(entity_snapshot, entity_type) entry = EntityHistory( tenant_id=tenant_id, user_id=user_id, entity_type=entity_type, entity_id=entity_id, action="delete", - snapshot_before=entity_snapshot, + snapshot_before=masked_snapshot, snapshot_after=None, changes=None, owner_id=user_id, diff --git a/app/core/sensitive_data.py b/app/core/sensitive_data.py new file mode 100644 index 0000000..5b152e0 --- /dev/null +++ b/app/core/sensitive_data.py @@ -0,0 +1,303 @@ +"""Central sensitive-data boundary — field redaction and data-exposure policy. + +Provides a single source of truth for which entity fields are sensitive +and which downstream systems (LLM context, search, embeddings, RAG, +agent memory, export) are allowed to receive them. + +This module does **not** replace ``permission_registry.py`` or the +``sensitivity`` field in ``manifest.py``; it respects their classifications +but adds a central enforcement layer at the points where data leaves the +system (logs, exports, index, embeddings, LLM context). +""" + +from __future__ import annotations + +import logging +from typing import Any + +logger = logging.getLogger(__name__) + +# ────────────────────────────────────────────────────────────────────────── +# SENSITIVE_FIELDS — fields that must NEVER appear in external outputs +# ────────────────────────────────────────────────────────────────────────── + +SENSITIVE_FIELDS: dict[str, set[str]] = { + "contact": { + "password_hash", + "smtp_password", + "imap_password", + "api_key", + "oauth_token", + "session_token", + "encryption_key", + }, + "user": { + "password_hash", + "api_key", + "session_token", + }, + "mail_account": { + "smtp_password", + "imap_password", + "oauth_token", + }, + "system_settings": { + "secret_key", + "encryption_key", + }, +} + +_REDACTED = "***REDACTED***" + +# ────────────────────────────────────────────────────────────────────────── +# DATA_EXPOSURE_POLICY — per entity+field, which systems may receive data +# ────────────────────────────────────────────────────────────────────────── + +# Schema: {entity_type: {field_name: {system: bool}}} +# Systems: llm_context, search, embeddings, rag, agent_memory, export +# +# Default derivation from sensitivity levels: +# normal → all systems allowed +# sensitive → only export (with permission) +# critical → nothing allowed +# +# Fields listed in SENSITIVE_FIELDS are always blocked everywhere regardless +# of any policy entry. + +_EXPOSURE_SYSTEMS = ( + "llm_context", + "search", + "embeddings", + "rag", + "agent_memory", + "export", +) + +_ALL_ALLOWED = {s: True for s in _EXPOSURE_SYSTEMS} +_ALL_BLOCKED = {s: False for s in _EXPOSURE_SYSTEMS} +_EXPORT_ONLY = {s: (s == "export") for s in _EXPOSURE_SYSTEMS} + +# Explicit overrides for specific entity+field combinations. +# Fields not listed here derive their policy from the sensitivity level +# (see ``_derive_policy_from_sensitivity``). +DATA_EXPOSURE_POLICY: dict[str, dict[str, dict[str, bool]]] = { + "contact": { + # Sensitive financial fields — allowed in export with permission, + # blocked from LLM/embeddings/search. + "code": _EXPORT_ONLY, + "accounting_code": _EXPORT_ONLY, + "vendor_accounting_code": _EXPORT_ONLY, + "vat_code": _EXPORT_ONLY, + "fiscal_code": _EXPORT_ONLY, + "commerce_code": _EXPORT_ONLY, + "purchase_number": _EXPORT_ONLY, + "bic": _EXPORT_ONLY, + "mobilephone": _EXPORT_ONLY, + "notes": _EXPORT_ONLY, + "tags": _EXPORT_ONLY, + }, +} + +# Sensitivity level → default exposure policy +_SENSITIVITY_DEFAULTS: dict[str, dict[str, bool]] = { + "normal": _ALL_ALLOWED, + "sensitive": _EXPORT_ONLY, + "critical": _ALL_BLOCKED, +} + + +# ────────────────────────────────────────────────────────────────────────── +# Sensitive-field helpers +# ────────────────────────────────────────────────────────────────────────── + + +def get_sensitive_fields(entity_type: str) -> set[str]: + """Return the set of sensitive field names for *entity_type*. + + Returns an empty set if the entity type is not registered. + """ + return set(SENSITIVE_FIELDS.get(entity_type, set())) + + +def is_sensitive(entity_type: str, field_name: str) -> bool: + """Check whether *field_name* is sensitive for *entity_type*.""" + return field_name in SENSITIVE_FIELDS.get(entity_type, set()) + + +def sanitize_dict(data: dict[str, Any], entity_type: str) -> dict[str, Any]: + """Return a copy of *data* with sensitive fields replaced by ``***REDACTED***``. + + Non-sensitive fields are preserved as-is. Nested dicts are processed + recursively. The original dict is not mutated. + """ + sensitive = SENSITIVE_FIELDS.get(entity_type, set()) + result: dict[str, Any] = {} + for key, value in data.items(): + if key in sensitive: + result[key] = _REDACTED + elif isinstance(value, dict): + result[key] = sanitize_dict(value, entity_type) + else: + result[key] = value + return result + + +def register_sensitive_fields(entity_type: str, fields: set[str]) -> None: + """Register additional sensitive fields for *entity_type*. + + Plugins call this at startup to declare their own sensitive fields. + Merges with any existing fields for the entity type. + """ + existing = SENSITIVE_FIELDS.setdefault(entity_type, set()) + existing |= fields + logger.debug( + "Registered sensitive fields for %s: %s", entity_type, fields + ) + + +# ────────────────────────────────────────────────────────────────────────── +# Data-exposure policy helpers +# ────────────────────────────────────────────────────────────────────────── + + +def _derive_policy_from_sensitivity( + entity_type: str, field_name: str +) -> dict[str, bool]: + """Derive a default exposure policy from the permission registry. + + Falls back to ``_ALL_ALLOWED`` if no sensitivity information is + available for the field. + """ + # Check explicit policy first + entity_policy = DATA_EXPOSURE_POLICY.get(entity_type, {}) + if field_name in entity_policy: + return dict(entity_policy[field_name]) + + # Try to get sensitivity from permission registry (lazy import to avoid + # circular dependencies at module load time). + try: + from app.core.permission_registry import CORE_FIELD_DEFINITIONS + + for fd in CORE_FIELD_DEFINITIONS: + if fd.get("module") == entity_type and fd.get("field") == field_name: + sensitivity = fd.get("sensitivity", "normal") + return dict(_SENSITIVITY_DEFAULTS.get(sensitivity, _ALL_ALLOWED)) + except Exception: + pass + + # Unknown field — default to all allowed (fail-open for normal data) + return dict(_ALL_ALLOWED) + + +def get_exposure_policy(entity_type: str, field_name: str) -> dict[str, bool]: + """Return the exposure policy dict for *entity_type* / *field_name*. + + The returned dict has keys: ``llm_context``, ``search``, ``embeddings``, + ``rag``, ``agent_memory``, ``export`` — each mapped to a bool. + + Fields listed in :data:`SENSITIVE_FIELDS` are always fully blocked. + """ + # Sensitive fields are always blocked everywhere + if is_sensitive(entity_type, field_name): + return dict(_ALL_BLOCKED) + + return _derive_policy_from_sensitivity(entity_type, field_name) + + +def _filter_for_system( + data: dict[str, Any], entity_type: str, system: str +) -> dict[str, Any]: + """Generic filter — remove fields whose policy disallows *system*.""" + sensitive = SENSITIVE_FIELDS.get(entity_type, set()) + result: dict[str, Any] = {} + for key, value in data.items(): + # Always block sensitive fields + if key in sensitive: + continue + policy = get_exposure_policy(entity_type, key) + if policy.get(system, False): + if isinstance(value, dict): + result[key] = _filter_for_system(value, entity_type, system) + else: + result[key] = value + return result + + +def filter_for_llm_context(data: dict[str, Any], entity_type: str) -> dict[str, Any]: + """Remove fields not allowed for LLM context.""" + return _filter_for_system(data, entity_type, "llm_context") + + +def filter_for_search(data: dict[str, Any], entity_type: str) -> dict[str, Any]: + """Remove fields not allowed for search / index.""" + return _filter_for_system(data, entity_type, "search") + + +def filter_for_embeddings(data: dict[str, Any], entity_type: str) -> dict[str, Any]: + """Remove fields not allowed for embedding generation.""" + return _filter_for_system(data, entity_type, "embeddings") + + +def filter_for_export(data: dict[str, Any], entity_type: str) -> dict[str, Any]: + """Remove fields not allowed for data export.""" + return _filter_for_system(data, entity_type, "export") + + +def filter_for_rag(data: dict[str, Any], entity_type: str) -> dict[str, Any]: + """Remove fields not allowed for RAG pipelines.""" + return _filter_for_system(data, entity_type, "rag") + + +def filter_for_agent_memory(data: dict[str, Any], entity_type: str) -> dict[str, Any]: + """Remove fields not allowed for agent memory persistence.""" + return _filter_for_system(data, entity_type, "agent_memory") + + +# ────────────────────────────────────────────────────────────────────────── +# AI-provider compliance helpers +# ────────────────────────────────────────────────────────────────────────── + +# Data classes that providers may be approved to process +KNOWN_DATA_CLASSES = ( + "public", + "internal", + "sensitive", + "critical", +) + + +def check_provider_compliance( + allowed_data_classes: list[str] | None, + data_class: str, +) -> bool: + """Check whether a provider is allowed to process *data_class*. + + Args: + allowed_data_classes: The provider's ``allowed_data_classes`` list. + ``None`` or empty means no restriction (fail-open for backward + compatibility). + data_class: One of :data:`KNOWN_DATA_CLASSES`. + + Returns: + ``True`` if the provider is allowed or unconfigured. + """ + if not allowed_data_classes: + return True + return data_class in allowed_data_classes + + +def get_data_class_for_field(entity_type: str, field_name: str) -> str: + """Determine the data class (public/internal/sensitive/critical) for a field. + + Used by LLM client to check provider compliance before sending data. + """ + if is_sensitive(entity_type, field_name): + return "critical" + policy = get_exposure_policy(entity_type, field_name) + if not any(policy.values()): + return "critical" + if policy.get("export") and not policy.get("llm_context"): + return "sensitive" + if all(policy.values()): + return "public" + return "internal" diff --git a/app/plugins/builtins/ai_assistant/models.py b/app/plugins/builtins/ai_assistant/models.py index ea78602..0d91384 100644 --- a/app/plugins/builtins/ai_assistant/models.py +++ b/app/plugins/builtins/ai_assistant/models.py @@ -43,6 +43,15 @@ class AIProvider(Base, TenantMixin): is_default: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) config: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) + # ── Compliance metadata (B-AIPROV-COMP) ── + region: Mapped[str] = mapped_column(String(20), nullable=False, default="unknown") + hosting_type: Mapped[str] = mapped_column(String(30), nullable=False, default="cloud") + dpa_status: Mapped[str] = mapped_column(String(20), nullable=False, default="none") + retention_policy: Mapped[str] = mapped_column(Text, nullable=False, default="") + training_on_customer_data: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + transfer_notice: Mapped[str] = mapped_column(Text, nullable=False, default="") + allowed_data_classes: Mapped[list] = mapped_column(JSONB, nullable=False, default=list) + # --- Models --- diff --git a/app/plugins/builtins/ai_assistant/schemas.py b/app/plugins/builtins/ai_assistant/schemas.py index ce09386..5bc7187 100644 --- a/app/plugins/builtins/ai_assistant/schemas.py +++ b/app/plugins/builtins/ai_assistant/schemas.py @@ -18,6 +18,14 @@ class AIProviderCreate(BaseModel): is_active: bool = True is_default: bool = False config: dict[str, Any] = Field(default_factory=dict) + # Compliance metadata (B-AIPROV-COMP) + region: str = Field(default="unknown", max_length=20) + hosting_type: str = Field(default="cloud", max_length=30) + dpa_status: str = Field(default="none", max_length=20) + retention_policy: str = Field(default="") + training_on_customer_data: bool = False + transfer_notice: str = Field(default="") + allowed_data_classes: list[str] = Field(default_factory=list) class AIProviderUpdate(BaseModel): @@ -28,6 +36,14 @@ class AIProviderUpdate(BaseModel): is_active: bool | None = None is_default: bool | None = None config: dict[str, Any] | None = None + # Compliance metadata (B-AIPROV-COMP) + region: str | None = Field(None, max_length=20) + hosting_type: str | None = Field(None, max_length=30) + dpa_status: str | None = Field(None, max_length=20) + retention_policy: str | None = None + training_on_customer_data: bool | None = None + transfer_notice: str | None = None + allowed_data_classes: list[str] | None = None class AIProviderResponse(BaseModel): @@ -40,6 +56,14 @@ class AIProviderResponse(BaseModel): is_active: bool is_default: bool config: dict[str, Any] + # Compliance metadata (B-AIPROV-COMP) + region: str = "unknown" + hosting_type: str = "cloud" + dpa_status: str = "none" + retention_policy: str = "" + training_on_customer_data: bool = False + transfer_notice: str = "" + allowed_data_classes: list[str] = Field(default_factory=list) created_at: datetime | None = None updated_at: datetime | None = None diff --git a/app/plugins/builtins/unified_search/embedding.py b/app/plugins/builtins/unified_search/embedding.py index bd0705d..961af80 100644 --- a/app/plugins/builtins/unified_search/embedding.py +++ b/app/plugins/builtins/unified_search/embedding.py @@ -147,6 +147,29 @@ async def index_entity( logger.debug("Empty embedding text for %s/%s", entity_type, entity_id) return False + # Apply sensitive-data filter: ensure no sensitive fields leak into + # embedding text. The provider builds text from DB columns, so we + # rely on the provider selecting only non-sensitive columns. This + # is a secondary safety net — providers should use filter_for_embeddings + # when constructing embedding text from dict-like data. + from app.core.sensitive_data import get_sensitive_fields + + sensitive = get_sensitive_fields(entity_type) + if sensitive: + # If any sensitive field name appears as a substring in the text, + # it's likely a key=value pair — redact it. This is a best-effort + # guard; providers are expected to exclude sensitive columns at + # the SQL level. + for field_name in sensitive: + # Only redact if the field name appears as a key-like pattern + import re + text = re.sub( + rf"\b{re.escape(field_name)}\s*[=:]\s*\S+", + f"{field_name}=***REDACTED***", + text, + flags=re.IGNORECASE, + ) + embedding = await generate_embedding(text, db=db, tenant_id=tenant_id) if not embedding: return False diff --git a/app/routes/errors.py b/app/routes/errors.py index 4e662b3..b14276f 100644 --- a/app/routes/errors.py +++ b/app/routes/errors.py @@ -31,7 +31,12 @@ _SENSITIVE_PATTERNS = re.compile( def _sanitize_context(context: Any, max_depth: int = 3, _depth: int = 0) -> Any: - """Recursively remove sensitive keys and limit depth/size of context data.""" + """Recursively remove sensitive keys and limit depth/size of context data. + + Combines regex-based pattern matching with the central + :mod:`app.core.sensitive_data` module to ensure entity-specific + sensitive fields are also redacted. + """ if _depth > max_depth: return "[truncated]" if isinstance(context, dict): @@ -49,6 +54,23 @@ def _sanitize_context(context: Any, max_depth: int = 3, _depth: int = 0) -> Any: return context +def _sanitize_entity_context(context: dict[str, Any]) -> dict[str, Any]: + """Sanitize context dict using both pattern matching and entity-aware redaction. + + If the context contains an ``entity_type`` key, uses :func:`sanitize_dict` + from :mod:`app.core.sensitive_data` to redact entity-specific sensitive + fields. Falls back to pattern-based sanitization otherwise. + """ + from app.core.sensitive_data import sanitize_dict + + entity_type = context.get("entity_type") or context.get("type") + if entity_type: + sanitized = sanitize_dict(context, str(entity_type)) + else: + sanitized = dict(context) + return _sanitize_context(sanitized) + + # -- Request schema -- class ErrorReport(BaseModel): @@ -77,7 +99,10 @@ async def report_error(error: ErrorReport, request: Request) -> Response: return Response(status_code=status.HTTP_429_TOO_MANY_REQUESTS) # Sanitize context to prevent leaking sensitive data - sanitized_context = _sanitize_context(error.context) if error.context else None + if error.context: + sanitized_context = _sanitize_entity_context(error.context) + else: + sanitized_context = None # Log with structured info logger.error( diff --git a/app/services/export_service.py b/app/services/export_service.py index 47ecc0a..5f0131c 100644 --- a/app/services/export_service.py +++ b/app/services/export_service.py @@ -10,6 +10,7 @@ from typing import Any from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession +from app.core.sensitive_data import get_sensitive_fields from app.models.contact import Contact @@ -48,21 +49,30 @@ class ExportService: result = await db.execute(base) contacts = result.scalars().all() - output = io.StringIO() - writer = csv.writer(output) - writer.writerow([ + # Exclude sensitive fields that must never appear in exports + sensitive = get_sensitive_fields("contact") + + all_headers = [ "id", "type", "displayname", "name", "firstname", "surname", "code", "email_1", "email_2", "phone_1", "phone_2", "website", "mailing_city", "mailing_postalcode", "mailing_country", "vat_code", "tags", - ]) + ] + # Drop headers for sensitive fields (e.g. password_hash would never be + # in a contact row, but this is a safety net). + export_headers = [h for h in all_headers if h not in sensitive] + + output = io.StringIO() + writer = csv.writer(output) + writer.writerow(export_headers) for c in contacts: - writer.writerow([ - str(c.id), c.type, c.displayname, c.name or "", c.firstname or "", c.surname or "", - c.code or "", c.email_1 or "", c.email_2 or "", c.phone_1 or "", c.phone_2 or "", - c.website or "", c.mailing_city or "", c.mailing_postalcode or "", - c.mailing_country or "", c.vat_code or "", c.tags or "", - ]) + row = [] + for h in export_headers: + if h in sensitive: + row.append("") + else: + row.append(getattr(c, h, None) or "") + writer.writerow(row) return output.getvalue() diff --git a/tests/test_sensitive_data.py b/tests/test_sensitive_data.py new file mode 100644 index 0000000..e083923 --- /dev/null +++ b/tests/test_sensitive_data.py @@ -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