add more security analisys

README.md

@@ -8,7 +8,7 @@ Python module for structured prompt orchestration with:
 - configurable summary LLM with provider selection
 - TTL cache backends
 - optional RAG providers
-- safety checks (injection + contradiction heuristics)
+- safety checks (config-driven grouped threats, weighted groups, bilingual patterns, contradiction pairs)
 - prompt efficiency analyzer
 - token counting with tiktoken
 - centralized mutable config (Pydantic)
@@ -115,6 +115,29 @@ Use it as a panel/query blueprint in SigNoz to create a dashboard for prompt bui
 - `ModuleConfig`: full module config in one object
 - `ConfigStore`: mutable config holder (`get`, `set_config`, `as_dict`)
 
+## Safety Engine
+
+The safety layer is configured from [prompt_orchestrator/safety/threats.json](prompt_orchestrator/safety/threats.json). The catalog is grouped by threat family, and each family has its own weight so the final severity is still computed by the maximum matched threat score.
+
+What changed:
+
+- threat families are defined in `threats.json` and loaded at runtime
+- regular lexical rules live under `patterns`
+- contradiction rules live under `contradictions` and are matched as pairs
+- each family can include English and Russian analogs for the same threat family
+- duplicate patterns were removed from the catalog
+- each matched rule keeps its threat code in the report
+
+`SafetyReport` now includes:
+
+- `issues`: flat list of matched safety issues
+- `threat_groups`: grouped report by threat family
+- `severity`: overall severity (`none`, `low`, `medium`, `high`)
+- `threat_score`: weighted maximum score used for the final severity
+- `sanitized_prompt`: optional rewritten prompt when auto rewrite is enabled
+
+Each grouped report includes the threat family name, the number of matches, the matched codes, and the family weight. Use `result.safety.grouped_summary` or `result.safety.model_dump()` to inspect the grouped output.
+
 ### OrchestratorSettings.debug_mode
 
 By default, section headers (`=== STATIC PART (CACHE-FRIENDLY) ===`, etc.) are **excluded** from the final prompt sent to LLMs to save tokens. 

prompt_orchestrator/safety/__init__.py

@@ -1,4 +1,4 @@
 from .engine import PromptSafetyEngine
-from .report import SafetyIssue, SafetyReport
+from .report import SafetyIssue, SafetyReport, SafetyThreatGroupReport
 
-__all__ = ["PromptSafetyEngine", "SafetyIssue", "SafetyReport"]
+__all__ = ["PromptSafetyEngine", "SafetyIssue", "SafetyReport", "SafetyThreatGroupReport"]

prompt_orchestrator/safety/engine.py

@@ -1,49 +1,233 @@
 from __future__ import annotations
 
+import json
 import re
+import unicodedata
+from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path
+from typing import Literal
 
