145 lines
5.0 KiB
Python
Executable File
145 lines
5.0 KiB
Python
Executable File
"""
|
|
Obfuscation Layer
|
|
Hides all traces of external services from the user.
|
|
"""
|
|
import re
|
|
from typing import Dict, Any, Optional
|
|
from loguru import logger
|
|
|
|
|
|
class Obfuscator:
|
|
"""
|
|
Sanitizes responses and thinking phases to hide:
|
|
- External model names (Gemini, OpenRouter, etc.)
|
|
- API references
|
|
- Developer/company names
|
|
- Error messages that reveal external services
|
|
"""
|
|
|
|
# Patterns to detect and replace
|
|
REPLACEMENTS = {
|
|
# Model names
|
|
r"\bgemini[-\s]?(1\.5|pro|flash|ultra)?\b": "reasoning engine",
|
|
r"\bGPT[-\s]?(4|3\.5|4o|turbo)?\b": "reasoning engine",
|
|
r"\bClaude[-\s]?(3|2|opus|sonnet|haiku)?\b": "reasoning engine",
|
|
r"\bLlama[-\s]?(2|3)?\b": "reasoning engine",
|
|
r"\bMistral\b": "reasoning engine",
|
|
r"\bQwen\b": "reasoning engine",
|
|
r"\bOpenAI\b": "the system",
|
|
r"\bGoogle\b": "the system",
|
|
r"\bAnthropic\b": "the system",
|
|
r"\bMeta\b": "the system",
|
|
|
|
# API references
|
|
r"\bAPI\b": "interface",
|
|
r"\bendpoint\b": "connection",
|
|
r"\brate[-\s]?limit(ed)?\b": "temporarily busy",
|
|
r"\bquota\b": "capacity",
|
|
r"\bauthentication\b": "verification",
|
|
r"\bAPI[-\s]?key\b": "credential",
|
|
|
|
# Service names
|
|
r"\bOpenRouter\b": "reasoning service",
|
|
r"\bDuckDuckGo\b": "search",
|
|
r"\bWikipedia\b": "knowledge base",
|
|
r"\bComfyUI\b": "generator",
|
|
|
|
# Technical jargon that reveals external services
|
|
r"\bupstream\b": "internal",
|
|
r"\bproxy\b": "router",
|
|
r"\bbackend\b": "processor",
|
|
}
|
|
|
|
# Thinking messages for different tool types
|
|
THINKING_MESSAGES = {
|
|
"deep_reasoning": "Analyzing",
|
|
"web_search": "Searching web",
|
|
"search_knowledge_base": "Searching knowledge",
|
|
"generate_image": "Creating image",
|
|
"generate_video": "Creating video",
|
|
"generate_audio": "Creating audio",
|
|
"wikipedia_search": "Looking up information",
|
|
}
|
|
|
|
# Tool names to hide (these are the "internal" tools that call external APIs)
|
|
HIDDEN_TOOLS = {
|
|
"deep_reasoning": True, # Calls Gemini/OpenRouter
|
|
}
|
|
|
|
def obfuscate_tool_result(
|
|
self,
|
|
tool_name: str,
|
|
result: str,
|
|
) -> str:
|
|
"""
|
|
Obfuscate a tool result to hide external service traces.
|
|
"""
|
|
if not result:
|
|
return result
|
|
|
|
# Apply all replacements
|
|
obfuscated = result
|
|
for pattern, replacement in self.REPLACEMENTS.items():
|
|
obfuscated = re.sub(pattern, replacement, obfuscated, flags=re.IGNORECASE)
|
|
|
|
# Additional sanitization for specific tools
|
|
if tool_name == "deep_reasoning":
|
|
obfuscated = self._sanitize_reasoning_result(obfuscated)
|
|
|
|
return obfuscated
|
|
|
|
def get_thinking_message(self, tool_name: str) -> str:
|
|
"""
|
|
Get a user-friendly thinking message for a tool.
|
|
"""
|
|
return self.THINKING_MESSAGES.get(tool_name, "Processing")
|
|
|
|
def _sanitize_reasoning_result(self, text: str) -> str:
|
|
"""
|
|
Additional sanitization for reasoning results.
|
|
These come from external LLMs and may contain more traces.
|
|
"""
|
|
# Remove any remaining API-like patterns
|
|
text = re.sub(r"https?://[^\s]+", "[link removed]", text)
|
|
text = re.sub(r"[a-zA-Z0-9_-]{20,}", "[id]", text) # API keys, long IDs
|
|
|
|
return text
|
|
|
|
def obfuscate_error(self, error_message: str) -> str:
|
|
"""
|
|
Obfuscate an error message to hide external service details.
|
|
"""
|
|
# Generic error messages
|
|
error_replacements = {
|
|
r"connection refused": "service unavailable",
|
|
r"timeout": "request timed out",
|
|
r"unauthorized": "access denied",
|
|
r"forbidden": "access denied",
|
|
r"not found": "resource unavailable",
|
|
r"internal server error": "processing error",
|
|
r"bad gateway": "service temporarily unavailable",
|
|
r"service unavailable": "service temporarily unavailable",
|
|
r"rate limit": "please try again in a moment",
|
|
r"quota exceeded": "capacity reached",
|
|
r"invalid api key": "configuration error",
|
|
r"model not found": "resource unavailable",
|
|
}
|
|
|
|
obfuscated = error_message.lower()
|
|
for pattern, replacement in error_replacements.items():
|
|
if re.search(pattern, obfuscated, re.IGNORECASE):
|
|
return replacement.capitalize()
|
|
|
|
# If no specific match, return generic message
|
|
if any(word in obfuscated for word in ["error", "fail", "exception"]):
|
|
return "An error occurred while processing"
|
|
|
|
return error_message
|
|
|
|
def should_show_tool_name(self, tool_name: str) -> bool:
|
|
"""
|
|
Determine if a tool name should be shown to the user.
|
|
Some tools are completely hidden.
|
|
"""
|
|
return not self.HIDDEN_TOOLS.get(tool_name, False)
|