Create Detector.py

Detector.py

@@ -0,0 +1,250 @@
+"""
+Detector de Prompt Injection — Motor principal
+Combina regex, heurísticas e scoring semântico
+"""
+
+import re
+import unicodedata
+import hashlib
+import time
+import random
+from dataclasses import dataclass, field
+from typing import Optional
+from enum import Enum
+
+
+class ThreatLevel(str, Enum):
+    CLEAN = "CLEAN"
+    SUSPICIOUS = "SUSPICIOUS"
+    BLOCKED = "BLOCKED"
+
+
+@dataclass
+class DetectionResult:
+    trace_id: str
+    threat_level: ThreatLevel
+    risk_score: int
+    threats_found: list
+    modifications: list
+    sanitized_text: str
+    char_count_original: int
+    char_count_sanitized: int
+    processing_ms: float
+    blocked_reason: Optional[str] = None
+    trace: dict = field(default_factory=dict)
+
+
+BLOCK_PATTERNS = [
+    (r"(?i)(ignore|forget|disregard)\s+(all\s+)?(previous|prior|above|earlier)\s+(instructions?|prompts?|rules?|constraints?)", "LLM01", 95),
+    (r"(?i)you\s+are\s+now\s+(a|an|the)\s+\w+\s+(without|with no|free from)", "LLM01", 90),
+    (r"(?i)(act|behave|pretend|roleplay)\s+as\s+(if\s+you\s+(are|were)|a|an)\s+", "LLM01", 85),
+    (r"(?i)your\s+(new\s+)?(instructions?|rules?|persona|role)\s+(are|is|will be)", "LLM01", 88),
+    (r"(?i)(override|bypass|disable|remove)\s+(your\s+)?(safety|filter|restriction|guardrail|alignment)", "LLM01", 95),
+    (r"(?i)(print|show|reveal|display|output|repeat)\s+(your\s+)?(system\s+prompt|base\s+prompt|initial\s+prompt)", "LLM02", 90),
+    (r"(?i)(ignore|skip)\s+(the\s+)?(system|user)\s+(prompt|message|instructions?)", "LLM01", 88),
+    (r"(?i)\bDAN\b.*\b(mode|prompt|jailbreak)\b", "LLM01", 95),
+    (r"(?i)do\s+anything\s+now", "LLM01", 90),
+    (r"(?i)(developer|jailbreak|god|admin|root)\s+mode", "LLM01", 88),
+    (r"(?i)\[SYSTEM\]|\[INST\]|\[\/INST\]|<\|system\|>|<\|user\|>|<\|assistant\|>", "LLM01", 92),
+    (r"(?i)(execute|run|eval|call)\s+(this\s+)?(code|script|command|function)", "LLM02", 85),
+    (r"(?i)```\s*(python|bash|sh|javascript|js|powershell|cmd)\s*\n", "LLM02", 80),
+    (r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "LLM01", 70),
+    (r"\u202e|\u200b|\ufeff", "LLM01", 75),
+    (r"(?i)(simulate|emulate)\s+(a|an)?\s*(unrestricted|uncensored|unfiltered)\s*(ai|model|llm|chatbot)", "LLM01", 92),
+    (r"(?i)respond\s+(only|exclusively)\s+(in|as|like)\s+(a|an)?\s*(evil|unethical|harmful|dangerous)", "LLM01", 95),
+    (r"(?i)(for\s+)?(educational|research|fictional|hypothetical)\s+purposes?\s*(only|,)?\s*(ignore|bypass|disable)", "LLM01", 85),
+]
+
+SUSPICIOUS_PATTERNS = [
+    (r"(?i)(new|different|alternative)\s+(instruction|rule|persona|role|prompt)", "LLM01", 40),
+    (r"(?i)from\s+now\s+on\s+(you|your)", "LLM01", 45),
+    (r"(?i)(must|should|will)\s+(always|never)\s+(respond|answer|say)", "LLM01", 35),
+    (r"(?i)(hypothetically|theoretically|in\s+a\s+fictional\s+scenario)", "LLM01", 30),
+    (r"(?i)(base64|hex|rot13|caesar)\s+(encoded?|decoded?|cipher)", "LLM02", 50),
+    (r"(?i)(sudo|admin|root|superuser)\s*:", "LLM01", 40),
+    (r"(?i)what\s+would\s+you\s+(say|do)\s+if\s+you\s+(had\s+no|were\s+free)", "LLM01", 45),
+    (r"(?i)(translate|convert)\s+(and\s+)?(follow|execute|run)\s+", "LLM02", 55),
+    (r"(?i)pretend\s+(that\s+)?(you|there)\s+(are|is|have)\s+no", "LLM01", 50),
+    (r"(?i)imagine\s+(you\s+are|being)\s+(a|an)\s+\w+\s+(without|with no)", "LLM01", 45),
+]
+
+LIMITS = {
+    "baixa":   {"max_chars": 8000, "max_lines": 100, "block_threshold": 85},
+    "média":   {"max_chars": 4000, "max_lines": 50,  "block_threshold": 75},
+    "alta":    {"max_chars": 2000, "max_lines": 30,  "block_threshold": 65},
+    "máxima":  {"max_chars": 1000, "max_lines": 20,  "block_threshold": 50},
+}
+
+OWASP_MAP = {
+    "LLM01": "Prompt Injection",
+    "LLM02": "Insecure Output Handling",
+    "LLM03": "Training Data Poisoning",
+    "LLM06": "Sensitive Information Disclosure",
+}
+
+
+class PromptInjectionDetector:
+
+    def __init__(self):
+        self._block = [(re.compile(p), cat, score) for p, cat, score in BLOCK_PATTERNS]
+        self._suspicious = [(re.compile(p), cat, score) for p, cat, score in SUSPICIOUS_PATTERNS]
+
+    def get_owasp_category(self, threat: str) -> str:
+        for cat_id, cat_name in OWASP_MAP.items():
+            if cat_id in threat:
+                return f"{cat_id}: {cat_name}"
+        return "OWASP LLM01: Prompt Injection"
+
+    def analyze(self, text: str, max_chars: int = 4000, sensitivity: str = "alta") -> DetectionResult:
+        return self._run(text, max_chars, sensitivity, trace_mode=False)
+
+    def analyze_with_trace(self, text: str, max_chars: int = 4000, sensitivity: str = "alta") -> DetectionResult:
+        return self._run(text, max_chars, sensitivity, trace_mode=True)
+
+    def _run(self, text: str, max_chars: int, sensitivity: str, trace_mode: bool) -> DetectionResult:
+        t0 = time.perf_counter()
+        limits = LIMITS.get(sensitivity, LIMITS["alta"])
+        effective_max = min(max_chars, limits["max_chars"])
+
+        trace_id = hashlib.sha256(f"{text}{time.time()}".encode()).hexdigest()[:16]
+        threats = []
+        mods = []
+        trace = {}
+        risk_score = 0
+
+        # ── step 1: unicode ─────────────────────────────────────────────────
+        t1 = time.perf_counter()
+        normalized = unicodedata.normalize("NFKC", text)
+        changed = normalized != text
+        if changed:
+            mods.append("unicode_normalized")
+        text = normalized
+        if trace_mode:
+            trace["unicode"] = {"status": "flagged" if changed else "pass", "detail": "Normalized" if changed else "OK", "ms": round((time.perf_counter()-t1)*1000, 2)}
+
+        # ── step 2: control chars ────────────────────────────────────────────
+        t1 = time.perf_counter()
+        cleaned = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f\u200b\u202e\ufeff]", "", text)
+        changed = cleaned != text
+        if changed:
+            mods.append("control_chars_removed")
+            threats.append("control_characters [LLM01]")
+            risk_score = max(risk_score, 30)
+        text = cleaned
+        if trace_mode:
+            trace["control_chars"] = {"status": "flagged" if changed else "pass", "detail": f"Removed {len(normalized)-len(cleaned)} chars" if changed else "OK", "ms": round((time.perf_counter()-t1)*1000, 2)}
+
+        # ── step 3: size ─────────────────────────────────────────────────────
+        t1 = time.perf_counter()
+        size_threats = []
+        if len(text) > effective_max:
+            text = text[:effective_max]
+            mods.append(f"truncated_to_{effective_max}_chars")
+            size_threats.append("oversized_input")
+            risk_score = max(risk_score, 20)
+        lines = text.split("\n")
+        if len(lines) > limits["max_lines"]:
+            text = "\n".join(lines[:limits["max_lines"]])
+            mods.append(f"truncated_to_{limits['max_lines']}_lines")
+            size_threats.append("too_many_lines")
+        rep_match = re.search(r"(.)\1{99,}", text)
+        if rep_match:
+            text = re.sub(r"(.)\1{99,}", lambda m: m.group(1)*100, text)
+            mods.append("repetition_collapsed")
+            size_threats.append("excessive_repetition")
+            risk_score = max(risk_score, 25)
+        threats.extend(size_threats)
+        if trace_mode:
+            trace["size"] = {"status": "flagged" if size_threats else "pass", "detail": ", ".join(size_threats) if size_threats else "Within limits", "ms": round((time.perf_counter()-t1)*1000, 2)}
+
+        # ── step 4: pattern matching ─────────────────────────────────────────
+        t1 = time.perf_counter()
+        blocked_match = None
+        for pattern, cat, score in self._block:
+            m = pattern.search(text)
+            if m:
+                if score >= limits["block_threshold"]:
+                    blocked_match = (m.group(0)[:80], cat, score)
+                    break
+                else:
+                    threats.append(f"near_block_pattern [{cat}]: {m.group(0)[:40]}")
+                    risk_score = max(risk_score, score - 10)
+
+        if blocked_match:
+            elapsed = (time.perf_counter() - t0) * 1000
+            if trace_mode:
+                trace["patterns"] = {"status": "blocked", "detail": f"Matched: {blocked_match[0]}", "ms": round((time.perf_counter()-t1)*1000, 2)}
+                trace["semantic"] = {"status": "skipped", "detail": "Pipeline aborted", "ms": 0}
+                trace["risk"] = {"status": "blocked", "detail": f"Score: 100", "ms": 0}
+                trace["output"] = {"status": "skipped", "detail": "Pipeline aborted", "ms": 0}
+            return DetectionResult(
+                trace_id=trace_id,
+                threat_level=ThreatLevel.BLOCKED,
+                risk_score=100,
+                threats_found=threats + [f"block_pattern [{blocked_match[1]}]: {blocked_match[0]}"],
+                modifications=mods,
+                sanitized_text="",
+                char_count_original=len(normalized),
+                char_count_sanitized=0,
+                processing_ms=round(elapsed, 2),
+                blocked_reason=f"Injection pattern detected: '{blocked_match[0]}'",
+                trace=trace,
+            )
+
+        susp_found = []
+        for pattern, cat, score in self._suspicious:
+            m = pattern.search(text)
+            if m:
+                susp_found.append(f"suspicious_pattern [{cat}]: {m.group(0)[:40]}")
+                risk_score = max(risk_score, score)
+        threats.extend(susp_found)
+
+        if trace_mode:
+            trace["patterns"] = {"status": "flagged" if susp_found else "pass", "detail": f"{len(susp_found)} suspicious patterns" if susp_found else "No patterns matched", "ms": round((time.perf_counter()-t1)*1000, 2)}
+
+        # ── step 5: semantic score (heurístico sem modelo externo) ────────────
+        t1 = time.perf_counter()
+        semantic_score = self._heuristic_semantic_score(text)
+        risk_score = max(risk_score, semantic_score)
+        if trace_mode:
+            trace["semantic"] = {"status": "flagged" if semantic_score > 30 else "pass", "detail": f"Heuristic score: {semantic_score}", "ms": round((time.perf_counter()-t1)*1000, 2)}
+
+        # ── step 6: risk aggregation ─────────────────────────────────────────
+        t1 = time.perf_counter()
+        if len(threats) > 3:
+            risk_score = min(100, risk_score + 10)
+        if trace_mode:
+            trace["risk"] = {"status": "flagged" if risk_score > 30 else "pass", "detail": f"Final score: {risk_score}", "ms": round((time.perf_counter()-t1)*1000, 2)}
+
+        # ── step 7: output filter ─────────────────────────────────────────────
+        t1 = time.perf_counter()
+        if trace_mode:
+            trace["output"] = {"status": "pass", "detail": "Output filter applied", "ms": round((time.perf_counter()-t1)*1000, 2)}
+
+        threat_level = ThreatLevel.CLEAN if not threats else ThreatLevel.SUSPICIOUS
+        elapsed = (time.perf_counter() - t0) * 1000
+
+        return DetectionResult(
+            trace_id=trace_id,
+            threat_level=threat_level,
+            risk_score=risk_score,
+            threats_found=threats,
+            modifications=mods,
+            sanitized_text=text,
+            char_count_original=len(normalized),
+            char_count_sanitized=len(text),
+            processing_ms=round(elapsed, 2),
+            trace=trace,
+        )
+
+    def _heuristic_semantic_score(self, text: str) -> int:
+        score = 0
+        t = text.lower()
+        injection_keywords = ["instruction", "system prompt", "previous", "ignore", "override", "bypass", "jailbreak", "restriction", "filter", "safety", "pretend", "act as", "role", "now you", "forget"]
+        hits = sum(1 for kw in injection_keywords if kw in t)
+        score += min(hits * 8, 60)
+        if len(re.findall(r"\b(you|your|yourself)\b", t)) > 5:
+            score += 15
+        if "?" not in text and len(text) > 100:
+            score += 10
+        return min(score, 70)