sqlmap/tests/test_timeless.py

#!/usr/bin/env python

"""
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
See the file 'LICENSE' for copying permission

Coverage for the HTTP/2 timeless-timing oracle in lib/request/timeless.py: the
sequential decision engine that turns response-order votes into bits, the pair
transport's replay-safety and response validation, request-spec conversion, the
sentinel negation, and oracle teardown.

Network-free - a fake _H2Connection stands in for the transport, so every vote
sequence is exactly reproducible.

stdlib unittest only (no pytest / no pip); works on Python 2.7 and 3.x.
"""

import os
import socket
import sys
import threading
import unittest

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from _testutils import bootstrap
bootstrap()

from lib.request import timeless as _timeless
from lib.request.http2 import _UnprocessedStream
from lib.request.timeless import (
    LIGHT_HEAVY,
    LIGHT_HEAVY_COSTS,
    SPRT_CAP_FACTOR,
    TimelessOracle,
    TimelessUnusable,
    _pairOrder,
    _specToReq,
    buildConditionPair,
    calibrate,
    lightHeavyVector,
    negateCondition,
    negatePayload,
    readBit,
)

REQ_COND = {"method": "GET", "path": "/?id=1", "authority": "h"}
REQ_NEG = {"method": "GET", "path": "/?id=2", "authority": "h"}


class FakeConnection(object):
    """Minimal _H2Connection stand-in. `orders` is an iterable of booleans - True means the SECOND request
    of the pair finished last (so exchange_pair reports the first-sent stream as finishing first)."""

    def __init__(self, orders=(), statuses=None, raises=None, host="target"):
        self.host, self.port = host, 443
        self.next_sid = 1
        self.usable = True
        self.closed = 0
        self.sent = []
        self._orders = list(orders)
        self._statuses = list(statuses) if statuses is not None else None
        self._raises = list(raises) if raises is not None else []

    def exchange_pair(self, requests, timeout):
        if self._raises:
            error = self._raises.pop(0)
            if error is not None:
                raise error
        self.sent.append(list(requests))
        lo, hi = self.next_sid, self.next_sid + 2
        self.next_sid += 4
        secondLast = self._orders.pop(0) if self._orders else True
        order = [lo, hi] if secondLast else [hi, lo]
        status = self._statuses.pop(0) if self._statuses else 200
        statuses = status if isinstance(status, tuple) else (status, status)
        return order, {lo: (statuses[0], [], b""), hi: (statuses[1], [], b"")}

    def close(self):
        self.closed += 1
        self.usable = False


def votes(*heavyLast):
    """Build the `orders` sequence for readBit()/calibrate() votes: heavyLast[i] True means the request
    under test (reqCond / reqSlow) finished LAST on vote i. Both alternate the send order, so which slot it
    occupies - and therefore whether 'it finished last' means 'the second-sent stream finished last' -
    flips every vote."""
    return [last if i % 2 else not last for i, last in enumerate(heavyLast)]