-from .report import SafetyIssue, SafetyReport
+from .report import SafetyIssue, SafetyReport, SafetyThreatGroupReport
+
+
+Severity = Literal["none", "low", "medium", "high"]
+
+
+@dataclass(frozen=True, slots=True)
+class ThreatRule:
+    code: str
+    pattern: str | None = None
+    contradiction: tuple[str, str] | None = None
+    compiled_pattern: re.Pattern[str] | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class ThreatGroup:
+    key: str
+    description: str
+    risk_level: Severity
+    weight: float
+    rules: tuple[ThreatRule, ...]
+
+
+def _severity_from_score(score: float) -> Severity:
+    if score >= 0.85:
+        return "high"
+    if score >= 0.5:
+        return "medium"
+    if score > 0:
+        return "low"
+    return "none"
+
+
+def _weight_from_risk_level(risk_level: str) -> float:
+    mapping = {
+        "high": 1.0,
+        "medium": 0.65,
+        "low": 0.35,
+        "none": 0.0,
+    }
+    return mapping.get(risk_level, 0.0)
+
+
+@lru_cache(maxsize=1)
+def _load_threat_groups() -> tuple[ThreatGroup, ...]:
+    threats_path = Path(__file__).with_name("threats.json")
+    raw_data = json.loads(threats_path.read_text(encoding="utf-8"))
+
+    seen_rules: set[tuple[str, str]] = set()
+    groups: list[ThreatGroup] = []
+
+    for key, payload in raw_data.items():
+        description = str(payload.get("description", ""))
+        risk_level = str(payload.get("risk_level", "none"))
+        weight = float(payload.get("weight", _weight_from_risk_level(risk_level)))
+        rules: list[ThreatRule] = []
+
+        for entry in payload.get("patterns", payload.get("data", [])):
+            if "pattern" in entry:
+                pattern = str(entry["pattern"])
+                dedupe_key = ("pattern", pattern)
+                if dedupe_key in seen_rules:
+                    continue
+                seen_rules.add(dedupe_key)
+                rules.append(
+                    ThreatRule(
+                        code=str(entry["code"]),
+                        pattern=pattern,
+                        compiled_pattern=re.compile(pattern, re.IGNORECASE),
+                    )
+                )
+        for entry in payload.get("contradictions", []):
+            contradiction_values = tuple(str(value) for value in entry["contradiction"])
+            dedupe_key = ("contradiction", "||".join(contradiction_values))
+            if dedupe_key in seen_rules:
+                continue
+            seen_rules.add(dedupe_key)
+            if len(contradiction_values) != 2:
+                continue
+            rules.append(
+                ThreatRule(
+                    code=str(entry["code"]),
+                    contradiction=(contradiction_values[0], contradiction_values[1]),
+                )
+            )
+
+        if not rules and "data" in payload:
+            for entry in payload.get("data", []):
+                if "pattern" in entry:
+                    pattern = str(entry["pattern"])
+                    dedupe_key = ("pattern", pattern)
+                    if dedupe_key in seen_rules:
+                        continue
+                    seen_rules.add(dedupe_key)
+                    rules.append(
+                        ThreatRule(
+                            code=str(entry["code"]),
+                            pattern=pattern,
+                            compiled_pattern=re.compile(pattern, re.IGNORECASE),
+                        )
+                    )
+                elif "contradiction" in entry:
+                    contradiction_values = tuple(str(value) for value in entry["contradiction"])
+                    dedupe_key = ("contradiction", "||".join(contradiction_values))
+                    if dedupe_key in seen_rules:
+                        continue
+                    seen_rules.add(dedupe_key)
+                    if len(contradiction_values) != 2:
+                        continue
+                    rules.append(
+                        ThreatRule(
+                            code=str(entry["code"]),
+                            contradiction=(contradiction_values[0], contradiction_values[1]),
+                        )
+                    )
+
+        groups.append(
+            ThreatGroup(
+                key=key,
+                description=description,
+                risk_level=risk_level if risk_level in {"none", "low", "medium", "high"} else "none",
+                weight=weight,
+                rules=tuple(rules),
+            )
+        )
+
+    return tuple(groups)
 
 
 class PromptSafetyEngine:
-    INJECTION_PATTERNS = [
-        re.compile(r"ignore\s+previous\s+instructions", re.IGNORECASE),
-        re.compile(r"(reveal|print|show)\s+(the\s+)?system\s+prompt", re.IGNORECASE),
-        re.compile(r"(reveal|print|show)\s+(the\s+)?developer\s+message", re.IGNORECASE),
-        re.compile(r"reveal\s+hidden", re.IGNORECASE),
-    ]
+    def __init__(self) -> None:
+        self._threat_groups = _load_threat_groups()
+
+    def _too_many_new_lines(self, prompt: str) -> bool:
+        return prompt.count("\n") > 300
+
+    def _prompt_too_long(self, prompt: str) -> bool:
+        return len(prompt) > 15000
+
+    def analize(self, prompt: str) -> SafetyReport:
+        return self.analyze(prompt)
 
     def analyze(self, prompt: str) -> SafetyReport:
+        prompt = unicodedata.normalize("NFKC", prompt)
+        prompt_casefold = prompt.casefold()
+
         issues: list[SafetyIssue] = []
+        threat_groups: list[SafetyThreatGroupReport] = []
+        highest_score = 0.0
 
