Security handler

README.md

@@ -23,6 +23,7 @@ The Python SDK offers a clean, type-safe API following Python best practices whi
 - **ObjectStore Service**
 - **Secret Resolver**
 - **Telemetry & Observability**
+- **Security Handler**
 - **Data Anonymization Service**
 
 ## Requirements and Setup
@@ -77,6 +78,7 @@ Each module has comprehensive usage guides:
 - [ObjectStore](src/sap_cloud_sdk/objectstore/user-guide.md)
 - [Secret Resolver](src/sap_cloud_sdk/core/secret_resolver/user-guide.md)
 - [Telemetry](src/sap_cloud_sdk/core/telemetry/user-guide.md)
+- [Security Handler](src/sap_cloud_sdk/security_handler/user-guide.md)
 - [Data Anonymization](src/sap_cloud_sdk/core/data_anonymization/user-guide.md)
 
 ## Support, Feedback, Contributing

src/sap_cloud_sdk/security_handler/__init__.py

@@ -0,0 +1,174 @@
+"""security_handler — Generic LLM security utilities for agent pipelines.
+
+Provides a composable pre-LLM gate covering:
+  - Input sanitization (control chars, whitespace, length)
+  - Prompt injection detection (instruction override, persona redefinition,
+    safety bypass, system prompt extraction, templating tokens)
+  - Pluggable security guardrails (forbidden patterns, custom rules)
+
+Typical usage::
+
+    from sap_cloud_sdk.security_handler import SecurityHandler
+
+    handler = SecurityHandler()
+    result = handler.scan(user_input)
+    if result.is_blocked:
+        raise ValueError(result.violations[0].description)
+    process(result.sanitized_text)
+
+With custom config::
+
+    from sap_cloud_sdk.security_handler import SecurityHandler, SecurityConfig, Severity
+
+    handler = SecurityHandler(SecurityConfig(
+        min_blocking_severity=Severity.HIGH,
+        custom_forbidden_patterns=[r"competitor_name"],
+    ))
+"""
+
+import logging
+from dataclasses import dataclass, field
+
+from .guardrails import ForbiddenPatternGuardrail, Guardrail
+from .injection_detector import InjectionDetector, PatternRule
+from .models import ScanResult, Severity, Violation, ViolationType
+from .sanitizer import InputTooLongError, MAX_INPUT_LENGTH, escape_xml_tags, sanitize
+
+logger = logging.getLogger(__name__)
+
+_SEVERITY_ORDER: dict[Severity, int] = {
+    Severity.LOW: 1,
+    Severity.MEDIUM: 2,
+    Severity.HIGH: 3,
+    Severity.CRITICAL: 4,
+}
+
+
+@dataclass
+class SecurityConfig:
+    """Configuration for SecurityHandler.
+
+    Attributes:
+        max_input_length: Hard cap on input length. Inputs exceeding this are
+            blocked immediately without further scanning.
+        min_blocking_severity: Violations at or above this severity cause
+            is_blocked=True. Defaults to MEDIUM.
+        injection_detection: Enable/disable the built-in injection detector.
+        templating_token_detection: Enable/disable ${...}, {{...}}, <<...>> checks.
+        custom_forbidden_patterns: Extra case-insensitive regex patterns to treat
+            as forbidden (evaluated as a ForbiddenPatternGuardrail at HIGH severity).
+        extra_injection_rules: Additional PatternRule instances appended to the
+            built-in injection rule set.
+        custom_guardrails: Fully custom Guardrail implementations.
+    """
+
+    max_input_length: int = MAX_INPUT_LENGTH
+    min_blocking_severity: Severity = Severity.MEDIUM
+    injection_detection: bool = True
+    templating_token_detection: bool = True
+    custom_forbidden_patterns: list[str] = field(default_factory=list)
+    extra_injection_rules: list[PatternRule] = field(default_factory=list)
+    custom_guardrails: list[Guardrail] = field(default_factory=list)
+
+
+class SecurityHandler:
+    """Orchestrates sanitization, injection detection, and guardrail enforcement.
+
+    Designed as a pre-LLM gate: call scan() on every user message before it
+    reaches any prompt template or agent node.
+    """
+
+    def __init__(self, config: SecurityConfig | None = None) -> None:
+        self.config = config or SecurityConfig()
+
+        self._detector = (
+            InjectionDetector(
+                extra_rules=self.config.extra_injection_rules,
+                include_templating_checks=self.config.templating_token_detection,
+            )
+            if self.config.injection_detection
+            else None
+        )
+
+        self._guardrails: list[Guardrail] = []
+        if self.config.custom_forbidden_patterns:
+            self._guardrails.append(
+                ForbiddenPatternGuardrail(self.config.custom_forbidden_patterns)
+            )
+        self._guardrails.extend(self.config.custom_guardrails)
+
+    def scan(self, text: str | None) -> ScanResult:
+        """Sanitize and scan text for security violations.
+
+        Always returns a ScanResult. sanitized_text is safe to embed in a prompt.
+        is_blocked is True when any violation meets or exceeds
+        config.min_blocking_severity.
+        """
+        if not text:
+            return ScanResult(original_text=text or "", sanitized_text=text or "")
+
+        violations: list[Violation] = []
+
+        # 1. Sanitize — strips control chars, normalises whitespace, enforces length.
+        try:
+            sanitized, san_violations = sanitize(text, self.config.max_input_length)
+            violations.extend(san_violations)
+        except InputTooLongError as exc:
+            violations.append(
+                Violation(
+                    type=ViolationType.INPUT_TOO_LONG,
+                    severity=Severity.HIGH,
+                    description=str(exc),
+                )
+            )
+            return ScanResult(
+                original_text=text,
+                sanitized_text=text[: self.config.max_input_length],
+                violations=violations,
+                is_blocked=True,
+            )
+
+        sanitized = sanitized or ""
+
+        # 2. Injection detection.
+        if self._detector and sanitized:
+            violations.extend(self._detector.detect(sanitized))
+
+        # 3. Custom guardrails.
+        for guardrail in self._guardrails:
+            if sanitized:
+                violations.extend(guardrail.check(sanitized))
+
+        # 4. Block decision.
+        threshold = _SEVERITY_ORDER[self.config.min_blocking_severity]
+        is_blocked = any(_SEVERITY_ORDER[v.severity] >= threshold for v in violations)
+
+        if is_blocked:
+            logger.warning(
+                "SecurityHandler blocked input — %d violation(s): %s",
+                len(violations),
+                [v.description for v in violations],
+            )
+
+        return ScanResult(
+            original_text=text,
+            sanitized_text=sanitized,
+            violations=violations,
+            is_blocked=is_blocked,
+        )
+
+
+__all__ = [
+    "SecurityHandler",
+    "SecurityConfig",
+    "ScanResult",
+    "Violation",
+    "Severity",
+    "ViolationType",
+    "Guardrail",
+    "ForbiddenPatternGuardrail",
+    "InjectionDetector",
+    "PatternRule",
+    "InputTooLongError",
+    "escape_xml_tags",
+]