class TimelessDecisionTest(unittest.TestCase):
    def test_unanimous_true_and_false(self):
        conn = FakeConnection(votes(*([True] * 20)))
        self.assertTrue(readBit(conn, REQ_COND, REQ_NEG, votes=4))
        conn = FakeConnection(votes(*([False] * 20)))
        self.assertFalse(readBit(conn, REQ_COND, REQ_NEG, votes=4))

    def test_decisions_only_on_balanced_samples(self):
        """Both stream-id orderings must contribute equally, so a bit can never be decided on an odd
        number of votes (which would give one ordering an extra say)."""
        for pattern in ([True] * 20, [False] * 20, [True, False] * 10):
            conn = FakeConnection(votes(*pattern))
            readBit(conn, REQ_COND, REQ_NEG, votes=4)
            self.assertEqual(len(conn.sent) % 2, 0)

    def test_late_disagreement_does_not_flip_a_true_bit(self):
        """The regression the sequential test fixes: three cond-last then two cond-first used to hit
        fraction 0.6 -> escalate, and the next cond-first (3/6 = 0.5) returned False on the spot."""
        pattern = [True, True, True, False, False, False] + [True] * 30
        conn = FakeConnection(votes(*pattern))
        self.assertTrue(readBit(conn, REQ_COND, REQ_NEG, votes=4))
        self.assertGreater(len(conn.sent), 6)

    def test_coin_flip_reads_false(self):
        """A DBMS that errors past the end of a string makes both requests error, so the order is a coin
        flip - that must terminate the string, not invent a character."""
        conn = FakeConnection(votes(*([True, False] * 40)))
        self.assertFalse(readBit(conn, REQ_COND, REQ_NEG, votes=4))

    def test_cap_is_respected(self):
        conn = FakeConnection(votes(*([True, False] * 200)))
        readBit(conn, REQ_COND, REQ_NEG, votes=4)
        self.assertLessEqual(len(conn.sent), 4 * SPRT_CAP_FACTOR)

    def test_clean_model_costs_no_more_than_the_classic_rule(self):
        """A target whose tuning sweep was unanimous is read at SPRT_P_TRUE_CLEAN. Live measurement on such
        a target (6990 votes, idle and under 4x load) never produced a disagreeing vote, so the model has to
        pay about what the old fixed five-pair rule did or the accuracy work is a straight slowdown."""
        conn = FakeConnection(votes(*([True] * 20)))
        self.assertTrue(readBit(conn, REQ_COND, REQ_NEG, votes=4, pTrue=_timeless.SPRT_P_TRUE_CLEAN))
        self.assertLessEqual(len(conn.sent), 6)

        conn = FakeConnection(votes(*([False] * 20)))
        self.assertFalse(readBit(conn, REQ_COND, REQ_NEG, votes=4, pTrue=_timeless.SPRT_P_TRUE_CLEAN))
        self.assertLessEqual(len(conn.sent), 4)

    def test_clean_model_still_terminates_a_string(self):
        """The end-of-string coin flip is the failure this engine exists to prevent, so it must hold under
        the cheaper model too."""
        conn = FakeConnection(votes(*([True, False] * 60)))
        self.assertFalse(readBit(conn, REQ_COND, REQ_NEG, votes=4, pTrue=_timeless.SPRT_P_TRUE_CLEAN))

    def test_minimum_sample_is_honoured(self):
        conn = FakeConnection(votes(*([False] * 40)))
        readBit(conn, REQ_COND, REQ_NEG, votes=12)
        self.assertGreaterEqual(len(conn.sent), 12)

    def test_error_rates_are_bounded(self):
        """Drive the decision engine with a deterministic pseudo-random vote stream and check that the
        realised error rate on both hypotheses stays inside the configured bound (with slack for the
        finite cap)."""
        def stream(p, seed, count):
            state, out = seed, []
            for _ in range(count):
                state = (1103515245 * state + 12345) % (1 << 31)
                out.append((state >> 16) % 1000 < p * 1000)
            return out

        wrong = 0
        for trial in range(60):             # genuine TRUE bits at the modelled per-vote reliability
            conn = FakeConnection(votes(*stream(_timeless.SPRT_P_TRUE, trial + 1, 2000)))
            wrong += not readBit(conn, REQ_COND, REQ_NEG, votes=4)
        self.assertLessEqual(wrong, 6)      # bound is 2%, allow slack for the finite sample

        wrong = 0
        for trial in range(60):             # end-of-string coin flip: must never invent a character
            conn = FakeConnection(votes(*stream(0.5, trial + 1, 2000)))
            wrong += bool(readBit(conn, REQ_COND, REQ_NEG, votes=4))
        self.assertLessEqual(wrong, 6)

        wrong = 0
        for trial in range(60):             # real FALSE bits (cond is strictly the cheap one)
            conn = FakeConnection(votes(*stream(0.02, trial + 1, 2000)))
            wrong += bool(readBit(conn, REQ_COND, REQ_NEG, votes=4))
        self.assertEqual(wrong, 0)