-        for pattern in self.INJECTION_PATTERNS:
-            if pattern.search(prompt):
-                issues.append(
-                    SafetyIssue(
-                        code="prompt_injection",
-                        message=f"Potential injection marker detected: {pattern.pattern}",
-                        severity="high",
-                    )
+        if self._prompt_too_long(prompt):
+            score = 0.5
+            issues.append(
+                SafetyIssue(
+                    code="prompt_too_long",
+                    message=f"Prompt length {len(prompt)} exceeds safe threshold.",
+                    severity="medium",
+                    weight=score,
                 )
+            )
+            highest_score = max(highest_score, score)
 
-        if "always answer in json" in prompt.lower() and "never use json" in prompt.lower():
+        if self._too_many_new_lines(prompt):
+            score = 0.5
             issues.append(
                 SafetyIssue(
-                    code="contradiction",
-                    message="Conflicting output constraints found.",
+                    code="too_many_newlines",
+                    message=f"Prompt contains {prompt.count(chr(10))} newlines, which may indicate an attempt to obfuscate content.",
                     severity="medium",
+                    weight=score,
                 )
             )
+            highest_score = max(highest_score, score)
+
+        for group in self._threat_groups:
+            findings: list[SafetyIssue] = []
+            for rule in group.rules:
+                matched = False
+                if rule.compiled_pattern is not None and rule.compiled_pattern.search(prompt):
+                    matched = True
+                elif rule.contradiction is not None:
+                    left = unicodedata.normalize("NFKC", rule.contradiction[0]).casefold()
+                    right = unicodedata.normalize("NFKC", rule.contradiction[1]).casefold()
+                    matched = left in prompt_casefold and right in prompt_casefold
+
+                if not matched:
+                    continue
+
+                severity = _severity_from_score(group.weight)
+                issue = SafetyIssue(
+                    code=rule.code,
+                    message=f"[{group.key}] matched threat code {rule.code}",
+                    severity=severity,
+                    group=group.key,
+                    pattern=rule.pattern if rule.pattern is not None else " / ".join(rule.contradiction or ()),
+                    weight=group.weight,
+                )
+                issues.append(issue)
+                findings.append(issue)
+                highest_score = max(highest_score, group.weight)
+
+            if findings:
+                threat_groups.append(
+                    SafetyThreatGroupReport(
+                        name=group.key,
+                        description=group.description,
+                        risk_level=group.risk_level,
+                        weight=group.weight,
+                        issues=findings,
+                    )
+                )
 
-        severity = "none"
-        if any(i.severity == "high" for i in issues):
-            severity = "high"
-        elif any(i.severity == "medium" for i in issues):
-            severity = "medium"
-        elif issues:
-            severity = "low"
+        severity = _severity_from_score(highest_score)
 
-        return SafetyReport(issues=issues, severity=severity)
+        return SafetyReport(
+            issues=issues,
+            threat_groups=threat_groups,
+            severity=severity,
+            threat_score=highest_score,
+        )
 
     def sanitize(self, prompt: str) -> str:
         sanitized = prompt
@@ -53,6 +237,18 @@ def sanitize(self, prompt: str) -> str:
             sanitized,
             flags=re.IGNORECASE,
         )
+        sanitized = re.sub(
+            r"ignore\s+all\s+rules",
+            "[REMOVED_INJECTION_PATTERN]",
+            sanitized,
+            flags=re.IGNORECASE,
+        )
+        sanitized = re.sub(
+            r"from\s+now\s+on",
+            "[REMOVED_INJECTION_PATTERN]",
+            sanitized,
+            flags=re.IGNORECASE,
+        )
         sanitized = re.sub(
             r"reveal\s+hidden",
             "[REMOVED_SENSITIVE_REQUEST]",

prompt_orchestrator/safety/report.py

@@ -5,17 +5,52 @@
 from pydantic import BaseModel, Field
 
 
+Severity = Literal["none", "low", "medium", "high"]
+
+
 class SafetyIssue(BaseModel):
     code: str
     message: str
-    severity: Literal["low", "medium", "high"] = "low"
+    severity: Severity = "none"
+    group: str | None = None
+    pattern: str | None = None
+    weight: float | None = None
+
+
+class SafetyThreatGroupReport(BaseModel):
+    name: str
+    description: str
+    risk_level: Severity = "none"
+    weight: float = 0.0
+    issues: list[SafetyIssue] = Field(default_factory=list)
+
+    @property
+    def count(self) -> int:
+        return len(self.issues)
+
+    @property
+    def codes(self) -> list[str]:
+        return [issue.code for issue in self.issues]
 
 
 class SafetyReport(BaseModel):
     issues: list[SafetyIssue] = Field(default_factory=list)
-    severity: Literal["none", "low", "medium", "high"] = "none"
+    threat_groups: list[SafetyThreatGroupReport] = Field(default_factory=list)
+    severity: Severity = "none"
+    threat_score: float = 0.0
     sanitized_prompt: str | None = None
 
     @property
     def is_safe(self) -> bool:
         return self.severity in {"none", "low"}
+
+    @property
+    def grouped_summary(self) -> str:
+        if not self.threat_groups:
+            return ""
+
+        lines: list[str] = []
+        for index, group in enumerate(self.threat_groups, start=1):
+            codes = ", ".join(group.codes) if group.codes else "None"
+            lines.append(f"{index}. {group.name}: {group.count} threat(s), codes: {codes}")
+        return "\n".join(lines)