src/sap_cloud_sdk/security_handler/guardrails.py

@@ -0,0 +1,49 @@
+"""Security guardrails — composable, pluggable rule checks.
+
+Extend Guardrail to add domain-specific rules without touching SecurityHandler.
+"""
+
+import re
+from abc import ABC, abstractmethod
+
+from .models import Severity, Violation, ViolationType
+
+
+class Guardrail(ABC):
+    """Pluggable guardrail interface. Implement check() to add custom rules."""
+
+    @abstractmethod
+    def check(self, text: str) -> list[Violation]: ...
+
+
+class ForbiddenPatternGuardrail(Guardrail):
+    """Block input matching any of the provided regex patterns (case-insensitive).
+
+    Useful for domain-specific banned terms, competitor names, or internal
+    code-words that should never appear in user input.
+    """
+
+    def __init__(
+        self,
+        patterns: list[str],
+        severity: Severity = Severity.HIGH,
+        description_prefix: str = "Forbidden pattern matched",
+    ) -> None:
+        self._compiled = [(re.compile(p, re.IGNORECASE), p) for p in patterns]
+        self._severity = severity
+        self._prefix = description_prefix
+
+    def check(self, text: str) -> list[Violation]:
+        violations: list[Violation] = []
+        for pattern, raw in self._compiled:
+            match = pattern.search(text)
+            if match:
+                violations.append(
+                    Violation(
+                        type=ViolationType.FORBIDDEN_PATTERN,
+                        severity=self._severity,
+                        description=f"{self._prefix}: {raw[:80]}",
+                        matched_text=match.group(0)[:100],
+                    )
+                )
+        return violations

src/sap_cloud_sdk/security_handler/injection_detector.py

