modules/log_masking.py

"""

════════════════════════════════════════════════════════════════════════════════

MODULE: log_masking.py

PURPOSE: Proteção AGRESSIVA contra vazamento de THINK e PROVIDER

════════════════════════════════════════════════════════════════════════════════



Eliminacompletamente exposição de:

  • Pensamento interno (ThinkingEngine)

  • URL do provedor (OpenRouter, etc)

  • Modelo específico (Mistral, GPT-4, etc)

  • Embedding dimensionalidade

  • User IDs reais

  • Intent classifications

  • File paths/estrutura

  • Cloud storage endpoints



IMPLEMENTAÇÃO CRÍTICA: NÃO remove logs, apenas ofusca informação sensível.

"""

import hashlib
import hmac
import os
from datetime import datetime
from typing import Any, Dict, List, Optional
import json


class LogMasking:
    """Ofuscação agressiva de informações sensíveis em logs"""
    
    # Chave secreta para hashing (deve estar em .env)
    SECRET_SALT = os.getenv('LOG_MASKING_SALT', 'fallback-insecure-salt-change-in-env')
    
    # Dicionário de cache para IDs de usuário (memória)
    _user_id_cache: Dict[str, str] = {}
    _think_hash_cache: Dict[str, str] = {}
    _provider_cache: Dict[str, str] = {}
    
    @classmethod
    def mask_user_id(cls, user_id: str) -> str:
        """

        Converte ID do usuário em hash anônimo.

        Nunca expõe número original.

        

        Exemplo:

            Input: "111596437241877"

            Output: "[USR-a7f3c2b1]"

        """
        if not user_id:
            return "[USR-UNKNOWN]"
        
        # Check cache
        if user_id in cls._user_id_cache:
            return cls._user_id_cache[user_id]
        
        # Generate hash
        data = f"{user_id}{cls.SECRET_SALT}".encode()
        token = hashlib.sha256(data).hexdigest()[:8]
        masked = f"[USR-{token}]"
        
        # Cache
        cls._user_id_cache[user_id] = masked
        
        return masked
    
    @classmethod
    def mask_thinking(cls, thinking_content: str, depth: str = None, max_chars: int = None) -> str:
        """

        MODO DEBUG: Mostra conteúdo COMPLETO do thinking para desenvolvimento.

        Retorna o pensamento inteiro SEM truncar.

        """
        if not thinking_content:
            return "[THINK-EMPTY]"
        
        # DEBUG MODE: Mostra TUDO, sem limite
        return thinking_content
    
    @classmethod
    def mask_provider_url(cls, url: str) -> str:
        """

        Ofusca URL do provedor (OpenRouter, Azure, etc).

        Nunca expõe endpoint específico ou domínio.

        

        Exemplo:

            Input: "https://openrouter.ai/api/v1/chat/completions"

            Output: "[LLM-4d9e2a1f]"

        """
        if not url:
            return "[LLM-UNKNOWN]"
        
        # Check cache
        if url in cls._provider_cache:
            return cls._provider_cache[url]
        
        # Extract domain
        try:
            from urllib.parse import urlparse
            domain = urlparse(url).netloc or url
        except:
            domain = url
        
        # Generate hash
        data = f"{domain}{cls.SECRET_SALT}".encode()
        provider_hash = hashlib.md5(data).hexdigest()[:8]
        masked = f"[LLM-{provider_hash}]"
        
        # Cache
        cls._provider_cache[url] = masked
        
        return masked
    
    @classmethod
    def mask_model_name(cls, model_name: str) -> str:
        """

        Ofusca nome do modelo (Mistral, GPT-4, etc).

        Nunca expõe modelo específico.

        

        Exemplo:

            Input: "mistral"

            Output: "[MODEL-8c5f1a3e]"

        """
        if not model_name:
            return "[MODEL-UNKNOWN]"
        
        data = f"{model_name}{cls.SECRET_SALT}".encode()
        model_hash = hashlib.sha256(data).hexdigest()[:8]
        return f"[MODEL-{model_hash}]"
    
    @classmethod
    def mask_embedding_dim(cls, dimension: int) -> str:
        """

        Ofusca dimensionalidade de embedding.

        Expõe apenas que existe, não o valor.

        

        Exemplo:

            Input: 384

            Output: "[EMB-***]"

        """
        if not dimension:
            return "[EMB-UNKNOWN]"
        
        # Não expõe valor real
        return "[EMB-***]"
    
    @classmethod
    def mask_intent(cls, intent_list: List[str]) -> str:
        """

        Ofusca classificação de intent.

        Nunca expõe algoritmo de classificação.

        

        Exemplo:

            Input: ["indefinido", "pergunta_tecnica"]

            Output: "[INT-a7f3c2b1]"

        """
        if not intent_list:
            return "[INT-EMPTY]"
        
        intent_str = json.dumps(intent_list, sort_keys=True)
        data = f"{intent_str}{cls.SECRET_SALT}".encode()
        intent_hash = hashlib.sha256(data).hexdigest()[:8]
        return f"[INT-{intent_hash}]"
    
    @classmethod
    def mask_path(cls, path: str) -> str:
        """

        Ofusca caminhos de arquivo/estrutura.

        Nunca expõe estrutura de pastas ou cloud storage.

        

        Exemplo:

            Input: "/akira/data/cloud_sync/akira.db"

            Output: "[PATH-8f2e1c5a]"

        """
        if not path:
            return "[PATH-UNKNOWN]"
        
        data = f"{path}{cls.SECRET_SALT}".encode()
        path_hash = hashlib.md5(data).hexdigest()[:8]
        return f"[PATH-{path_hash}]"
    
    @classmethod
    def mask_group_id(cls, group_id: str) -> str:
        """

        Ofusca ID de grupo (WhatsApp group JID).

        Nunca expõe número real do grupo.

        

        Exemplo:

            Input: "120363000000000-1234567890@g.us"

            Output: "[GRP-4d9e2a1f]"

        """
        if not group_id:
            return "[GRP-UNKNOWN]"
        
        data = f"{group_id}{cls.SECRET_SALT}".encode()
        group_hash = hashlib.md5(data).hexdigest()[:8]
        return f"[GRP-{group_hash}]"
    
    @classmethod
    def mask_phone_number(cls, phone: str) -> str:
        """

        Ofusca número de telefone.

        Nunca expõe número completo.

        

        Exemplo:

            Input: "5511999999999"

            Output: "[TEL-***-9999]"

        """
        if not phone or len(phone) < 4:
            return "[TEL-UNKNOWN]"
        
        # Show only last 4 digits
        masked = f"[TEL-***-{phone[-4:]}]"
        return masked
    
    @classmethod
    def mask_response_content(cls, content: str, max_chars: int = 500) -> str:
        """

        DEBUG: Expõe conteúdo completo da resposta para debug.

        Os logs são internos apenas (dev use, não user-facing).

        """
        if not content:
            return "[RESP-EMPTY]"
        # Retorna conteúdo completo para debug
        if max_chars and len(content) > max_chars:
            return content[:max_chars] + f"... (truncated, total length={len(content)})"
        return content

    
    @classmethod
    def mask_http_request(cls, method: str, url: str, status_code: int = None) -> str:
        """

        Ofusca HTTP request completo.

        Nunca expõe URL ou endpoint.

        

        Exemplo:

            Input: ("POST", "https://openrouter.ai/api/v1/chat/completions", 200)

            Output: "[HTTP-POST-LLM-4d9e2a1f-200]"

        """
        masked_url = cls.mask_provider_url(url)
        
        if status_code:
            return f"[HTTP-{method}-{masked_url}-{status_code}]"
        else:
            return f"[HTTP-{method}-{masked_url}]"