class TimelessTransportTest(unittest.TestCase):
    def test_unprocessed_stream_is_retried_on_a_factory(self):
        conns = [FakeConnection([True], raises=[_UnprocessedStream("goaway")]), FakeConnection([True])]
        made = []

        def factory():
            made.append(conns[len(made)])
            return made[-1]

        first, loSid, hiSid, status = _pairOrder(factory, REQ_COND, REQ_NEG, 5)
        self.assertEqual(len(made), 2)
        self.assertEqual(status, 200)
        self.assertEqual((loSid, hiSid), (1, 3))
        self.assertEqual(first, loSid)

    def test_ambiguous_transport_error_is_not_replayed(self):
        """A generic drop proves nothing about whether the application saw the requests - re-sending it
        could re-run a state-changing injection point."""
        conn = FakeConnection([True], raises=[socket.error("connection reset")])
        self.assertRaises(socket.error, _pairOrder, lambda: conn, REQ_COND, REQ_NEG, 5)
        self.assertEqual(conn.closed, 1)

    def test_status_mismatch_between_streams_raises(self):
        conn = FakeConnection([True] * 8, statuses=[(200, 403)] * 8)
        self.assertRaises(TimelessUnusable, _pairOrder, conn, REQ_COND, REQ_NEG, 5)

    def test_drift_from_the_calibrated_status_raises(self):
        conn = FakeConnection([True] * 8, statuses=[302] * 8)
        self.assertRaises(TimelessUnusable, _pairOrder, conn, REQ_COND, REQ_NEG, 5, expectStatus=200)

    def test_transient_status_blip_is_tolerated(self):
        conn = FakeConnection([True] * 8, statuses=[500, 200])
        _first, _lo, _hi, status = _pairOrder(conn, REQ_COND, REQ_NEG, 5, expectStatus=200)
        self.assertEqual(status, 200)

    def test_calibrate_reports_the_control_status(self):
        conn = FakeConnection(votes(*([True] * 10)), statuses=[201] * 10)
        usable, confidence, status = calibrate(conn, REQ_COND, REQ_NEG, trials=10)
        self.assertTrue(usable)
        self.assertEqual(confidence, 1.0)
        self.assertEqual(status, 201)

    def test_calibrate_rejects_a_serializing_backend(self):
        conn = FakeConnection([True] * 20)              # order tracks arrival, not work
        usable, confidence, _status = calibrate(conn, REQ_COND, REQ_NEG, trials=10)
        self.assertFalse(usable)
        self.assertLess(confidence, 0.9)


class TimelessSpecTest(unittest.TestCase):
    def test_authority_comes_from_the_host_header(self):
        spec = ("https://10.0.0.5/path?a=1", "GET", {"Host": "vhost.example", "X-A": "1"}, None)
        req = _specToReq(spec, "fallback")
        self.assertEqual(req["authority"], "vhost.example")
        self.assertEqual(req["path"], "/path?a=1")

    def test_authority_falls_back_to_the_url(self):
        spec = ("https://10.0.0.5/path", "GET", {"X-A": "1"}, None)
        self.assertEqual(_specToReq(spec, "fallback")["authority"], "10.0.0.5")

    def test_connection_nominated_fields_reach_the_http2_normalizer(self):
        """Stripping Connection here would leave the fields it names in the h2 request, which is
        malformed - the HTTP/2 layer needs to see it to remove them."""
        spec = ("https://h/p", "GET", {"Connection": "X-Remove", "X-Remove": "v"}, None)
        req = _specToReq(spec, "h")
        self.assertIn("Connection", req["headers"])

        from lib.request.http2 import _normalize_request_headers
        names = [name for name, _value in _normalize_request_headers(req["headers"], b"")]
        self.assertNotIn(b"x-remove", names)
        self.assertNotIn(b"connection", names)

    def test_userinfo_is_dropped_from_the_authority(self):
        spec = ("https://user:pass@host.example:8443/p", "GET", {}, None)
        self.assertEqual(_specToReq(spec, "h")["authority"], "host.example:8443")


