backend/scanners/deserialization_scanner.py

"""
deserialization_scanner.py — Insecure Deserialization Scanner
==============================================================
Expert-grade rewrite (GAP-006 fix):
  1. Scans Set-Cookie headers (original)
  2. Scans POST/PUT request bodies for serialized content
  3. Scans custom headers (X-Java-Serialized-Object, X-Serialized-Object)
  4. Extended signatures: .NET ViewState, Ruby Marshal, Node ND_FUNC
  5. Active probe: sends crafted Java serialized header and checks error signals
  6. Fixes false-positive-prone PHP "O:" pattern (now length-validates)
  7. Java, PHP, Python, .NET, Ruby YAML markers
"""
import base64, urllib.parse, json, re
from scanners.base_scanner import BaseScanner
from utils.anomaly import TimingAnomalyDetector, SizeAnomalyDetector
from utils.evasion import waf_evade

# ── Serialization signatures ──────────────────────────────────────────────
SIGNATURES = {
    "Java (ysoserial / ACED)": [
        (lambda v: isinstance(v, bytes) and v.startswith(b"\xac\xed\x00\x05"), "raw_bytes"),
        (lambda v: isinstance(v, str) and v.startswith("rO0AB"), "base64"),
        (lambda v: isinstance(v, str) and "java.io.Serializable" in v, "raw"),
        (lambda v: isinstance(v, str) and "java.beans" in v and "ObjectInputStream" in v, "raw"),
    ],
    "PHP Object Serialize": [
        (lambda v: isinstance(v, str) and bool(re.match(r'O:\d+:"', v)), "raw"),
        (lambda v: isinstance(v, str) and bool(re.match(r'a:\d+:\{', v)), "raw"),
        (lambda v: isinstance(v, str) and bool(re.match(r'C:\d+:"', v)), "raw"),
        (lambda v: isinstance(v, str) and bool(re.match(r'R:\d+;', v)), "raw"),
    ],
    "Python Pickle": [
        (lambda v: isinstance(v, str) and "c__main__" in v, "raw"),
        (lambda v: isinstance(v, str) and v.startswith("gASV"), "base64"),
        (lambda v: isinstance(v, str) and v.startswith("gAN9cQAutSw="), "base64"),
        (lambda v: isinstance(v, bytes) and v.startswith(b"\x80\x04\x95"), "raw_bytes"),
        (lambda v: isinstance(v, bytes) and v.startswith(b"\x80\x05\x95"), "raw_bytes"),
        (lambda v: isinstance(v, str) and "__reduce__" in v, "raw"),
    ],
    "Node.js (serialize-javascript)": [
        (lambda v: isinstance(v, str) and "_$$ND_FUNC$$_" in v, "raw"),
        (lambda v: isinstance(v, str) and "$$ND_OBJ$$_" in v, "raw"),
    ],
    ".NET ViewState": [
        (lambda v: isinstance(v, str) and v.startswith("/wEy"), "base64"),
        (lambda v: isinstance(v, str) and "__VIEWSTATE" in v and len(v) > 40, "form_field"),
        (lambda v: isinstance(v, str) and v.startswith("/wE"), "base64"),
        (lambda v: isinstance(v, str) and "__EVENTVALIDATION" in v and len(v) > 20, "form_field"),
    ],
    "Ruby Marshal": [
        (lambda v: isinstance(v, bytes) and v.startswith(b"\x04\x08"), "raw_bytes"),
        (lambda v: isinstance(v, str) and v.startswith("BAh"), "base64"),
        (lambda v: isinstance(v, str) and "ruby/object:" in v, "raw"),
    ],
    "Ruby YAML": [
        (lambda v: isinstance(v, str) and "!ruby/object:" in v, "raw"),
        (lambda v: isinstance(v, str) and "!ruby/class:" in v, "raw"),
    ],
}

# Active probe: a truncated Java serialized stream header (safe — won't RCE)
JAVA_PROBE_HEADER = base64.b64encode(b"\xac\xed\x00\x05t\x00\x1dWSS-DESER-SAFE-PROBE").decode()

JAVA_PAYLOADS = [
    (b"\xac\xed\x00\x05t\x00\x1dWSS-DESER-SAFE-PROBE", "application/x-java-serialized-object", "Java ACED magic bytes"),
    (b"\xac\xed\x00\x05sr\x00\x0cwss.Calc\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00", "application/x-java-serialized-object", "Java gadget probe"),
]

PHP_PAYLOADS = [
    (b'O:10:"WssProbe":1:{s:4:"test";s:2:"ok";}', "application/x-www-form-urlencoded", "PHP serialize mark"),
    (b'a:2:{s:4:"test";s:2:"ok";s:4:"waka";s:3:"waka";}', "application/x-www-form-urlencoded", "PHP array serialized"),
    (b'C:10:"WssProbe":8:{testdata}', "application/x-www-form-urlencoded", "PHP custom serialized"),
]

