leocrm/app/core/sensitive_data.py

"""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"