class TimelessPayloadTest(unittest.TestCase):
    def test_negation_spans_newlines(self):
        value = "AND 1=(CASE WHEN (%sASCII(SUBSTR(x,\n1,1))>64%s) THEN 1 ELSE 1 END)" % (
            _timeless.INFERENCE_BEGIN, _timeless.INFERENCE_END)
        negated = negatePayload(value)
        self.assertIsNotNone(negated)
        self.assertIn("ASCII(SUBSTR(x,\n1,1))>64", negated)
        self.assertIn("THEN 1 ELSE 0 END)=0", negated)

    def test_negation_requires_sentinels(self):
        self.assertIsNone(negatePayload("AND 1=1"))
        self.assertIsNone(negatePayload(None))

    def test_negation_is_null_safe(self):
        """Plain NOT() would leave both requests cheap at end-of-string (NOT NULL is NULL), so neither
        branch runs heavy and the order is noise."""
        self.assertNotIn("NOT(", negateCondition("ORD(x)>64"))
        self.assertEqual(negateCondition("ORD(x)>64"), "(CASE WHEN (ORD(x)>64) THEN 1 ELSE 0 END)=0")

    def test_every_primitive_consumes_the_cost(self):
        """A primitive that ignores [COST] makes every rung of the ladder send the identical query while
        reporting a different cost."""
        for dbms, primitive in LIGHT_HEAVY.items():
            self.assertIn("[COST]", primitive, "%s primitive ignores [COST]" % dbms)

    def test_every_primitive_yields_a_sentinel_vector(self):
        for dbms in LIGHT_HEAVY:
            vector = lightHeavyVector(dbms, LIGHT_HEAVY_COSTS[0])
            self.assertIn(_timeless.INFERENCE_BEGIN, vector)
            self.assertIn(_timeless.INFERENCE_END, vector)
            self.assertIsNotNone(negatePayload(vector))

    def test_condition_pair_runs_heavy_exactly_once(self):
        cond, neg = buildConditionPair("ORD(x)>64", "HEAVY")
        self.assertEqual(cond.count("HEAVY"), 1)
        self.assertEqual(neg.count("HEAVY"), 1)
        self.assertNotEqual(cond, neg)


class TimelessOracleTest(unittest.TestCase):
    def _oracle(self, connections, **kwargs):
        oracle = TimelessOracle("h", 443, REQ_COND, REQ_NEG, votes=4, status=200, **kwargs)
        made = []

        def opener(host, port, proxy, timeout):
            made.append(connections[len(made)])
            return made[-1]

        oracle._opened = made
        _timeless.connect, self._savedConnect = opener, _timeless.connect
        return oracle

    def tearDown(self):
        if hasattr(self, "_savedConnect"):
            _timeless.connect = self._savedConnect
            del self._savedConnect

    def test_a_fresh_connection_is_calibrated_before_it_reads(self):
        conn = FakeConnection(votes(*([True] * 40)))
        oracle = self._oracle([conn])
        oracle._conn()
        self.assertEqual(len(conn.sent), _timeless.VERIFY_TRIALS)

    def test_a_serializing_replacement_connection_is_refused(self):
        """The initial sweep proves one connection to one backend node; a later connection can land on a
        node that serializes streams and would return wrong bits without ever raising."""
        conn = FakeConnection([True] * 40)              # first-sent always finishes first
        oracle = self._oracle([conn])
        self.assertRaises(TimelessUnusable, oracle._conn)
        self.assertEqual(conn.closed, 1)
        self.assertEqual(oracle._conns, [])

    def test_close_does_not_resurrect_a_concurrently_opened_connection(self):
        conn = FakeConnection(votes(*([True] * 40)))
        oracle = self._oracle([conn])
        started, release = threading.Event(), threading.Event()
        opener = _timeless.connect

        def blocking(*args, **kwargs):
            started.set()
            release.wait(5)
            return opener(*args, **kwargs)

        _timeless.connect = blocking
        worker = threading.Thread(target=lambda: self.assertRaises(TimelessUnusable, oracle._conn))
        worker.daemon = True
        worker.start()
        started.wait(5)
        oracle.close()                      # closes while the worker is still inside connect()
        release.set()
        worker.join(5)
        self.assertEqual(oracle._conns, [])
        self.assertEqual(conn.closed, 1)

    def test_clean_model_demands_a_unanimous_connection(self):
        """The cheap model is only sound where votes are deterministic, so a connection that cannot
        reproduce the unanimity the tuning sweep saw must not be read at it."""
        oneMiss = votes(*([True] * 3 + [False] + [True] * 40))
        oracle = self._oracle([FakeConnection(oneMiss)], verifyThreshold=_timeless.VERIFY_THRESHOLD_CLEAN,
                              pTrue=_timeless.SPRT_P_TRUE_CLEAN)
        self.assertRaises(TimelessUnusable, oracle._conn)

        oracle = self._oracle([FakeConnection(oneMiss)])         # conservative model tolerates it
        self.assertTrue(oracle._conn())

    def test_close_is_idempotent(self):
        conn = FakeConnection(votes(*([True] * 40)))
        oracle = self._oracle([conn])
        oracle._conn()
        oracle.close()
        oracle.close()
        self.assertEqual(conn.closed, 1)