PYTHON_PICKLE_PAYLOADS = [
    (b"\x80\x05\x95\x10\x00\x00\x00\x00\x00\x00\x00\x8c\x06WSS-OK\x94.", "application/octet-stream", "Python pickle (protocol 5)"),
    (b"\x80\x04\x95\x10\x00\x00\x00\x00\x00\x00\x00\x8c\x06WSS-OK\x94.", "application/octet-stream", "Python pickle (protocol 4)"),
    (b"gASVHQAAAAAAAABdlCh9lCiMAXABlIwFdGVzdJRSlJRSlIWUUpQa", "application/python-pickle", "Python pickle base64"),
]

DOTNET_VIEWSTATE_PAYLOADS = [
    (b"/wEyWSS==", "application/x-www-form-urlencoded", ".NET ViewState probe"),
    (b"/wE", "application/x-www-form-urlencoded", ".NET ViewState short"),
]

RUBY_YAML_PAYLOADS = [
    (b"--- !ruby/object:WssProbe\nname: test\n", "application/x-yaml", "Ruby YAML object"),
    (b"--- !ruby/class:WssProbe\n", "application/x-yaml", "Ruby YAML class"),
]

DESER_RESPONSE_SIGNALS = [
    "ClassNotFoundException", "ObjectInputStream", "deserializ",
    "java.io", "InvalidClassException", "readObject",
    "unserialize", "Serializable", "PersistenceException",
    "com.sun.jndi", "log4j", "jndi:", "InvocationTargetException",
]


