"""Canonical form of request text: obfuscation layers peeled off until a fixed point. An attacker's encodings (percent, double percent, %uXXXX, HTML entities, \\xHH / \\uXXXX escapes, fullwidth unicode, MySQL /*! */ comments, inline /**/ comments, case games, whitespace tricks) do not change what reaches the application, so the model sees the decoded form. How many layers had to be peeled is itself a signal (`rounds`). Pure python, no allocation-heavy regex backtracking (all patterns are linear); ~5-20 µs for a typical parameter. """ from __future__ import annotations import html import re import unicodedata PCT_RUN = re.compile(r"(?:%[0-9a-fA-F]{2})+") PCT_U = re.compile(r"%u([0-9a-fA-F]{4})") ESC_X = re.compile(r"\\x([0-9a-fA-F]{2})") ESC_U = re.compile(r"\\u([0-9a-fA-F]{4})") MYSQL_EXEC = re.compile(r"/\*!\d*(.*?)\*/", re.S) # /*!50000select*/ -> select (executed by MySQL) C_COMMENT = re.compile(r"/\*.*?\*/", re.S) # inline comments split keywords: un/**/ion # a SQL line comment that a line break ends: '#' (MySQL) or '--' (every dialect) up to the newline. The newline is what # separates the next keyword ("union#xselect"), so the comment must go *before* whitespace folding turns the line # break into a space - after it, the comment would swallow the rest of the value (V-02) LINE_COMMENT = re.compile(r"(?:#|--)[^\r\n]*[\r\n]") # URL schemes with an ASCII tab / LF / CR inside: a browser's URL parser deletes them anywhere in a URL, so # "javascript:" is "javascript:" (V-03). See _join_schemes. _SCHEME_CH = frozenset("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+.-\t\n\r") _ASCII_LETTER = frozenset("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ") WS = re.compile(r"[\s\x00\x0b\x0c]+") CTRL = re.compile(r"[\x01-\x08\x0e-\x1f\x7f]") SURROGATE = re.compile("[\ud800-\udfff]") # lone surrogates (%uD83D, �) are not encodable: never kept MAX_LEN = 8192 def _pct(s: str) -> str: """Percent-decoding that never raises: each run of %XX is decoded as one byte string, so multi-byte UTF-8 (Persian, emoji) stays whole; invalid bytes become U+FFFD.""" if "%" not in s: return s s = PCT_U.sub(lambda m: chr(int(m.group(1), 16)), s) return PCT_RUN.sub(lambda m: bytes.fromhex(m.group(0).replace("%", "")).decode("utf-8", "replace"), s) def decode_layer(s: str, plus_is_space: bool = False) -> str: if plus_is_space and "+" in s: s = s.replace("+", " ") s = _pct(s) if "&" in s: s = html.unescape(s) if "\\" in s: s = ESC_U.sub(lambda m: chr(int(m.group(1), 16)), s) s = ESC_X.sub(lambda m: chr(int(m.group(1), 16)), s) return s def decode_fixpoint(s: str, max_rounds: int = 4, plus_is_space: bool = False) -> tuple[str, int]: """Apply decode_layer until nothing changes. -> (decoded text, layers peeled).""" s = s[:MAX_LEN] rounds = 0 for _ in range(max_rounds): t = decode_layer(s, plus_is_space and rounds == 0) if t == s: break s, rounds = t, rounds + 1 return s, rounds def _join_schemes(s: str) -> str: """Delete ASCII tab / LF / CR inside every URL-scheme-shaped run that ends in ':', as a browser's URL parser does. A run is a maximal stretch of scheme characters (ASCII letters, digits, '+', '.', '-', and tab / LF / CR); it is a scheme when it starts with a letter and the next character is ':'. One linear pass, no look-behind (the Go fast path runs the same scan: go/dibaguard/canon.go joinSchemes).""" out, i, n = [], 0, len(s) while i < n: if s[i] not in _SCHEME_CH: out.append(s[i]) i += 1 continue j = i while j < n and s[j] in _SCHEME_CH: j += 1 run = s[i:j] if s[i] in _ASCII_LETTER and j < n and s[j] == ":" and ("\t" in run or "\n" in run or "\r" in run): run = run.replace("\t", "").replace("\n", "").replace("\r", "") out.append(run) i = j return "".join(out) def _fold(s: str) -> str: s = SURROGATE.sub("�", CTRL.sub("", s)) s = WS.sub(" ", s) return s.lower().strip() def views(s: str) -> tuple[str, str | None]: """Decoded text -> (SQL view, plain view or None). Both: NFKC (fullwidth / compatibility letters), URL schemes without tab / newline, control characters removed, lower case, one space. The SQL view also unwraps MySQL /*! */ comments and folds /* */ and line comments, so "un/**/ion", "union/**/select" and "union#xselect" read the way the database reads them. Folding is right for SQL *outside* a string literal and wrong inside one, and whether a value's quote opens or closes a literal depends on where the application puts the value: "/*' union select password from users-- -*/" is one comment in a numeric context but live code after the literal '/*' in a string context; for HTML and the shell a comment marker is plain text (, echo '/*'). No single fold is right (a quote-aware scanner would only pick the other wrong half), so nothing is deleted unseen: when folding removed anything, the plain view - the same text with every comment kept - is returned as well and judged next to the SQL view (V-01, V-02).""" s = unicodedata.normalize("NFKC", s) if "\t" in s or "\n" in s or "\r" in s: s = _join_schemes(s) q = s if "/*" in q: q = MYSQL_EXEC.sub(lambda m: m.group(1), q) q = C_COMMENT.sub(" ", q) if ("#" in q or "--" in q) and ("\n" in q or "\r" in q): q = LINE_COMMENT.sub(" ", q) a = _fold(q) if q == s: return a, None b = _fold(s) return a, (None if b == a else b) def normalize(s: str) -> str: """Decoded text -> normal form (the SQL view of `views`).""" return views(s)[0] def canon(s: str, plus_is_space: bool = False) -> tuple[str, int]: """Raw request text -> (canonical text, layers of encoding peeled).""" d, rounds = decode_fixpoint(s or "", plus_is_space=plus_is_space) return normalize(d), rounds def canon_views(s: str, plus_is_space: bool = False) -> tuple[str, str | None, int]: """Raw request text -> (canonical SQL view, the plain view when comment folding changed the text else None, layers of encoding peeled).""" d, rounds = decode_fixpoint(s or "", plus_is_space=plus_is_space) a, b = views(d) return a, b, rounds PERSIAN = re.compile(r"[\u0600-\u06ff\ufb50-\ufdff\ufe70-\ufeff\u200c]") def charset_profile(s: str) -> dict: """Cheap per-text counts used as features (on the canonical text).""" n = len(s) or 1 return {"len": len(s), "persian": len(PERSIAN.findall(s)) / n, "non_ascii": sum(1 for c in s if ord(c) > 127) / n, "special": sum(1 for c in s if c in "'\"<>;()`|&$*{}[]\\") / n, "digits": sum(c.isdigit() for c in s) / n}