| """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) |
| C_COMMENT = re.compile(r"/\*.*?\*/", re.S) |
| |
| |
| |
| LINE_COMMENT = re.compile(r"(?:#|--)[^\r\n]*[\r\n]") |
| |
| |
| _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]") |
| 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#x<LF>select" 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 (<img src="/*" onerror=... |
| x="*/">, 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} |
|
|