class DeserializationScanner(BaseScanner):
    SCANNER_NAME = "Insecure Deserialization Scanner"
    _SCANNER_KEY = "deserialization"

    def __init__(self, scan_id, target, domain, **kwargs):
        super().__init__(scan_id, target, domain, **kwargs)

    def run(self) -> list:
        self.log("INFO", f"[Deserialization] Scanning {self.target}...")

        self._timing_detector = TimingAnomalyDetector()

        body, status, headers = self._make_request(self.target, return_response_obj=True)
        if headers:
            self._check_response_headers(headers)

        self._check_url_params()

        self._test_active_post()

        self._test_header_injection()

        self._test_yaml_pickle_probes()

        self._test_expanded_formats()

        self._test_timing_deser()

        if not self.vulns:
            self.log("SUCCESS", "[Deserialization] No insecure deserialization detected.")
        return self.vulns

    # ── 1. Response headers / cookies ────────────────────────────────────
    def _check_response_headers(self, headers):
        cookies = []
        try:
            all_cookies = headers.get_all("Set-Cookie") or []
            cookies.extend(all_cookies)
        except Exception as e:
            self.log("ERROR", f"[Deserialization] Failed to parse cookies: {e}")

        for cookie in cookies:
            parts = cookie.split(";")
            if not parts: continue
            c_val = parts[0].split("=", 1)
            if len(c_val) == 2:
                name, val = c_val[0].strip(), urllib.parse.unquote(c_val[1].strip())
                self._check_value("Cookie", name, val)

    # ── 2. URL params ─────────────────────────────────────────────────────
    def _check_url_params(self):
        parsed = urllib.parse.urlparse(self.target)
        for k, v in urllib.parse.parse_qsl(parsed.query):
            self._check_value("URL param", k, urllib.parse.unquote(v))

    # ── 3. Active POST body probing ───────────────────────────────────────
    def _test_active_post(self):
        probes = JAVA_PAYLOADS + PHP_PAYLOADS + PYTHON_PICKLE_PAYLOADS + DOTNET_VIEWSTATE_PAYLOADS + RUBY_YAML_PAYLOADS

        for payload_bytes, content_type, label in probes:
            body, status = self._make_request(
                self.target, "POST", payload_bytes,
                {"Content-Type": content_type}
            )
            if body and any(sig.lower() in body.lower() for sig in DESER_RESPONSE_SIGNALS):
                self.log("CRITICAL", f"[Deserialization] Active probe triggered deser signal: {label}")
                self.add_vuln(
                    title=f"Active Deserialization Response — {label}",
                    severity="Critical",
                    category="Insecure Deserialization",
                    cvss_score=9.8,
                    cwe_ids=["CWE-502"],
                    owasp_category="A08:2021 – Software and Data Integrity Failures",
                    confidence="High",
                    cve_ids=[],
                    references=["https://owasp.org/www-project-top-ten/2017/A8_2017-Insecure_Deserialization"],
                    description=(
                        f"Sending a crafted `{label}` payload to `{self.target}` triggered "
                        f"deserialization error signals in the response (`{status}`).\n\n"
                        "The server appears to be deserializing the untrusted POST body, "
                        "which can lead to Remote Code Execution via gadget chains."
                    ),
                    remediation=(
                        "1. Never deserialize untrusted data.\n"
                        "2. Use safe formats: JSON, XML with schema validation.\n"
                        "3. Implement deserialization allowlisting (Java: `ObjectInputFilter`).\n"
                        "4. Run deserialization in a sandboxed/restricted ClassLoader.\n"
                        "5. Deploy serialization kill-switches (ysoserial DefensiveObjectInputStream)."
                    ),
                    payload=label,
                    evidence=f"Error signal matched for {label}",
                    request_details=f"POST {self.target} Content-Type: {content_type}",
                    response_details=f"HTTP {status} with error signal in response",
                )
                return

    # ── 4. Header injection probe ─────────────────────────────────────────
    def _test_header_injection(self):
        headers_to_test = {
            "X-Java-Serialized-Object": JAVA_PROBE_HEADER,
            "X-Serialized-Object":      JAVA_PROBE_HEADER,
            "X-ViewState":              "/wEyWSS==",
            "X-Yaml-Object":            base64.b64encode(b"--- !ruby/object:WssProbe\nname: test\n").decode(),
            "X-Pickle-Object":          base64.b64encode(b"\x80\x05\x95\x10\x00\x00\x00\x00\x00\x00\x00\x8c\x06WSS-OK\x94.").decode(),
            "X-PHP-Serialize":          base64.b64encode(b'O:10:"WssProbe":1:{s:4:"test";s:2:"ok";}').decode(),
        }
        for header_name, value in headers_to_test.items():
            for eva_name, eva_value in waf_evade(value):
                body, status = self._make_request(self.target, headers={header_name: eva_value})
                if body and any(sig.lower() in body.lower() for sig in DESER_RESPONSE_SIGNALS):
                    self.add_vuln(
                        title=f"Deserialization via HTTP Header `{header_name}`",
                        severity="Critical",
                        category="Insecure Deserialization",
                        cvss_score=9.8,
                        cwe_ids=["CWE-502"],
                        owasp_category="A08:2021 – Software and Data Integrity Failures",
                        confidence="High",
                        description=(
                            f"Injecting a serialized object via the `{header_name}` header triggered "
                            "deserialization error signals, suggesting the application deserializes "
                            "this header value without validation."
                        ),
                        remediation=(
                            "Reject or ignore custom serialization headers unless explicitly required. "
                            "Validate and type-check all incoming data before deserialization."
                        ),
                        payload=eva_value[:50] + "...",
                    evidence=f"Error signal matched for header {header_name}",
                    request_details=f"GET {self.target} with {header_name} header",
                    response_details=f"HTTP {status} with error signature",
                )
                return

    # ── 5. YAML / Pickle probes ──────────────────────────────────────────
    def _test_yaml_pickle_probes(self):
        probes = [
            (b"--- !ruby/object:WssProbe\nname: test\n",
             "application/x-yaml", "Ruby YAML (header)"),
            (b"\x80\x04\x95\x10\x00\x00\x00\x00\x00\x00\x00\x8c\x06WSS-OK\x94.",
             "application/python-pickle", "Python pickle (header)"),
        ]
        for payload_bytes, content_type, label in probes:
            body, status = self._make_request(
                self.target, headers={"X-Serialized-Payload": base64.b64encode(payload_bytes).decode()}
            )
            if body and any(sig.lower() in body.lower() for sig in DESER_RESPONSE_SIGNALS):
                self.log("CRITICAL", f"[Deserialization] {label} triggered error signal")
                self.add_vuln(
                    title=f"Deserialization Signal via Header — {label}",
                    severity="Critical",
                    category="Insecure Deserialization",
                    cvss_score=9.5,
                    cwe_ids=["CWE-502"],
                    owasp_category="A08:2021 – Software and Data Integrity Failures",
                    confidence="Medium",
                    description=(
                        f"Sending base64-encoded {label} via custom header triggered "
                        "deserialization errors."
                    ),
                    remediation="Restrict custom headers. Validate all incoming data.",
                    payload=label,
                    evidence="Error signal in response",
                    request_details=f"GET {self.target} with X-Serialized-Payload header",
                    response_details=f"HTTP {status}",
                )
                return

    # ── 6. Expanded format probing ────────────────────────────────────────
    def _test_expanded_formats(self):
        all_payloads = JAVA_PAYLOADS + PHP_PAYLOADS + PYTHON_PICKLE_PAYLOADS + DOTNET_VIEWSTATE_PAYLOADS + RUBY_YAML_PAYLOADS
        for payload_bytes, content_type, label in all_payloads:
            body, status = self._make_request(
                self.target, "POST", payload_bytes,
                {"Content-Type": content_type}
            )
            if body and any(sig.lower() in body.lower() for sig in DESER_RESPONSE_SIGNALS):
                self.add_vuln(
                    title=f"Deserialization via POST — {label}",
                    severity="Critical",
                    category="Insecure Deserialization",
                    cvss_score=9.8,
                    cwe_ids=["CWE-502"],
                    owasp_category="A08:2021 – Software and Data Integrity Failures",
                    confidence="High",
                    description=f"Sending `{label}` payload to {self.target} triggered deserialization error signals.",
                    remediation="Never deserialize untrusted data. Use JSON/XML with schema validation.",
                    payload=label,
                    evidence=f"Error signal matched for {label}",
                    request_details=f"POST {self.target} Content-Type: {content_type}",
                    response_details=f"HTTP {status} with error signal",
                )
                return

    # ── 7. Timing-based deserialization detection ─────────────────────────
    def _test_timing_deser(self):
        self._timing_detector.build_baseline(lambda u, m, d, h, t: self._make_request(u, m, d, h, t), self.target, n=5, headers={})
        for payload_bytes, content_type, label in JAVA_PAYLOADS[:2]:
            _, _, elapsed = self._make_timed_request(
                self.target, "POST", payload_bytes,
                {"Content-Type": content_type}, timeout=15
            )
            if self._timing_detector.test_payload(f"deser_time_{label}", elapsed, label, z_threshold=2.5):
                self.add_vuln(
                    title=f"Possible Timing-Based Deserialization — {label}",
                    severity="High",
                    category="Insecure Deserialization",
                    cvss_score=7.5,
                    cwe_ids=["CWE-502"],
                    owasp_category="A08:2021 – Software and Data Integrity Failures",
                    confidence="Low",
                    description=f"Sending `{label}` payload produced anomalous response time ({elapsed:.1f}s vs baseline {self._timing_detector.mean:.1f}s), suggesting deserialization processing.",
                    remediation="Never deserialize untrusted data. Implement ObjectInputFilter.",
                    payload=label,
                    evidence=f"Timing: {elapsed:.1f}s vs baseline {self._timing_detector.mean:.1f}s",
                    request_details=f"POST {self.target} Content-Type: {content_type}",
                    response_details=f"Response time: {elapsed:.2f}s",
                )
                return

    # ── Signature checker ─────────────────────────────────────────────────
    def _check_value(self, source: str, key: str, value: str):
        self._match_signatures(source, key, value)

        try:
            padded = value + "=" * ((4 - len(value) % 4) % 4)
            decoded_bytes = base64.b64decode(padded)
            decoded_str = decoded_bytes.decode("utf-8", errors="ignore")
            self._match_signatures(source, key, decoded_str, encoding="base64")
            for tech, checks in SIGNATURES.items():
                for fn, enc in checks:
                    if enc == "raw_bytes" and fn(decoded_bytes):
                        self._report(source, key, tech, "base64->bytes")
        except Exception as e:
            self.log("ERROR", f"[Deserialization] Base64 decode error: {e}")

    def _match_signatures(self, source, key, content, encoding="raw"):
        for tech, checks in SIGNATURES.items():
            for fn, enc in checks:
                if enc in ("raw", "form_field", "base64") and fn(content):
                    if len(content) > 8:
                        self._report(source, key, tech, encoding)
                        return

    def _report(self, source, key, tech, encoding):
        if any(v["title"].startswith(f"Insecure Deserialization ({tech})") for v in self.vulns):
            return
        self.add_vuln(
            title=f"Insecure Deserialization ({tech}) in {source}",
            severity="Critical",
            category="Insecure Deserialization",
            cvss_score=9.8,
            cwe_ids=["CWE-502"],
            owasp_category="A08:2021 – Software and Data Integrity Failures",
            confidence="High",
            cve_ids=["CVE-2015-4852"] if "Java" in tech else [],
            references=["https://cheatsheetseries.owasp.org/cheatsheets/Deserialization_Cheat_Sheet.html"],
            description=(
                f"Detected a **{tech}** serialized object in `{source}` -> `{key}` ({encoding}).\n\n"
                "Untrusted deserialization typically leads to **Remote Code Execution (RCE)** "
                "via gadget chains (e.g., ysoserial, phpggc)."
            ),
            remediation=(
                "1. Use JSON/XML with schema validation instead of native serialization.\n"
                "2. Implement type allowlisting before deserialization.\n"
                "3. Java: use `ObjectInputFilter` (JDK 9+) to block unexpected classes.\n"
                "4. PHP: set `allowed_classes` in `unserialize()` calls.\n"
                "5. Monitor for deserialization exceptions in production logs."
            ),
            payload=key,
            evidence=f"Serialization format: {tech}",
            request_details=f"Source: {source}",
            response_details=f"Encoded as: {encoding}",
        )
        self.log("CRITICAL", f"[Deserialization] {tech} signature in {source}:{key}")