class SecureLogger:
    """Logger que aplica masking automaticamente"""
    
    def __init__(self, logger_instance):
        """

        Wrapper para logger existente

        

        Uso:

            from modules.log_masking import SecureLogger

            from modules.config import logger

            

            secure_log = SecureLogger(logger)

            secure_log.thinking(thinking_content, depth="simples")

            secure_log.provider_request("POST", url, 200)

        """
        self.logger = logger_instance
    
    def thinking(self, content: str, depth: str = None, user_id: str = None):
        """Log thinking com proteção"""
        masked_content = LogMasking.mask_thinking(content, depth)
        masked_user = LogMasking.mask_user_id(user_id) if user_id else "[USR-UNKNOWN]"
        
        self.logger.info(f"🧠 ThinkingEngine: {masked_content} by {masked_user}")
    
    def provider_request(self, method: str, url: str, status_code: int = None):
        """Log HTTP request com proteção"""
        masked_request = LogMasking.mask_http_request(method, url, status_code)
        self.logger.info(f"🌐 {masked_request}")
    
    def embedding_saved(self, user_id: str = None, model_name: str = None, embedding_dim = None):
        """Log embedding com proteção"""
        masked_user = LogMasking.mask_user_id(user_id) if user_id else "[USR-UNKNOWN]"
        masked_model = LogMasking.mask_model_name(model_name) if model_name else "[MODEL-UNKNOWN]"
        masked_dim = LogMasking.mask_embedding_dim(embedding_dim) if embedding_dim else "[EMB-UNKNOWN]"
        
        self.logger.info(f"✅ [EMBEDDING] {masked_user}: {masked_model} {masked_dim}")
    
    def response(self, user_id: str = None, content: str = None, group_id: str = None):
        """Log resposta com proteção"""
        masked_user = LogMasking.mask_user_id(user_id) if user_id else "[USR-UNKNOWN]"
        masked_response = LogMasking.mask_response_content(content) if content else "[RESP-EMPTY]"
        masked_group = LogMasking.mask_group_id(group_id) if group_id else "[GRP-PV]"
        
        self.logger.info(f"📤 [AKIRA RESPONSE] {masked_user} in {masked_group}: {masked_response}")
    
    def checkpoint(self, user_id: str = None, user_name: str = None, message_type: str = None, is_group: bool = False, group_name: str = None, message_content: str = None):
        """Log checkpoint com proteção, exibindo a mensagem do usuário."""
        masked_user = LogMasking.mask_user_id(user_id) if user_id else "[USR-UNKNOWN]"
        grupo_label = f" [Grupo: {group_name}]" if is_group and group_name else (" [Grupo]" if is_group else " [PV]")
        
        texto_msg = f" | msg: {message_content[:300]}" if message_content else ""
        self.logger.info(f"✅ [CHECKPOINT] {user_name or masked_user}{grupo_label}: tipo={message_type or 'unknown'}{texto_msg}")


