CERBERUS/sqlmap
1
0
Fork 0
mirror of https://github.com/sqlmapproject/sqlmap.git synced 2026-08-27 04:05:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Fixing --eta

Browse source
This commit is contained in:
Miroslav Štampar 2026-07-09 13:25:50 +02:00
parent 767feba3a7
commit 4c12dda2f7
7 changed files with 285 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

42
tests/test_techniques.py
View file

@ -48,6 +48,7 @@ from lib.core.agent import agent
from lib.core.unescaper import unescaper
from lib.request.connect import Connect
from lib.request.connect import Connect as Request
from lib.request import inject
from lib.utils.hashdb import HashDB
import lib.techniques.union.use as uu
@ -1516,6 +1517,47 @@ class TestConfigUnion(unittest.TestCase):
self.assertEqual(kb.uChar, "SENTINEL")
class TestValueParallelEligibility(unittest.TestCase):
"""
inject.valueParallelEligible() picks the value-parallel path (job-level '--eta' bar / concurrency).
Safety invariant under test: classic time-based must never run concurrently (interfering SLEEP
measurements), so it qualifies only single-threaded under '--eta'; a concurrency-safe channel
(boolean or the timeless oracle) may run under either '--threads' or '--eta'.
"""
def setUp(self):
self._avail = set()
self._realAvail = inject.isTechniqueAvailable
inject.isTechniqueAvailable = lambda t: t in self._avail
self._saved = (conf.threads, conf.eta, kb.get("timeless"))
def tearDown(self):
inject.isTechniqueAvailable = self._realAvail
conf.threads, conf.eta, kb.timeless = self._saved
def _elig(self, threads, eta, techniques, timeless=None):
conf.threads, conf.eta, kb.timeless = threads, eta, timeless
self._avail = set(techniques)
return inject.valueParallelEligible()
def test_single_thread_eta_time_based_qualifies(self):
self.assertTrue(self._elig(1, True, {PAYLOAD.TECHNIQUE.TIME}))
def test_multi_thread_time_based_never_parallel(self):
self.assertFalse(self._elig(8, True, {PAYLOAD.TECHNIQUE.TIME}))
self.assertFalse(self._elig(8, False, {PAYLOAD.TECHNIQUE.TIME}))
def test_boolean_qualifies_under_threads_or_eta(self):
self.assertTrue(self._elig(8, False, {PAYLOAD.TECHNIQUE.BOOLEAN}))
self.assertTrue(self._elig(1, True, {PAYLOAD.TECHNIQUE.BOOLEAN}))
def test_plain_single_thread_no_eta_stays_classic(self):
self.assertFalse(self._elig(1, False, {PAYLOAD.TECHNIQUE.BOOLEAN}))
def test_timeless_is_concurrency_safe(self):
self.assertTrue(self._elig(8, True, {PAYLOAD.TECHNIQUE.TIME}, timeless=object()))
if __name__ == "__main__":
unittest.main(verbosity=2)