app/services/guardrails.py

"""Author RAG Chatbot SaaS — Input Sanitizer & Boundary Enforcer.

Three layers of protection on every chat turn:
1. Input sanitization — strip control chars, delimiter injection, obfuscation.
2. Input boundary check — jailbreak / piracy / off-topic patterns (regex + heuristics).
3. Output safety check — block system-prompt leaks, identity reveals, instruction echo.

RULE: Sanitize BEFORE any LLM call.
RULE: check_boundary runs on EVERY user message — never skip.
RULE: is_response_safe runs on EVERY generated response — never skip.
"""

from __future__ import annotations

import re
import unicodedata
import structlog

logger = structlog.get_logger(__name__)

_MAX_INPUT_CHARS = 2000

# ─── Characters & delimiters used in prompt-injection attacks ─────────────────

_CONTROL_CHAR_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
_ZERO_WIDTH_RE = re.compile(r"[\u200b-\u200f\u2028-\u202f\ufeff]")
_DELIMITER_INJECTION_RE = re.compile(
    r"<\|?(?:system|im_start|im_end|assistant|user|endoftext)\|?>"
    r"|\[INST\]|\[/INST\]|<<SYS>>|<</SYS>>"
    r"|###\s*(?:system|instruction|human|assistant)\b",
    re.IGNORECASE,
)

# ─── Jailbreak / prompt-injection patterns ───────────────────────────────────

_JAILBREAK_PATTERNS = [
    r"ignore\s+(all\s+)?(previous|prior|above|earlier)\s+(instructions?|rules?|prompts?|guidelines?)",
    r"disregard\s+(all\s+)?(previous|prior|above|your)\s+(instructions?|rules?|prompts?)",
    r"forget\s+(everything|all|your)\s+(you\s+)?(know|learned|instructions?|rules?|guidelines?)",
    r"override\s+(your\s+)?(instructions?|rules?|programming|guidelines?|safety)",
    r"bypass\s+(your\s+)?(restrictions?|filters?|rules?|guardrails?|safety)",
    r"you\s+are\s+now\s+(?!an?\s+(?:book\s+)?advisor)",
    r"from\s+now\s+on\s+you\s+(?:are|will|must|should)",
    r"act\s+as\s+(?:if\s+you\s+(?:are|were)\s+)?(?!a?\s+book)",
    r"pretend\s+(?:you\s+)?(?:are|have|to\s+be)\s+(?!a?\s+book)",
    r"roleplay\s+as",
    r"simulate\s+(?:being|a)\s+(?!book)",
    r"developer\s+mode",
    r"debug\s+mode",
    r"admin\s+mode",
    r"maintenance\s+mode",
    r"do\s+anything\s+now",
    r"\bdan\b",
    r"jailbreak",
    r"disable\s+(?:your\s+)?(?:content\s+)?(?:filter|restriction|limit|guardrail|safety)",
    r"remove\s+(?:your\s+)?(?:restrictions?|filters?|limits?|guardrails?)",
    r"without\s+(?:any\s+)?(?:restrictions?|filters?|limits?|guardrails?)",
    r"no\s+(?:rules?|restrictions?|limits?|guardrails?)\s+(?:apply|mode)",
    r"reveal\s+(?:your\s+)?(?:system\s+)?(?:prompt|instructions?|rules?|guidelines?)",
    r"(?:show|print|repeat|output|display)\s+(?:your\s+)?(?:system\s+)?(?:prompt|instructions?|rules?)",
    r"what\s+(?:are|is|were)\s+your\s+(?:system\s+)?(?:prompt|instructions?|rules?|guidelines?)",
    r"repeat\s+(?:the\s+)?(?:text|words?|message)\s+above",
    r"everything\s+(?:you\s+)?(?:were|are)\s+told",
    r"initiali[sz]ation\s+(?:string|prompt|message)",
    r"hidden\s+(?:prompt|instructions?|rules?)",
    r"internal\s+(?:prompt|instructions?|rules?|config)",
    r"api[_\s-]?key",
    r"secret[_\s-]?key",
    r"openai[_\s-]?key",
    r"(?:i\s+am|i'm)\s+(?:the\s+)?(?:developer|owner|admin|creator|engineer)",
    r"(?:i\s+am|i'm)\s+your\s+(?:developer|creator|owner|admin)",
    r"new\s+(?:instructions?|rules?|prompt)\s*:",
    r"system\s+prompt\s*:",
    r"<\s*system\s*>",
    r"\[system\]",
    r"hypothetically[,]?\s+(?:ignore|disregard|forget|if\s+you\s+(?:could|had))",
    r"for\s+(?:research|testing|educational)\s+purposes[,]?\s+(?:ignore|disregard|forget|bypass)",
]

