diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..31e6b2e --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,24 @@ +name: tests + +on: + push: + pull_request: + +jobs: + pytest: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ['3.9', '3.11'] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install + run: | + python -m pip install --upgrade pip + python -m pip install -e '.[test]' + - name: Run tests + run: pytest -q diff --git a/netbyte/core.py b/netbyte/core.py new file mode 100644 index 0000000..5e2c76f --- /dev/null +++ b/netbyte/core.py @@ -0,0 +1,87 @@ +"""Core helpers for netbyte. + +This module is intentionally free of socket/CLI code so it can be unit tested. +""" + +_SYMBOLS = set("~`!@#$%^&*()_-+={}[]:>;', bool: + return ch in _SYMBOLS + + +def _is_printable_reference_byte(b: int) -> bool: + # Keep the original tool's intent: annotate common readable bytes. + if b < 0x20 or b >= 0x7F: + return False + ch = chr(b) + return ch.isalpha() or ch.isdigit() or is_symbol(ch) + + +def to_hex(data: bytes) -> str: + """Convert bytes to formatted hex with lightweight ASCII references. + + Output is similar to the original Python 2 implementation, e.g.: + 48(H) 65(e) 6C(l) 6C(l) 6F(o) 0D 0A(\n) + + Newlines (0x0A) are rendered with an explicit (\n) marker and a real newline. + """ + + results: list[str] = [] + new_line = True + + for b in data: + hex_value = f"{b:02X}" + + if _is_printable_reference_byte(b): + hex_value = f"{hex_value}({chr(b)})" + + if not new_line: + hex_value = " " + hex_value + + if b == 0x0A and not data.isspace(): + hex_value = hex_value + "(\\n)\n" + new_line = True + else: + new_line = False + + results.append(hex_value) + + return "".join(results) + + +_HEX_PAIR = set("0123456789abcdefABCDEF") + + +def parse_hex_bytes(text: str) -> bytes: + """Parse a user-supplied hex string into bytes. + + Accepts common separators/spaces and optional 0x prefixes. + + Examples: + "DE AD BE EF" -> b"\xDE\xAD\xBE\xEF" + "0xDE,0xAD" -> b"\xDE\xAD" + "deadbeef" -> b"\xDE\xAD\xBE\xEF" + """ + + # Strip 0x prefixes and keep only hex digits. + cleaned_chars = [] + i = 0 + while i < len(text): + if text[i : i + 2].lower() == "0x": + i += 2 + continue + ch = text[i] + if ch in _HEX_PAIR: + cleaned_chars.append(ch) + i += 1 + + cleaned = "".join(cleaned_chars) + + if len(cleaned) == 0: + return b"" + + if len(cleaned) % 2 != 0: + raise ValueError("hex input must contain an even number of hex digits") + + return bytes(int(cleaned[i : i + 2], 16) for i in range(0, len(cleaned), 2)) diff --git a/netbyte/netbyte.py b/netbyte/netbyte.py index 54e7686..1873236 100644 --- a/netbyte/netbyte.py +++ b/netbyte/netbyte.py @@ -13,9 +13,11 @@ import argparse import sys import time -from colorama import Fore, Style +from colorama import Fore, Style, init as colorama_init from threading import Thread -from Queue import Queue, Empty +from queue import Queue, Empty + +from netbyte.core import to_hex, parse_hex_bytes def is_symbol(character): @@ -32,66 +34,19 @@ def is_symbol(character): return True -def to_hex(string): - ''' - Converts string to formatted hex and ASCII reference values - - Returns: - str: Formatted hex & ASCII reference - ''' - - results = [] - - new_line = True - - for character in string: - - # Convert ASCII to unicode - unicode_value = ord(character) - - # Convert unicode to hex - hex_value = hex(unicode_value).replace('0x', '') - - # Add a preceding 0 - if len(hex_value) == 1: - hex_value = '0' + hex_value - - # Make upper case - hex_value = hex_value.upper() - - # Add reference ASCII for readability - if character.isalpha() or character.isdigit() or is_symbol(character): - hex_value = hex_value + '(' + character + ')' - - # Add a space in between hex values (not to a new line) - if not new_line: - hex_value = ' ' + hex_value - - # Add a newline for readability (corresponding to ASCII) - if '0A' in hex_value and not string.isspace(): - hex_value = hex_value + '(\\n)\n' - # Next line will be a newline - new_line = True - else: - new_line = False - - results.append(hex_value) - - full_hex = '' - for result in results: - full_hex += result - - return full_hex - - -def print_ascii(string): +def print_ascii(data): ''' Print string with ASCII color configuration ''' - if string.isspace(): + if isinstance(data, (bytes, bytearray)): + text = bytes(data).decode("utf-8", errors="replace") + else: + text = str(data) + + if text.isspace(): # Add a space to show colors on a non-ASCII line - string = ' ' + string - print(Fore.MAGENTA + Style.BRIGHT + string + Style.RESET_ALL) + text = ' ' + text + print(Fore.MAGENTA + Style.BRIGHT + text + Style.RESET_ALL) def print_hex(string): @@ -106,7 +61,7 @@ def print_error(string): Print string with error color configuration and exit with code 1 ''' print(Fore.RED + Style.BRIGHT + string + Style.RESET_ALL) - exit(1) + raise SystemExit(1) class ReadAsync(object): @@ -156,12 +111,33 @@ def parse_arguments(): parser.add_argument('hostname', metavar='HOSTNAME', help='Host or IP to connect to') parser.add_argument('port', metavar='PORT', help='Connection port') parser.add_argument('-u', dest='udp', action="store_true", default=False, help='Use UDP instead of default TCP') + parser.add_argument( + '-l', + '--listen', + dest='listen', + action='store_true', + default=False, + help='Listen instead of connect (TCP: accept one client; UDP: receive from a peer)', + ) + parser.add_argument( + '--bind', + dest='bind', + default=None, + help='Bind address for --listen (defaults to HOSTNAME)', + ) + parser.add_argument( + '--send-hex', + dest='send_hex', + action='store_true', + default=False, + help='Interpret stdin as hex bytes before sending (e.g. \"DE AD BE EF\" or \"0xDE,0xAD\")', + ) if len(sys.argv) == 1: parser.print_help() - exit(1) + raise SystemExit(1) args = parser.parse_args() return args @@ -172,24 +148,52 @@ def main(): Main function: Connects to host/port and spawns ReadAsync ''' + colorama_init() args = parse_arguments() - if args.udp: - connection = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - else: - connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + peer = None - connection.settimeout(2) + if args.listen: + bind_host = args.bind if args.bind is not None else args.hostname + bind_addr = (bind_host, int(args.port)) - address = (args.hostname, int(args.port)) + if args.udp: + connection = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + connection.bind(bind_addr) + except OSError: + print_error(f"Could not bind UDP listener on {bind_addr[0]}:{bind_addr[1]}") + print(Fore.GREEN + Style.BRIGHT + f"Listening (UDP) on {bind_addr[0]}:{bind_addr[1]}" + Style.RESET_ALL) + else: + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + server.bind(bind_addr) + server.listen(1) + except OSError: + print_error(f"Could not bind TCP listener on {bind_addr[0]}:{bind_addr[1]}") + print(Fore.GREEN + Style.BRIGHT + f"Listening (TCP) on {bind_addr[0]}:{bind_addr[1]}" + Style.RESET_ALL) + conn, addr = server.accept() + server.close() + connection = conn + peer = addr + print(Fore.GREEN + Style.BRIGHT + f"Connection established from {addr[0]}:{addr[1]}" + Style.RESET_ALL) + else: + if args.udp: + connection = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + else: + connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - try: - connection.connect(address) + connection.settimeout(2) + + address = (args.hostname, int(args.port)) - except socket.error: - print_error("Could not establish connection to " + address[0] + ":" + str(address[1])) + try: + connection.connect(address) + except OSError: + print_error("Could not establish connection to " + address[0] + ":" + str(address[1])) - print(Fore.GREEN + Style.BRIGHT + "Connection established" + Style.RESET_ALL) + print(Fore.GREEN + Style.BRIGHT + "Connection established" + Style.RESET_ALL) try: connection.setblocking(0) @@ -197,21 +201,54 @@ def main(): while True: try: - data = connection.recv(4096) - if not data: - raise socket.error - print_ascii(data) - print_hex(to_hex(data)) - except socket.error, e: - if e.errno != errno.EWOULDBLOCK: + if args.listen and args.udp: + data, addr = connection.recvfrom(4096) + if data: + peer = addr + print_ascii(data) + print_hex(to_hex(data)) + else: + data = connection.recv(4096) + if not data: + raise OSError + print_ascii(data) + print_hex(to_hex(data)) + except OSError as e: + if getattr(e, "errno", None) not in (errno.EWOULDBLOCK, errno.EAGAIN): raise try: - connection.send(stdin.dequeue()) + outbound = stdin.dequeue() + if outbound == "": + # EOF on stdin + raise KeyboardInterrupt + + if args.send_hex: + try: + outbound_bytes = parse_hex_bytes(outbound.strip()) + except ValueError as e: + print_error(f"Invalid hex input: {e}") + if args.listen and args.udp: + if peer is None: + print_error("No UDP peer yet (send requires receiving at least one datagram)") + connection.sendto(outbound_bytes, peer) + else: + connection.send(outbound_bytes) + else: + if isinstance(outbound, str): + outbound = outbound.encode("utf-8") + if args.listen and args.udp: + if peer is None: + print_error("No UDP peer yet (send requires receiving at least one datagram)") + connection.sendto(outbound, peer) + else: + connection.send(outbound) except Empty: time.sleep(0.1) + except (BlockingIOError, InterruptedError): + time.sleep(0.01) except KeyboardInterrupt: connection.close() print_error("\nExiting...") - except socket.error: + except OSError: print_error("Connection closed") \ No newline at end of file diff --git a/setup.py b/setup.py index 5652a34..71dfc99 100644 --- a/setup.py +++ b/setup.py @@ -13,6 +13,17 @@ install_requires=[ 'colorama', ], + extras_require={ + 'test': ['pytest>=7'], + }, + python_requires='>=3.9', + classifiers=[ + 'Environment :: Console', + 'Intended Audience :: Developers', + 'License :: OSI Approved :: MIT License', + 'Programming Language :: Python :: 3', + 'Topic :: System :: Networking', + ], entry_points = { "console_scripts" : ['netbyte = netbyte.netbyte:main'] }, diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..043d1cc --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,35 @@ +import pytest + +from netbyte.core import parse_hex_bytes, to_hex + + +@pytest.mark.parametrize( + "text,expected", + [ + ("DE AD BE EF", b"\xDE\xAD\xBE\xEF"), + ("de ad be ef", b"\xDE\xAD\xBE\xEF"), + ("0xDE,0xAD,0xBE,0xEF", b"\xDE\xAD\xBE\xEF"), + ("deadbeef", b"\xDE\xAD\xBE\xEF"), + ("\n\t ", b""), + ], +) +def test_parse_hex_bytes(text, expected): + assert parse_hex_bytes(text) == expected + + +def test_parse_hex_bytes_odd_length_rejected(): + with pytest.raises(ValueError, match="even number"): + parse_hex_bytes("ABC") + + +def test_to_hex_includes_printable_references_and_newline_marker(): + out = to_hex(b"Hi\n") + # Printable bytes show references, newline shows (\n) marker. + assert "48(H)" in out + assert "69(i)" in out + assert "0A(\\n)" in out + assert out.endswith("\n") + + +def test_parse_hex_bytes_ignores_non_hex_separators(): + assert parse_hex_bytes("DE-AD:BE_EF") == b"\xDE\xAD\xBE\xEF" diff --git a/testserver.py b/testserver.py index 4d99fb4..f88b416 100644 --- a/testserver.py +++ b/testserver.py @@ -1,4 +1,4 @@ -#!/usr/bin/python +#!/usr/bin/env python3 # Author: sc0tfree # Twitter: @sc0tfree @@ -12,9 +12,11 @@ def generate_random_hex(length): ''' Generates a hex string of arbitrary length - 1, ending in a newline. ''' - hex_string = os.urandom(length - 1) - hex_string += '\x0a' - return hex_string + if length <= 0: + return b"" + hex_bytes = os.urandom(max(0, length - 1)) + hex_bytes += b"\x0a" + return hex_bytes host = '127.0.0.1' @@ -32,35 +34,35 @@ def generate_random_hex(length): c, addr = s.accept() - print 'Connection established from', addr[0], ':', addr[1] + print('Connection established from', addr[0], ':', addr[1]) - c.send('Hello from Test Server\n') + c.send(b'Hello from Test Server\n') # Echo Test - c.send('Echo Test - enter string:') + c.send(b'Echo Test - enter string:') data = c.recv(1024) - print 'Echo Test - received: ', data - c.send('Echo Test - received: ' + data + '\n') + print('Echo Test - received: ', data) + c.send(b'Echo Test - received: ' + data + b'\n') # Hex Test - c.send('Hex Test - enter length:') + c.send(b'Hex Test - enter length:') data = c.recv(1024) try: - hex_length = int(data) + hex_length = int(data.strip() or b"0") except ValueError: - c.send('You must enter a number. Defaulting to 10.\n') + c.send(b'You must enter a number. Defaulting to 10.\n') hex_length = 10 hex_string = generate_random_hex(hex_length) - c.send('Sending hex string...\n\n') - print 'Hex Test - sending: ', hex_string + c.send(b'Sending hex string...\n\n') + print('Hex Test - sending: ', hex_string) c.send(hex_string) c.close() - print 'Closed connection to ', addr[0], ':', addr[1] + print('Closed connection to ', addr[0], ':', addr[1]) except KeyboardInterrupt: c.close() - print '\nExiting...' - exit(0) + print('\nExiting...') + raise SystemExit(0)