#!/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.
"""

import binascii
import os
import sys
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


def _hex(text):
    return binascii.hexlify(text.encode(settings.DEFAULT_CODEC)).decode()


class MockChannel:
    """Minimal OOB channel returning a fixed set of DNS interactions."""

    def __init__(self):
        self._interactions = {}  # token -> [Interaction]
        self._labels = {}        # id(interaction) -> [label, ...]
        self._np_counter = 0

    def add(self, token, index, total, hex_data):
        label0 = str(index) + "-" + (str(total) if total is not None else "")
        itr = Interaction(protocol="dns", identifier="dummy." + token, raw_request="")
        self._labels[id(itr)] = [label0] + hex_data if isinstance(hex_data, list) else [label0, hex_data]
        self._interactions.setdefault(token, []).append(itr)
        return itr

    def seen(self, token, protocol=None):
        hits = list(self._interactions.get(token, []))
        return [i for i in hits if i.protocol == protocol] if protocol else hits

    def data_of(self, interaction):
        return self._labels.get(id(interaction), [])

    def poll_now(self):
        pass

    def new_payload(self):
        self._np_counter += 1
        token = "t{:012d}".format(self._np_counter)
        return token, "corrid" + token + ".oast.fun"


# ---------------------------------------------------------------------------
# decode_dns_output — pure reassembly from a chunks dict
# ---------------------------------------------------------------------------

class TestDecodeDnsOutput(unittest.TestCase):

    def test_single_chunk_ascii(self):
        self.assertEqual(payloads.decode_dns_output({1: _hex("hello")}), "hello")

    def test_multiple_chunks_in_order(self):
        hexed = _hex("hello world")
        mid = (len(hexed) // 2) & ~1  # keep even so no half-byte
        self.assertEqual(
            payloads.decode_dns_output({1: hexed[:mid], 2: hexed[mid:]}),
            "hello world",
        )

    def test_chunks_out_of_order_reassembled_correctly(self):
        hexed = _hex("abcdef")
        mid = (len(hexed) // 2) & ~1
        # deliver chunk 2 first
        self.assertEqual(
            payloads.decode_dns_output({2: hexed[mid:], 1: hexed[:mid]}),
            "abcdef",
        )

    def test_trailing_newline_stripped(self):
        self.assertEqual(payloads.decode_dns_output({1: _hex("hello\n")}), "hello")

    def test_trailing_crlf_stripped(self):
        self.assertEqual(payloads.decode_dns_output({1: _hex("hello\r\n")}), "hello")

    def test_empty_chunks_returns_empty_string(self):
        self.assertEqual(payloads.decode_dns_output({}), "")

    def test_non_ascii_bytes_round_trip(self):
        # od -tx1 encodes every byte, so non-ASCII survives the round trip.
        raw = b"\xc3\xa9l\xc3\xa8ve"  # UTF-8 "élève"
        hexed = binascii.hexlify(raw).decode()
        result = payloads.decode_dns_output({1: hexed})
        self.assertEqual(result.encode(settings.DEFAULT_CODEC, errors="replace"), raw)

    def test_gap_in_chunk_indices_returns_partial_decode(self):
        # decode_dns_output joins whatever chunks arrived in index order; a gap does not
        # crash it.  INCOMPLETE_OUTPUT is set by _dns_output after the loop, not here.
        hexed = _hex("abcdef")
        third = (len(hexed) // 3) & ~1
        # chunks 1 and 3 present; chunk 2 missing — result is defined but not "abcdef"
        result = payloads.decode_dns_output({1: hexed[:third], 3: hexed[third * 2:]})
        self.assertIsInstance(result, str)

    def test_contiguous_chunks_do_not_set_incomplete(self):
        hexed = _hex("hello world")
        mid = (len(hexed) // 2) & ~1
        settings.INCOMPLETE_OUTPUT = False
        payloads.decode_dns_output({1: hexed[:mid], 2: hexed[mid:]})
        self.assertFalse(settings.INCOMPLETE_OUTPUT)

    def test_hex_labels_are_lowercased_by_resolver(self):
        # A resolver may uppercase labels; decode_dns_output must still work.
        hexed = _hex("hi").upper()
        self.assertEqual(payloads.decode_dns_output({1: hexed}), "hi")


# ---------------------------------------------------------------------------
# _dns_output — the chunk-collection loop that drives the mock channel
# ---------------------------------------------------------------------------

class TestDnsOutputLoop(unittest.TestCase):

    def setUp(self):
        settings.OOB_TIMEOUT = 15
        settings.INCOMPLETE_OUTPUT = False

    # --- loop-format (separator `;`/`\n`): label carries "index-total" ---

    def test_loop_format_single_chunk(self):
        token = "aaa"
        ch = MockChannel()
        ch.add(token, index=1, total=1, hex_data=_hex("hello"))
        # time: call 1 → deadline base (0), call 2 → inside loop, never reached because
        # total is known and len(chunks)>=total breaks before the deadline check.
        with mock.patch("time.time", side_effect=[0, 0]):
            with mock.patch("time.sleep"):
                result = oob_injector._dns_output(ch, token)
        self.assertEqual(result, "hello")

    def test_loop_format_multi_chunk_all_arrive(self):
        token = "bbb"
        hexed = _hex("hello world")
        mid = (len(hexed) // 2) & ~1
        ch = MockChannel()
        ch.add(token, index=1, total=2, hex_data=hexed[:mid])
        ch.add(token, index=2, total=2, hex_data=hexed[mid:])
        with mock.patch("time.time", side_effect=[0, 0]):
            with mock.patch("time.sleep"):
                result = oob_injector._dns_output(ch, token)
        self.assertEqual(result, "hello world")

    def test_loop_format_out_of_order_arrival(self):
        # Chunks registered in reverse order; index key must still sort correctly.
        token = "ccc"
        hexed = _hex("xyz123")
        mid = (len(hexed) // 2) & ~1
        ch = MockChannel()
        ch.add(token, index=2, total=2, hex_data=hexed[mid:])
        ch.add(token, index=1, total=2, hex_data=hexed[:mid])
        with mock.patch("time.time", side_effect=[0, 0]):
            with mock.patch("time.sleep"):
                result = oob_injector._dns_output(ch, token)
        self.assertEqual(result, "xyz123")

    # --- xargs-format (separator `&`/`|`): label carries "index-" (no total) ---
    # quiet_polls reaches 2 after two iterations with no new chunks → break.

    def test_xargs_format_exits_on_quiet_polls(self):
        token = "ddd"
        ch = MockChannel()
        ch.add(token, index=1, total=None, hex_data=_hex("world"))
        # time calls: [0] deadline base, [0] iter1 deadline check, [0] iter2 check.
        # iter3: quiet_polls==2 → break before deadline check.
        with mock.patch("time.time", side_effect=[0, 0, 0]):
            with mock.patch("time.sleep"):
                result = oob_injector._dns_output(ch, token)
        self.assertEqual(result, "world")

    def test_xargs_format_hex_label_split_across_parts(self):
        # `data_of` may return more than two elements; labels[1:] are joined.
        token = "eee"
        ch = MockChannel()
        ch.add(token, index=1, total=None, hex_data=["61", "62"])  # "ab" split
        with mock.patch("time.time", side_effect=[0, 0, 0]):
            with mock.patch("time.sleep"):
                result = oob_injector._dns_output(ch, token)
        self.assertEqual(result, "ab")

    # --- timeout and partial output ---

    def test_timeout_partial_chunks_sets_incomplete(self):
        token = "fff"
        hexed = _hex("hi there")
        mid = (len(hexed) // 2) & ~1
        ch = MockChannel()
        ch.add(token, index=1, total=2, hex_data=hexed[:mid])
        # chunk 2 never arrives; deadline passes immediately inside the loop.
        # time: call 1 → 0 (deadline base, deadline = 30), call 2 → 9999 (past deadline).
        settings.INCOMPLETE_OUTPUT = False
        with mock.patch("time.time", side_effect=[0, 9999]):
            with mock.patch("time.sleep"):
                oob_injector._dns_output(ch, token)
        self.assertTrue(settings.INCOMPLETE_OUTPUT)

    def test_no_interactions_returns_empty_string(self):
        ch = MockChannel()
        with mock.patch("time.time", side_effect=[0, 9999]):
            with mock.patch("time.sleep"):
                result = oob_injector._dns_output(ch, "ggg")
        self.assertEqual(result, "")

    def test_interaction_with_too_few_labels_skipped(self):
        # A label list shorter than 2 must be silently ignored.
        token = "hhh"
        ch = MockChannel()
        itr = Interaction(protocol="dns", identifier="dummy." + token, raw_request="")
        ch._labels[id(itr)] = ["1-1"]  # only one label, no hex part
        ch._interactions[token] = [itr]
        with mock.patch("time.time", side_effect=[0, 9999]):
            with mock.patch("time.sleep"):
                result = oob_injector._dns_output(ch, token)
        self.assertEqual(result, "")

    def test_non_digit_index_label_skipped(self):
        # A label whose first part is not a digit must not crash or add to chunks.
        token = "iii"
        ch = MockChannel()
        itr = Interaction(protocol="dns", identifier="dummy." + token, raw_request="")
        ch._labels[id(itr)] = ["probe-1", _hex("x")]  # "probe" is not a digit
        ch._interactions[token] = [itr]
        with mock.patch("time.time", side_effect=[0, 9999]):
            with mock.patch("time.sleep"):
                result = oob_injector._dns_output(ch, token)
        self.assertEqual(result, "")

    def test_gap_in_chunk_indices_sets_incomplete(self):
        # Chunks 1 and 3 arrive (total=3); chunk 2 never does.
        # Deadline passes before total is satisfied → INCOMPLETE_OUTPUT set.
        token = "kkk"
        hexed = _hex("abcdef")
        third = (len(hexed) // 3) & ~1
        ch = MockChannel()
        ch.add(token, index=1, total=3, hex_data=hexed[:third])
        ch.add(token, index=3, total=3, hex_data=hexed[third * 2:])
        settings.INCOMPLETE_OUTPUT = False
        # time: call 1 → 0 (deadline=30), call 2 → 9999 (past deadline immediately)
        with mock.patch("time.time", side_effect=[0, 9999]):
            with mock.patch("time.sleep"):
                oob_injector._dns_output(ch, token)
        self.assertTrue(settings.INCOMPLETE_OUTPUT)

    def test_trailing_newline_stripped_end_to_end(self):
        # The pipeline always appends a newline; _dns_output must strip it.
        token = "jjj"
        ch = MockChannel()
        ch.add(token, index=1, total=1, hex_data=_hex("result\n"))
        with mock.patch("time.time", side_effect=[0, 0]):
            with mock.patch("time.sleep"):
                result = oob_injector._dns_output(ch, token)
        self.assertEqual(result, "result")


# ---------------------------------------------------------------------------
# Gap 1: dns_exfil_command label format vs. _dns_output expectations
# ---------------------------------------------------------------------------

class TestDnsExfilCommandFormat(unittest.TestCase):

    def setUp(self):
        settings.TARGET_OS = settings.OS.UNIX

    def tearDown(self):
        settings.TARGET_OS = settings.OS.UNIX

    # Loop-format separators (`;` and `\n`) use a while-loop that knows its total.
    def test_loop_separator_contains_index_total_pattern(self):
        cmd = payloads.dns_exfil_command("tok.oast.fun", "id", ";")
        self.assertIn("$n-$t", cmd)

    def test_loop_separator_encodes_output_with_od(self):
        cmd = payloads.dns_exfil_command("tok.oast.fun", "id", ";")
        self.assertIn("od -An -v -tx1", cmd)

    def test_loop_separator_contains_lookup_command(self):
        cmd = payloads.dns_exfil_command("tok.oast.fun", "id", ";")
        self.assertIn("nslookup", cmd)

    # Xargs-format separators (`&` and `|`) number chunks with `cat -n` + `tr`.
    def test_xargs_separator_contains_cat_n(self):
        cmd = payloads.dns_exfil_command("tok.oast.fun", "id", "&")
        self.assertIn("cat -n", cmd)

    def test_xargs_separator_converts_tab_to_dot(self):
        cmd = payloads.dns_exfil_command("tok.oast.fun", "id", "&")
        self.assertIn("tr '\\t' '.'", cmd)

    def test_xargs_separator_has_no_total_label(self):
        # Xargs format produces plain numbering ("1.<hex>"), not "1-3"
        cmd = payloads.dns_exfil_command("tok.oast.fun", "id", "&")
        self.assertNotIn("$n-$t", cmd)

    def test_xargs_label_parses_in_dns_output_loop(self):
        # labels[0] = "1" (no dash) is still parsed correctly via partition("-").
        token = "parsexargs"
        ch = MockChannel()
        hexed = _hex("ok")
        itr = Interaction(protocol="dns", identifier="dummy." + token, raw_request="")
        ch._labels[id(itr)] = ["1", hexed]   # plain index, no dash
        ch._interactions[token] = [itr]
        with mock.patch("time.time", side_effect=[0, 0, 0]):
            with mock.patch("time.sleep"):
                result = oob_injector._dns_output(ch, token)
        self.assertEqual(result, "ok")

    def test_windows_uses_powershell_and_bitconverter(self):
        settings.TARGET_OS = settings.OS.WINDOWS
        cmd = payloads.dns_exfil_command("tok.oast.fun", "id", "&")
        self.assertIn("powershell", cmd.lower())
        self.assertIn("BitConverter", cmd)

    def test_windows_label_carries_index_total(self):
        settings.TARGET_OS = settings.OS.WINDOWS
        cmd = payloads.dns_exfil_command("tok.oast.fun", "id", "&")
        # PowerShell embeds ($i/chunk)+1 and $t in the nslookup target
        self.assertIn("$t", cmd)
        self.assertIn("nslookup", cmd)


# ---------------------------------------------------------------------------
# Gap 2: Interactsh identifier parsing (_token_of / data_of)
# ---------------------------------------------------------------------------

class TestInteractshParsing(unittest.TestCase):

    CORR = "aabbccddeeaabbccddee"   # 20-char correlation id
    TOKEN = "aaabbbcccddd"           # 13-char probe token

    def _client(self):
        from src.core.oob.interactsh import Interactsh
        c = Interactsh.__new__(Interactsh)
        c.correlation_id = self.CORR
        c.domain = "oast.fun"
        return c

    def _itr(self, identifier):
        return Interaction(protocol="dns", identifier=identifier, raw_request="")

    def test_token_of_plain_probe(self):
        identifier = self.CORR + self.TOKEN + ".oast.fun"
        self.assertEqual(self._client()._token_of(self._itr(identifier)), self.TOKEN)

    def test_token_of_data_carrying(self):
        identifier = "1-3.deadbeef." + self.CORR + self.TOKEN + ".oast.fun"
        self.assertEqual(self._client()._token_of(self._itr(identifier)), self.TOKEN)

    def test_token_of_unknown_returns_none(self):
        self.assertIsNone(self._client()._token_of(self._itr("unknown.label.oast.fun")))

    def test_data_of_plain_probe_is_empty(self):
        identifier = self.CORR + self.TOKEN + ".oast.fun"
        self.assertEqual(self._client().data_of(self._itr(identifier)), [])

    def test_data_of_single_chunk(self):
        chunk = "6865786f6b"
        identifier = "1-1." + chunk + "." + self.CORR + self.TOKEN + ".oast.fun"
        self.assertEqual(self._client().data_of(self._itr(identifier)), ["1-1", chunk])

    def test_data_of_multipart_chunk(self):
        identifier = "1-2.part1.part2." + self.CORR + self.TOKEN + ".oast.fun"
        self.assertEqual(self._client().data_of(self._itr(identifier)), ["1-2", "part1", "part2"])

    def test_data_and_token_agree_on_boundary(self):
        # data_of and _token_of must find the same boundary label.
        chunk = "6162"
        identifier = "1-1." + chunk + "." + self.CORR + self.TOKEN + ".oast.fun"
        c = self._client()
        itr = self._itr(identifier)
        self.assertEqual(c._token_of(itr), self.TOKEN)
        labels = c.data_of(itr)
        self.assertEqual(labels[0], "1-1")
        self.assertEqual(labels[1], chunk)

    def test_data_of_xargs_format_plain_index(self):
        # Xargs produces "1.<hex>.<corr+token>.<domain>" — label0 is a plain number.
        chunk = "6162"
        identifier = "1." + chunk + "." + self.CORR + self.TOKEN + ".oast.fun"
        labels = self._client().data_of(self._itr(identifier))
        self.assertEqual(labels, ["1", chunk])

    def test_token_of_is_case_insensitive(self):
        # interactsh identifiers arrive lowercased; _token_of lower()s the label.
        identifier = (self.CORR + self.TOKEN + ".oast.fun").upper()
        self.assertEqual(self._client()._token_of(self._itr(identifier)), self.TOKEN)


# ---------------------------------------------------------------------------
# Gap 3: injection() retry logic
# ---------------------------------------------------------------------------

class TestInjectionRetry(unittest.TestCase):

    def setUp(self):
        settings.TARGET_OS = settings.OS.UNIX
        settings.OOB_EVAL = False
        settings.OOB_TRANSPORT = "dns"
        settings.OOB_DNS_ATTEMPTS = 2
        settings.OOB_TIMEOUT = 15
        settings.INCOMPLETE_OUTPUT = False
        self._ch = MockChannel()
        settings.OOB_CHANNEL = self._ch

    def tearDown(self):
        settings.OOB_CHANNEL = None
        settings.OOB_TRANSPORT = ""

    def _run(self, dns_side_effects):
        with mock.patch("src.core.techniques.oob.oob_injector._dns_output",
                        side_effect=dns_side_effects) as m:
            with mock.patch("src.core.controller.handler.oob_inject"):
                result = oob_injector.injection(";", "id", "", "", " ", "GET",
                                                "http://t/", "p")
        return result, m

    def test_retries_on_empty_first_response(self):
        result, m = self._run(["", "data"])
        self.assertEqual(result, "data")
        self.assertEqual(m.call_count, 2)

    def test_no_retry_when_first_attempt_succeeds(self):
        result, m = self._run(["first"])
        self.assertEqual(result, "first")
        self.assertEqual(m.call_count, 1)

    def test_both_empty_returns_empty_after_attempts(self):
        result, m = self._run(["", ""])
        self.assertEqual(result, "")
        self.assertEqual(m.call_count, 2)

    def test_attempts_1_never_retries(self):
        settings.OOB_DNS_ATTEMPTS = 1
        result, m = self._run([""])
        self.assertEqual(result, "")
        self.assertEqual(m.call_count, 1)

    def test_null_channel_returns_empty_immediately(self):
        settings.OOB_CHANNEL = None
        with mock.patch("src.core.techniques.oob.oob_injector._dns_output") as m:
            result = oob_injector.injection(";", "id", "", "", " ", "GET", "http://t/", "p")
        self.assertEqual(result, "")
        m.assert_not_called()

    def test_unsupported_transport_returns_empty_immediately(self):
        settings.OOB_TRANSPORT = "notreal"
        with mock.patch("src.core.techniques.oob.oob_injector._dns_output") as m:
            result = oob_injector.injection(";", "id", "", "", " ", "GET", "http://t/", "p")
        self.assertEqual(result, "")
        m.assert_not_called()


# ---------------------------------------------------------------------------
# Gap 4: heuristic_payload DNS probe
# ---------------------------------------------------------------------------

class TestHeuristicPayload(unittest.TestCase):

    def setUp(self):
        settings.TARGET_OS = settings.OS.UNIX
        settings.OOB_SCHEME = "https"

    def tearDown(self):
        settings.TARGET_OS = settings.OS.UNIX

    def test_posix_produces_three_probes(self):
        ch = MockChannel()
        _, probes = payloads.heuristic_payload(ch, settings.OS.UNIX)
        self.assertEqual(len(probes), 3)

    def test_windows_produces_three_probes(self):
        ch = MockChannel()
        _, probes = payloads.heuristic_payload(ch, settings.OS.WINDOWS)
        self.assertEqual(len(probes), 3)

    def test_posix_probe_transports_in_order(self):
        ch = MockChannel()
        _, probes = payloads.heuristic_payload(ch, settings.OS.UNIX)
        # pairs = ((";","curl"), ("&","dns"), ("|","dns"))
        self.assertEqual([t for _, t in probes], ["curl", "dns", "dns"])

    def test_windows_probe_transports_in_order(self):
        ch = MockChannel()
        _, probes = payloads.heuristic_payload(ch, settings.OS.WINDOWS)
        # pairs = (("&","curl"), ("&","dns"), ("|","dns"))
        self.assertEqual([t for _, t in probes], ["curl", "dns", "dns"])

    def test_probe_tokens_are_unique(self):
        ch = MockChannel()
        _, probes = payloads.heuristic_payload(ch, settings.OS.UNIX)
        tokens = [tok for tok, _ in probes]
        self.assertEqual(len(tokens), len(set(tokens)))

    def test_payload_contains_dns_lookup_command(self):
        ch = MockChannel()
        payload, _ = payloads.heuristic_payload(ch, settings.OS.UNIX)
        self.assertIn("nslookup", payload)

    def test_payload_contains_curl_command(self):
        ch = MockChannel()
        payload, _ = payloads.heuristic_payload(ch, settings.OS.UNIX)
        self.assertIn("curl", payload)

    def test_each_probe_token_present_in_payload(self):
        # The hostname for each probe embeds its token; the payload must reference it.
        ch = MockChannel()
        payload, probes = payloads.heuristic_payload(ch, settings.OS.UNIX)
        for token, _ in probes:
            self.assertIn(token, payload)

    def test_payload_is_nonempty_string(self):
        ch = MockChannel()
        payload, _ = payloads.heuristic_payload(ch, settings.OS.UNIX)
        self.assertIsInstance(payload, str)
        self.assertTrue(len(payload) > 0)


# ---------------------------------------------------------------------------
# OOB_EVAL: exfiltrate_eval and injection() with OOB_EVAL=True
# ---------------------------------------------------------------------------

class TestOobEval(unittest.TestCase):

    _HOST = "aabbccddeeaabbccddeetest0000token.oast.fun"

    def setUp(self):
        settings.TARGET_OS = settings.OS.UNIX
        settings.OOB_EVAL = False

    def tearDown(self):
        settings.OOB_EVAL = False

    # --- exfiltrate_eval structure -----------------------------------------

    def test_returns_nonempty_for_dns(self):
        result = payloads.exfiltrate_eval(";", "dns", self._HOST, "echo hello")
        self.assertTrue(result, "exfiltrate_eval returned empty for dns transport")

    def test_wraps_in_eval_grammar_backticks(self):
        # Default eval grammar is PHP: print(`...`);
        result = payloads.exfiltrate_eval(";", "dns", self._HOST, "echo hello")
        self.assertIn("`", result)
        self.assertTrue(result.strip().startswith("print("))

    def test_contains_dns_pipeline_structure(self):
        result = payloads.exfiltrate_eval(";", "dns", self._HOST, "echo hello")
        self.assertIn("nslookup", result)
        self.assertIn("od -An", result)

    def test_inner_command_matches_plain_exfiltrate(self):
        # The backtick content must be exactly what exfiltrate("", ...) builds.
        plain = payloads.exfiltrate("", "dns", self._HOST, "echo hello")
        wrapped = payloads.exfiltrate_eval(";", "dns", self._HOST, "echo hello")
        inner = wrapped.split("`")[1]
        self.assertEqual(inner, plain)

    def test_terminator_appended_when_separator_given(self):
        result = payloads.exfiltrate_eval(";", "dns", self._HOST, "echo hello")
        self.assertTrue(result.endswith(settings.EVAL_GRAMMAR.TERMINATOR))

    def test_no_extra_terminator_when_separator_empty(self):
        # With no separator the eval wrapper adds no extra terminator beyond
        # what _rendered() already produces (the PHP print statement ends in ;).
        with_sep = payloads.exfiltrate_eval(";", "dns", self._HOST, "echo hello")
        without_sep = payloads.exfiltrate_eval("", "dns", self._HOST, "echo hello")
        self.assertEqual(len(without_sep) + len(settings.EVAL_GRAMMAR.TERMINATOR),
                         len(with_sep))

    # --- injection() with OOB_EVAL=True ------------------------------------

    def test_injection_oob_eval_calls_dns_output(self):
        ch = MockChannel()
        settings.OOB_EVAL = True
        settings.OOB_TRANSPORT = "dns"
        settings.OOB_DNS_ATTEMPTS = 1
        settings.OOB_CHANNEL = ch
        try:
            with mock.patch("src.core.techniques.oob.oob_injector._dns_output",
                            return_value="eval_result") as m:
                with mock.patch("src.core.controller.handler.oob_inject"):
                    result = oob_injector.injection(";", "echo hello", "", "", " ",
                                                    "GET", "http://t/", "p")
            self.assertEqual(result, "eval_result")
            self.assertEqual(m.call_count, 1)
        finally:
            settings.OOB_CHANNEL = None
            settings.OOB_TRANSPORT = ""
            settings.OOB_EVAL = False

    def test_injection_oob_eval_retries_same_as_plain(self):
        # OOB_EVAL=True does not change the retry count.
        ch = MockChannel()
        settings.OOB_EVAL = True
        settings.OOB_TRANSPORT = "dns"
        settings.OOB_DNS_ATTEMPTS = 2
        settings.OOB_CHANNEL = ch
        try:
            with mock.patch("src.core.techniques.oob.oob_injector._dns_output",
                            side_effect=["", "second"]) as m:
                with mock.patch("src.core.controller.handler.oob_inject"):
                    result = oob_injector.injection(";", "echo hello", "", "", " ",
                                                    "GET", "http://t/", "p")
            self.assertEqual(result, "second")
            self.assertEqual(m.call_count, 2)
        finally:
            settings.OOB_CHANNEL = None
            settings.OOB_TRANSPORT = ""
            settings.OOB_EVAL = False


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