_PIRACY_PATTERNS = [
    r"free\s+(?:pdf|copy|download|epub|ebook|version)",
    r"pirated?\s+(?:copy|version|pdf|download|link)",
    r"(?:send|give|share)\s+me\s+(?:the\s+)?(?:full|entire|complete)\s+(?:book|text|pdf)",
    r"(?:torrent|epub|mobi)\s+(?:link|download|file)",
    r"download\s+(?:the\s+)?book\s+for\s+free",
    r"read\s+(?:it|the\s+book)\s+for\s+free",
    r"without\s+(?:paying|buying|purchasing)",
]

_OFF_TOPIC_KEYWORDS = [
    "stock market", "cryptocurrency", "medical advice",
    "legal advice", "write malware", "write a virus", "exploit code",
    "porn", "adult content", "gambling",
]

# R-172: Competitor mentions — decline without engaging
_COMPETITOR_PATTERNS = [
    r"\bchatgpt\b",
    r"\bclaude\b",
    r"\bgemini\b",
    r"\bcompetitor\b",
    r"\balternative\s+(?:to|for)\s+this\s+book\b",
    r"(?:better|worse)\s+than\s+(?:other|another)\s+(?:book|author)s?\b",
    r"compare.*(?:to|with).*(?:other|another)\s+author",
]

# Book-adjacent phrasing — never treat as off-topic (e.g. purchase objections)
_BOOK_CONTEXT_SIGNALS: tuple[str, ...] = (
    "book", "buy", "purchase", "read", "copy", "author", "story",
    "why should", "convince me", "worth", "no interest", "not interested",
    "don't like", "do not like", "this book", "the book",
)

# Heuristic: multiple injection signals in one message
_INJECTION_SIGNALS = (
    "ignore", "disregard", "forget", "override", "bypass", "jailbreak",
    "system prompt", "your instructions", "developer mode", "act as",
    "pretend you", "reveal your", "no restrictions", "without filters",
)

_COMPILED_JAILBREAK = [re.compile(p, re.IGNORECASE) for p in _JAILBREAK_PATTERNS]
_COMPILED_PIRACY = [re.compile(p, re.IGNORECASE) for p in _PIRACY_PATTERNS]
_COMPILED_COMPETITOR = [re.compile(p, re.IGNORECASE) for p in _COMPETITOR_PATTERNS]

# ─── Output leak / unsafe response patterns ──────────────────────────────────

_RESPONSE_LEAK_PATTERNS = [
    r"system\s+prompt\s*:",
    r"my\s+(?:system\s+)?instructions?\s+(?:are|say|tell|state|instruct)",
    r"i\s+(?:was|am)\s+(?:instructed|programmed|told)\s+to",
    r"here\s+(?:are|is)\s+my\s+(?:instructions?|rules?|guidelines?|system\s+prompt)",
    r"i\s+am\s+(?:gpt|openai|chatgpt|claude|gemini|an?\s+(?:ai|llm|language\s+model))",
    r"as\s+an?\s+(?:ai|language\s+model|llm|chatbot)\b",
    r"openai\s+api",
    r"sk-[a-zA-Z0-9]{10,}",
    r"RETRIEVED\s+CONTEXT:",
    r"Interest score:\s*\d",
    r"RULES YOU MUST FOLLOW",
    r"AUTHOR_CHATBOT_RULES",
    r"MASTER_SYSTEM_PROMPT",
]