# Aplicação em api.py
"""

INTEGRAÇÃO EM api.py:



1. Imports:

   from modules.log_masking import SecureLogger, LogMasking

   

2. Inicializar:

   secure_log = SecureLogger(logger)

   

3. Usar nos endpoints:

   

   # Antes (INSEGURO):

   logger.info(f"🧠 ThinkingEngine: depth={depth}, intent={intent} | 💭 {thinking}")

   logger.info(f"HTTP Request: POST {url}")

   

   # Depois (SEGURO):

   secure_log.thinking(thinking, depth=depth, user_id=user_id)

   secure_log.provider_request("POST", url, 200)

   

4. Em checkpoints:

   

   # Antes (INSEGURO):

   logger.info(f"Checkpoint concluído em: /akira/data/cloud_sync/akira.db")

   

   # Depois (SEGURO):

   secure_log.checkpoint("/akira/data/cloud_sync/akira.db")



5. Em responses:

   

   # Antes (INSEGURO):

   logger.info(f"[AKIRA RESPONSE] resposta=738chars | remote_actions=0")

   

   # Depois (SEGURO):

   secure_log.response(user_id, response_content, group_id)

"""


# Configuration check
if __name__ == "__main__":
    print("✅ Log Masking module loaded")
    print(f"✅ Salt configured: {LogMasking.SECRET_SALT[:10]}...")
    print("✅ Ready to mask sensitive data")