Jarvis-Ai/src/javis/security/privacy.py

"""Conservative local privacy classification for cloud routing."""

from __future__ import annotations

import re
from dataclasses import dataclass
from enum import StrEnum

from javis.providers.base import ChatMessage


class CloudPolicy(StrEnum):
    ALLOWED = "allowed"
    ASK = "ask"
    NEVER = "never"


@dataclass(frozen=True, slots=True)
class PrivacyDecision:
    policy: CloudPolicy
    reason: str


_NEVER_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
    (
        re.compile(
            r"(?i)\b(api[-_ ]?key|passwort|password|access[-_ ]?token|refresh[-_ ]?token|"
            r"session[-_ ]?cookie|private[rn]? ssh[-_ ]?schl[üu]ssel)\b"
        ),
        "mögliches Secret",
    ),
    (
        re.compile(r"(?i)(AIza[\w-]{20,}|sk-[\w-]{16,}|BEGIN [A-Z ]*PRIVATE KEY)"),
        "Schlüsselmuster",
    ),
    (
        re.compile(
            r"(?i)\b(iban|bic|kreditkart|bankkonto|bankdaten|zahlungsdaten|"
            r"kontonummer|pin[- ]?code)\b"
        ),
        "Finanz- oder Zahlungsdaten",
    ),
    (
        re.compile(
            r"(?i)\b(symptom(?:e|en)?|blut(?:e|est|et|en|ung|ungen|wert|werte)?|"
            r"verletz(?:ung|ungen|t)|wunde[n]?|medikament(?:e|en)?|diagnos(?:e|en|tiziert)|"
            r"behandl(?:ung|ungen|e|en)|medizinisch(?:e[rsn]?|er)?\s+notfall|"
            r"notaufnahme|rettungsdienst|gesundheit|krankheit(?:en)?|patient(?:en)?|"
            r"arzt|ärzt(?:in|lich(?:e[rsn]?|er)?)|arztbericht|krankenhaus|"
            r"psychotherapie|schmerz(?:en)?|fieber|atemnot|bewusstlos|ohnmacht|"
            r"vergiftung|allergisch(?:e[rsn]?|er)?\s+reaktion)\b"
        ),
        "Gesundheitsinformation",
    ),
    (
        re.compile(
            r"(?i)\b(genaue anschrift|wohnanschrift|private[rs]? dokument|"
            r"vollständige[rs]? personenprofil|rohe? obsidian|nur lokal|"
            r"ausschließlich lokal)\b"
        ),
        "ausdrücklich lokal oder besonders sensibel",
    ),
)

_ASK_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
    (
        re.compile(
            r"(?i)\b(meine frau|mein mann|meine familie|mein kind|meine mutter|"
            r"mein vater|meine beziehung|über mich|über pascal)\b"
        ),
        "persönliche oder familiäre Information",
    ),
    (
        re.compile(r"(?i)\b[\w.+-]+@[\w.-]+\.[a-z]{2,}\b"),
        "E-Mail-Adresse",
    ),
    (
        re.compile(r"(?<!\w)(?:\+?\d[\d ()/-]{7,}\d)(?!\w)"),
        "mögliche Telefonnummer",
    ),
    (
        re.compile(
            r"(?i)\b(intern(?:e|er|es|en)?\s+"
            r"(?:hostname|server|domain|infrastruktur)|rootserver|"
            r"privates? projekt|nicht öffentlich|systeminformation|"
            r"private[rs]? termin)\b"
        ),
        "nicht öffentliche Infrastruktur oder Planung",
    ),
    (
        re.compile(r"(?i)\b(ich|mein(?:e[rsn]?|em)?|mir|mich)\b"),
        "möglicher persönlicher Bezug",
    ),
)

_ALLOWED_PATTERNS: tuple[re.Pattern[str], ...] = (
    re.compile(
        r"(?i)\b(python|programmier|quellcode|algorithmus|sqlite|linux|windows|"
        r"git|ram|arbeitsspeicher|ssd|solid state drive|http|statuscode|"
        r"transaktion|öffentliche dokumentation|allgemeine wissensfrage|"
        r"was ist|wie funktioniert)\b"
    ),
)


class PrivacyRouter:
    """Rule-based, entirely local and intentionally conservative."""

    def classify(self, text: str, *, cloud_requested: bool = False) -> PrivacyDecision:
        normalized = text.strip()
        for pattern, reason in _NEVER_PATTERNS:
            if pattern.search(normalized):
                return PrivacyDecision(CloudPolicy.NEVER, reason)
        for pattern, reason in _ASK_PATTERNS:
            if pattern.search(normalized):
                return PrivacyDecision(CloudPolicy.ASK, reason)
        if any(pattern.search(normalized) for pattern in _ALLOWED_PATTERNS):
            return PrivacyDecision(CloudPolicy.ALLOWED, "allgemeiner oder technischer Inhalt")
        if cloud_requested:
            return PrivacyDecision(
                CloudPolicy.ASK,
                "expliziter Cloudwunsch bei uneindeutigem Inhalt",
            )
        return PrivacyDecision(CloudPolicy.ASK, "Inhalt nicht eindeutig klassifizierbar")

    def minimal_context(
        self,
        messages: list[ChatMessage],
        *,
        current_policy: CloudPolicy,
        approved: bool,
        max_chars: int,
        max_messages: int,
    ) -> list[ChatMessage]:
        if current_policy is CloudPolicy.NEVER:
            return []
        if current_policy is CloudPolicy.ASK and not approved:
            return []

        selected: list[ChatMessage] = []
        remaining = max_chars
        for index in range(len(messages) - 1, max(-1, len(messages) - max_messages - 1), -1):
            message = messages[index]
            decision = self.classify(message.content)
            is_current = index == len(messages) - 1
            may_include = decision.policy is CloudPolicy.ALLOWED or (
                is_current and current_policy is CloudPolicy.ASK and approved
            )
            if not may_include:
                continue
            content = message.content[:remaining]
            if not content:
                break
            selected.append(ChatMessage(message.role, content))
            remaining -= len(content)
            if remaining <= 0:
                break
        selected.reverse()
        return selected