_COMPILED_RESPONSE_LEAKS = [re.compile(p, re.IGNORECASE) for p in _RESPONSE_LEAK_PATTERNS]


def sanitize_user_input(text: str, max_length: int = _MAX_INPUT_CHARS) -> str:
    """Normalize and strip dangerous characters from user input.

    Args:
        text: Raw user message.
        max_length: Hard character cap.

    Returns:
        Sanitized text safe for pipeline processing.
    """
    if not text:
        return ""

    text = unicodedata.normalize("NFKC", text)
    text = _ZERO_WIDTH_RE.sub("", text)
    text = _CONTROL_CHAR_RE.sub("", text)
    text = _DELIMITER_INJECTION_RE.sub("", text)
    text = re.sub(r"\s{3,}", "  ", text)
    text = text.strip()

    if len(text) > max_length:
        text = text[:max_length].rsplit(" ", 1)[0].strip() or text[:max_length].strip()
        logger.debug("User input truncated", max_length=max_length)

    return text


def sanitize_history_line(text: str, max_length: int = 300) -> str:
    """Sanitize a single history line before embedding in the system prompt."""
    cleaned = sanitize_user_input(text, max_length=max_length)
    # Neutralize lines that mimic system/assistant role injection in history
    if _DELIMITER_INJECTION_RE.search(cleaned):
        cleaned = _DELIMITER_INJECTION_RE.sub("", cleaned)
    return cleaned


def check_boundary(query: str) -> tuple[str | None, str]:
    """Check if a user query violates content boundaries.

    Returns:
        Tuple of (violation_type | None, details).
    """
    if not query or not query.strip():
        return None, ""

    query_lower = query.lower()

    for pattern in _COMPILED_JAILBREAK:
        if pattern.search(query):
            logger.warning("Jailbreak attempt detected", query=query[:120])
            return "jailbreak_attempt", "Jailbreak pattern matched"

    for pattern in _COMPILED_PIRACY:
        if pattern.search(query):
            logger.warning("Piracy request detected", query=query[:120])
            return "piracy_request", "Piracy pattern matched"

    for pattern in _COMPILED_COMPETITOR:
        if pattern.search(query_lower):
            logger.info("Competitor mention declined", query=query[:120])
            return "competitor_mention", "Competitor reference detected"

    if _count_injection_signals(query_lower) >= 2:
        logger.warning("Heuristic jailbreak detected", query=query[:120])
        return "jailbreak_attempt", "Multiple injection signals"

    for keyword in _OFF_TOPIC_KEYWORDS:
        if keyword in query_lower:
            if any(sig in query_lower for sig in _BOOK_CONTEXT_SIGNALS):
                logger.debug(
                    "Off-topic keyword ignored — book conversation context",
                    keyword=keyword,
                )
                continue
            logger.debug("Off-topic keyword detected", keyword=keyword)
            return "off_topic", f"Off-topic keyword: {keyword}"

    return None, ""


def _count_injection_signals(text_lower: str) -> int:
    return sum(1 for signal in _INJECTION_SIGNALS if signal in text_lower)


def is_response_in_scope(response: str) -> bool:
    """Backward-compatible wrapper — prefer is_response_safe()."""
    safe, _ = is_response_safe(response)
    return safe


def is_response_safe(response: str) -> tuple[bool, str]:
    """Verify a generated response does not leak secrets or break character.

    Returns:
        Tuple of (is_safe, reason).
    """
    if not response or not response.strip():
        return True, ""

    for pattern in _COMPILED_RESPONSE_LEAKS:
        if pattern.search(response):
            logger.warning("Unsafe response blocked", pattern=pattern.pattern)
            return False, "Response leak pattern matched"

    # Block responses that echo obvious instruction blocks
    if response.count("════") >= 2 or response.count("════════") >= 1:
        return False, "Formatted instruction block detected"

    return True, ""


def scrub_unsafe_response(response: str, fallback: str) -> str:
    """Return fallback if response fails safety check, else original."""
    safe, reason = is_response_safe(response)
    if not safe:
        logger.info("Response replaced with safe fallback", reason=reason)
        return fallback
    return response