"""Input validation: SSTI, SQL/NoSQL injection probes, ReDoS-safe string handling.""" import re from fastapi import HTTPException # Max sizes — long-password DoS / payload abuse MAX_PASSWORD_CHARS = 128 MAX_PASSWORD_BYTES = 72 # bcrypt limit MAX_TEXT_FIELD = 8_000 MAX_EMAIL_LEN = 254 MAX_NAME_LEN = 120 MAX_QUERY_LEN = 500 # SSTI / template injection probes (Jinja, Twig, ERB, etc.) _SSTI = re.compile( r"(\{\{|\}\}|{%|%}|#\{|<%|<\?|\$\{|\[\[|\]\]|" r"__class__|__mro__|__subclasses__|__globals__|" r"config\.|request\.|self\.)", re.IGNORECASE, ) # Common SQL / NoSQL injection signatures in user text _INJECTION = re.compile( r"(\bUNION\b\s+\bSELECT\b|\bDROP\b\s+\bTABLE\b|\bINSERT\b\s+\bINTO\b|" r"\bOR\b\s+['\"]?\d+['\"]?\s*=\s*['\"]?\d+|" r"\$where|\$gt|\$ne|\$regex|\{\s*\"\$)", re.IGNORECASE, ) # Safe bounded patterns only _ASIN = re.compile(r"^[A-Z0-9]{10}$") def assert_password_safe(password: str) -> None: """Block long-password DoS before bcrypt.""" if not password: raise HTTPException(status_code=400, detail="Password is required") if len(password) < 8: raise HTTPException(status_code=400, detail="Password must be at least 8 characters") if len(password) > MAX_PASSWORD_CHARS: raise HTTPException(status_code=400, detail=f"Password must be at most {MAX_PASSWORD_CHARS} characters") if len(password.encode("utf-8")) > MAX_PASSWORD_BYTES: raise HTTPException(status_code=400, detail="Password is too long for secure hashing") def clamp_str(value: str | None, max_len: int, field: str = "field") -> str: if value is None: return "" if not isinstance(value, str): raise HTTPException(status_code=400, detail=f"Invalid {field}") if len(value) > max_len: raise HTTPException(status_code=400, detail=f"{field} exceeds maximum length ({max_len})") return value def scan_user_text(value: str, field: str = "input") -> str: """Reject SSTI and injection probe strings in free-text fields.""" value = clamp_str(value, MAX_TEXT_FIELD, field) if _SSTI.search(value): raise HTTPException(status_code=400, detail="Invalid characters in request (template injection blocked)") if _INJECTION.search(value): raise HTTPException(status_code=400, detail="Invalid characters in request (injection blocked)") return value def strip_for_regex(text: str, max_len: int = 500) -> str: """ReDoS-safe prep: cap length before regex; strip without nested backtracking.""" if not text: return "" text = text[:max_len] # Simple character removal instead of heavy regex on long strings out = [] for ch in text: if ch in "()[]": continue out.append(ch) return "".join(out) def normalize_asin_safe(asin: str) -> str: cleaned = (asin or "").upper().strip()[:16] if not _ASIN.match(cleaned): raise HTTPException(status_code=400, detail="Invalid ASIN — must be 10 alphanumeric characters") return cleaned