app/guardrails/input.py

import re
import time
from collections import defaultdict, deque

from app.core.config import settings
from app.core.errors import CopilotError


_requests_by_client: dict[str, deque[float]] = defaultdict(deque)


PROMPT_INJECTION_PATTERNS = [
    "ignore previous",
    "ignore the instructions",
    "forget all previous",
    "system prompt",
    "developer message",
    "bypass system",
    "disregard instructions",
    "reveal hidden",
    "you are now an arbitrary",
    "do anything now",
    "ignore all constraints"
]


def validate_query(query: str) -> None:
    if not query.strip():
        raise CopilotError("Query cannot be empty.", status_code=400)
    if len(query) > settings.MAX_QUERY_LENGTH:
        raise CopilotError("Query is too long.", status_code=400)
    normalized = query.lower()
    for pattern in PROMPT_INJECTION_PATTERNS:
        if pattern in normalized:
            raise CopilotError("Prompt injection attempt detected.", status_code=400)


def enforce_rate_limit(client_id: str) -> None:
    if settings.RATE_LIMIT_PER_MINUTE <= 0:
        return
    now = time.time()
    bucket = _requests_by_client[client_id]
    while bucket and now - bucket[0] > 60:
        bucket.popleft()
    if len(bucket) >= settings.RATE_LIMIT_PER_MINUTE:
        raise CopilotError("Rate limit exceeded.", status_code=429)
    bucket.append(now)


async def async_enforce_rate_limit(client_id: str) -> None:
    if settings.RATE_LIMIT_PER_MINUTE <= 0:
        return
    try:
        from app.core.queue import get_redis_client
        redis = await get_redis_client()
        key = f"rate_limit:{client_id}"
        current = await redis.incr(key)
        if current == 1:
            await redis.expire(key, 60)
        if current > settings.RATE_LIMIT_PER_MINUTE:
            raise CopilotError("Rate limit exceeded.", status_code=429)
    except CopilotError:
        raise
    except Exception:
        enforce_rate_limit(client_id)



def contains_pii(text: str) -> bool:
    email = r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"
    phone = r"\b(?:\+?\d{1,3}[-.\s]?)?(?:\d{10}|\d{3}[-.\s]\d{3}[-.\s]\d{4})\b"
    return bool(re.search(email, text) or re.search(phone, text))