@@ -0,0 +1,137 @@
+"""Prompt injection detector.
+
+Detects instruction-override, persona-redefinition, safety-bypass, system-prompt
+extraction, and templating-token patterns derived from:
+  - OWASP LLM Top 10 (LLM01 — Prompt Injection)
+  - Common jailbreak taxonomies (DAN, role-play, mode-activation)
+  - UCL security preamble injection-defense requirements
+
+All violations are returned to the caller — blocking decisions live in SecurityHandler.
+"""
+
+import logging
+import re
+from typing import NamedTuple
+
+from .models import Severity, Violation, ViolationType
+
+logger = logging.getLogger(__name__)
+
+
+class PatternRule(NamedTuple):
+    pattern: str
+    severity: Severity
+    description: str
+
+
+# ---------------------------------------------------------------------------
+# Default injection detection rules
+# ---------------------------------------------------------------------------
+
+_DEFAULT_RULES: list[PatternRule] = [
+    # --- Instruction override ---
+    PatternRule(
+        r"ignore\s+((?:previous|all|above|prior)\s+){1,3}(instructions?|prompts?|rules?|constraints?|context)",
+        Severity.CRITICAL,
+        "Instruction override attempt",
+    ),
+    PatternRule(
+        r"(disregard|forget|bypass)\s+(your\s+)?(previous|all|above|prior)?\s*"
+        r"(instructions?|rules?|constraints?|guidelines?)",
+        Severity.CRITICAL,
+        "Instruction override attempt",
+    ),
+    # --- Activation phrases / jailbreak modes ---
+    PatternRule(r"\bjailbreak\b", Severity.HIGH, "Jailbreak keyword"),
+    PatternRule(r"\bDAN\b", Severity.HIGH, "DAN jailbreak reference"),
+    PatternRule(
+        r"(developer|admin|god|unrestricted|unlocked|superuser)\s+"
+        r"(mode|access|prompt|instructions?)",
+        Severity.HIGH,
+        "Privileged mode activation attempt",
+    ),
+    # --- Role / persona redefinition ---
+    PatternRule(
+        r"(act|pretend|imagine|suppose)\s+(you\s+)?(are|you'?re|as|to\s+be)\s+",
+        Severity.HIGH,
+        "Persona redefinition attempt",
+    ),
+    PatternRule(
+        r"you\s+are\s+(now|a\s+new|no\s+longer)\s+",
+        Severity.HIGH,
+        "Identity override attempt",
+    ),
+    PatternRule(r"role[\s-]?play\s+as\s+", Severity.MEDIUM, "Role-play directive"),
+    # --- Constraint / safety bypass ---
+    PatternRule(
+        r"(reset|unlock|remove|disable|override|bypass)\s+(your\s+)?"
+        r"(safety|restrictions?|constraints?|instructions?|rules?|filters?|guardrails?)",
+        Severity.HIGH,
+        "Safety bypass attempt",
+    ),
+    PatternRule(
+        r"(new\s+)?(paradigm|context|persona|directive)\s*(:\s*|is\s+now\b|begins?\b)",
+        Severity.MEDIUM,
+        "Context redefinition attempt",
+    ),
+    # --- System prompt extraction ---
+    PatternRule(
+        r"(reveal|show(?:\s+me)?|print|output|display|repeat|share|tell\s+me|what\s+(?:is|are))\s+"
+        r"(your\s+)?(system\s+)?(prompt|instructions?|rules?|directives?|preamble)",
+        Severity.HIGH,
+        "System prompt extraction attempt",
+    ),
+    # --- Response prefix injection ---
+    PatternRule(
+        r"(start|begin)\s+(your\s+)?(response\s+)?(with|by\s+saying|with\s+the\s+words?)\s+[\"']",
+        Severity.LOW,
+        "Response prefix injection",
+    ),
+]
+
+# Templating / shell-execution tokens (UCL preamble § Injection defenses)
+_TEMPLATING_RULES: list[PatternRule] = [
+    PatternRule(r"\$\{[^}]{0,200}\}", Severity.MEDIUM, "Template token ${...}"),
+    PatternRule(r"\{\{[^}]{0,200}\}\}", Severity.MEDIUM, "Template token {{...}}"),
+    PatternRule(r"<<[^>]{0,200}>>", Severity.MEDIUM, "Template token <<...>>"),
+    PatternRule(r"#!\s*\S+", Severity.MEDIUM, "Shell-command token #!"),
+]
+
+
+class InjectionDetector:
+    """Scan text for prompt injection patterns.
+
+    Returns all matching violations — the caller (SecurityHandler) decides
+    the blocking threshold.
+    """
+
+    def __init__(
+        self,
+        extra_rules: list[PatternRule] | None = None,
+        include_templating_checks: bool = True,
+    ) -> None:
+        rules: list[PatternRule] = list(_DEFAULT_RULES)
+        if include_templating_checks:
+            rules.extend(_TEMPLATING_RULES)
+        if extra_rules:
+            rules.extend(extra_rules)
+        self._compiled = [
+            (re.compile(r.pattern, re.IGNORECASE), r.severity, r.description)
+            for r in rules
+        ]
+
+    def detect(self, text: str) -> list[Violation]:
+        """Return all injection violations found in text. Does not modify text."""
+        violations: list[Violation] = []
+        for pattern, severity, description in self._compiled:
+            match = pattern.search(text)
+            if match:
+                violations.append(
+                    Violation(
+                        type=ViolationType.INJECTION_ATTEMPT,
+                        severity=severity,
+                        description=description,
+                        matched_text=match.group(0)[:100],
+                    )
+                )
+        return violations