moha_mind/agent/privacy.py

Rama master · réplica de lectura del repositorio original.

"""Privacy tiers for MohaMind memory.

Defines which memory categories are "sensitive" and provides helpers to
redact content before it leaves the primary agent — e.g. before the
verifier sees it, before it goes into the session store, or before a
daily log captures it.

Design:
- Sensitivity is defined per category, not per line.
- Redaction is conservative: we keep the shape of the text but replace
  likely sensitive spans (numbers, amounts, document IDs) with tokens.
- None of this is cryptography — it only reduces accidental leakage.
"""

from __future__ import annotations

import re
from dataclasses import dataclass

from moha_mind.config import settings

DEFAULT_SENSITIVE_CATEGORIES = ("finances", "health", "documents")


@dataclass(frozen=True)
class PrivacyPolicy:
    sensitive_categories: frozenset[str]
    redact_in_sessions: bool
    redact_in_verifier: bool
    redact_in_daily_log: bool

    @classmethod
    def from_settings(cls) -> "PrivacyPolicy":
        raw = getattr(settings, "sensitive_categories", "") or ""
        if isinstance(raw, str):
            parsed = [item.strip().lower() for item in raw.split(",") if item.strip()]
        else:
            parsed = [str(item).strip().lower() for item in raw if str(item).strip()]
        categories = frozenset(parsed) if parsed else frozenset(DEFAULT_SENSITIVE_CATEGORIES)

        return cls(
            sensitive_categories=categories,
            redact_in_sessions=bool(getattr(settings, "privacy_redact_sessions", True)),
            redact_in_verifier=bool(getattr(settings, "privacy_redact_verifier", True)),
            redact_in_daily_log=bool(getattr(settings, "privacy_redact_daily_log", True)),
        )

    def is_sensitive(self, category: str) -> bool:
        return category.lower() in self.sensitive_categories


_AMOUNT_RE = re.compile(r"\b\d{3,}(?:[.,]\d{1,3})?\s?(?:SAR|USD|EUR|ريال|\$|€)?\b", re.IGNORECASE)
_LONG_DIGITS_RE = re.compile(r"\b\d{6,}\b")  # IDs, account numbers, passport numbers
_EMAIL_RE = re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b")
_PHONE_RE = re.compile(r"(?:\+?\d[\d\s-]{7,})")


def redact(text: str) -> str:
    """Return a version of text with sensitive-looking spans replaced by tags.

    This is a conservative heuristic — it does NOT guarantee full redaction.
    It is used on content that will leave the primary agent (verifier,
    session store, daily log) so accidental leakage is reduced.
    """
    if not text:
        return text
    out = _EMAIL_RE.sub("[email]", text)
    out = _PHONE_RE.sub(lambda m: "[phone]" if sum(ch.isdigit() for ch in m.group(0)) >= 8 else m.group(0), out)
    out = _LONG_DIGITS_RE.sub("[id]", out)
    out = _AMOUNT_RE.sub("[amount]", out)
    return out


def redact_if(text: str, *, when: bool) -> str:
    """Shortcut: redact only when `when` is True."""
    if not when:
        return text
    return redact(text)


def summarize_sensitive(text: str, max_len: int = 80) -> str:
    """Return a short privacy-safe placeholder used when a full redaction is too noisy."""
    redacted = redact(text or "")
    cleaned = " ".join(redacted.split())
    if len(cleaned) <= max_len:
        return cleaned
    return cleaned[: max_len - 3].rstrip() + "..."