class TimelessEngagementTest(unittest.TestCase):
    """engage()/disengage() are the state machine the runtime fallback depends on. Restoring the vector
    alone is NOT enough - bisection freezes a comparison template from it when a value starts extracting,
    so disengage() must also publish the classic vector for queryPage to re-forge the stragglers with,
    or the rest of that value keeps sending heavy (no-delay) payloads that read as all-False."""

    def setUp(self):
        from lib.core.data import kb
        from lib.core.datatype import AttribDict
        from lib.core.enums import PAYLOAD

        self.kb, self.technique = kb, PAYLOAD.TECHNIQUE.TIME
        self._saved = (kb.get("injection"), kb.get("timeless"), kb.get("timelessRestore"))
        kb.injection = AttribDict({"data": AttribDict({self.technique: AttribDict({"vector": "AND SLEEP([SLEEPTIME]) [INFERENCE]"})})})
        kb.timeless = None
        kb.timelessRestore = None

    def tearDown(self):
        self.kb.injection, self.kb.timeless, self.kb.timelessRestore = self._saved

    def _oracle(self):
        return TimelessOracle("h", 443, REQ_COND, REQ_NEG, status=200)

    def test_engage_swaps_the_vector_and_publishes_last(self):
        oracle = self._oracle()
        _timeless.engage(oracle, self.technique, "AND HEAVY [INFERENCE]")
        self.assertEqual(self.kb.injection.data[self.technique].vector, "AND HEAVY [INFERENCE]")
        self.assertIs(self.kb.timeless, oracle)
        self.assertIsNone(self.kb.timelessRestore)

    def test_disengage_restores_and_arms_the_straggler_rewrite(self):
        oracle = self._oracle()
        _timeless.engage(oracle, self.technique, "AND HEAVY [INFERENCE]")
        _timeless.disengage()
        self.assertEqual(self.kb.injection.data[self.technique].vector, "AND SLEEP([SLEEPTIME]) [INFERENCE]")
        self.assertIsNone(self.kb.timeless)
        self.assertEqual(self.kb.timelessRestore, "AND SLEEP([SLEEPTIME]) [INFERENCE]")

    def test_re_engaging_clears_a_previous_targets_rewrite(self):
        """A stale rewrite vector outliving its target would re-forge the next one's payloads."""
        _timeless.engage(self._oracle(), self.technique, "AND HEAVY [INFERENCE]")
        _timeless.disengage()
        self.assertIsNotNone(self.kb.timelessRestore)
        _timeless.engage(self._oracle(), self.technique, "AND HEAVY2 [INFERENCE]")
        self.assertIsNone(self.kb.timelessRestore)

    def test_disengage_holds_the_data_object_not_the_technique(self):
        """disengage() can run after the controller moved to the next target; re-resolving the technique
        then would restore this target's vector onto the next one's injection data."""
        from lib.core.datatype import AttribDict

        oracle = self._oracle()
        _timeless.engage(oracle, self.technique, "AND HEAVY [INFERENCE]")
        stale = self.kb.injection.data[self.technique]
        self.kb.injection = AttribDict({"data": AttribDict({self.technique: AttribDict({"vector": "NEXT TARGET [INFERENCE]"})})})
        _timeless.disengage()
        self.assertEqual(self.kb.injection.data[self.technique].vector, "NEXT TARGET [INFERENCE]")
        self.assertEqual(stale.vector, "AND SLEEP([SLEEPTIME]) [INFERENCE]")

    def test_straggler_rewrite_ignores_a_payload_without_sentinels(self):
        """The rewrite must be a no-op on any ordinary time-based payload."""
        self.assertIsNone(_timeless.restoreClassicValue("id=1 AND SLEEP(5)", "AND SLEEP([SLEEPTIME]) [INFERENCE]"))
        self.assertIsNone(_timeless.restoreClassicValue(None, "AND SLEEP([SLEEPTIME]) [INFERENCE]"))


if __name__ == "__main__":
    unittest.main()