220 lines
8.2 KiB
Python
220 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Test which TCP ports on an authorized host respond to the DragonTCP BP probe."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import concurrent.futures
|
|
import csv
|
|
import hashlib
|
|
import socket
|
|
import struct
|
|
import sys
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
|
|
MODE_PROBE = 0
|
|
STATUS_OK = 0
|
|
REQUEST_HEADER_SIZE = 29
|
|
RESPONSE_HEADER_SIZE = 5
|
|
MAX_RESPONSE_BODY = 2 * 1024 * 1024
|
|
BP_PROBE = b"BHP1\x01\x00\x00\x00\x00\x00"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Result:
|
|
port: int
|
|
state: str
|
|
elapsed_ms: int
|
|
detail: str = ""
|
|
|
|
|
|
def sha256_ctr_mask(
|
|
data: bytes,
|
|
session_id: bytes,
|
|
mode: int,
|
|
sequence: int,
|
|
is_response: bool,
|
|
) -> bytes:
|
|
"""Apply the BP payload mask. Calling this twice restores the input."""
|
|
if not data:
|
|
return b""
|
|
sid = session_id[:16].ljust(16, b"\x00")
|
|
seed = sid + bytes((mode & 0xFF,)) + struct.pack(">Q", sequence)
|
|
seed += bytes((1 if is_response else 0,))
|
|
out = bytearray(len(data))
|
|
for offset in range(0, len(data), 32):
|
|
counter = offset // 32
|
|
block = hashlib.sha256(seed + struct.pack(">I", counter)).digest()
|
|
count = min(32, len(data) - offset)
|
|
for index in range(count):
|
|
out[offset + index] = data[offset + index] ^ block[index]
|
|
return bytes(out)
|
|
|
|
|
|
def read_exact(sock: socket.socket, size: int) -> bytes:
|
|
data = bytearray()
|
|
while len(data) < size:
|
|
chunk = sock.recv(size - len(data))
|
|
if not chunk:
|
|
raise EOFError(f"EOF after {len(data)}/{size} bytes")
|
|
data.extend(chunk)
|
|
return bytes(data)
|
|
|
|
|
|
def test_port(host: str, port: int, timeout: float) -> Result:
|
|
started = time.monotonic()
|
|
session_id = uuid.uuid4().bytes
|
|
encrypted = sha256_ctr_mask(BP_PROBE, session_id, MODE_PROBE, 0, False)
|
|
request = struct.pack(">B16sQI", MODE_PROBE, session_id, 0, len(encrypted)) + encrypted
|
|
|
|
try:
|
|
with socket.create_connection((host, port), timeout=timeout) as sock:
|
|
sock.settimeout(timeout)
|
|
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
|
|
sock.sendall(request)
|
|
header = read_exact(sock, RESPONSE_HEADER_SIZE)
|
|
status, body_size = struct.unpack(">BI", header)
|
|
if body_size > MAX_RESPONSE_BODY:
|
|
raise ValueError(f"response body too large: {body_size}")
|
|
body = read_exact(sock, body_size) if body_size else b""
|
|
decoded = sha256_ctr_mask(body, session_id, MODE_PROBE, 0, True)
|
|
elapsed = int((time.monotonic() - started) * 1000)
|
|
if status == STATUS_OK and decoded == BP_PROBE:
|
|
return Result(port, "bp", elapsed, "valid BP probe echo")
|
|
return Result(
|
|
port,
|
|
"open",
|
|
elapsed,
|
|
f"non-BP response status={status} body={decoded[:16].hex()}",
|
|
)
|
|
except (ConnectionRefusedError, TimeoutError, socket.timeout):
|
|
return Result(port, "closed", int((time.monotonic() - started) * 1000))
|
|
except OSError as exc:
|
|
return Result(
|
|
port,
|
|
"closed",
|
|
int((time.monotonic() - started) * 1000),
|
|
str(exc),
|
|
)
|
|
except Exception as exc: # A TCP service answered, but not with a valid BP frame.
|
|
return Result(
|
|
port,
|
|
"open",
|
|
int((time.monotonic() - started) * 1000),
|
|
str(exc),
|
|
)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description=(
|
|
"Test TCP ports for a valid BP probe response. Only scan hosts you own "
|
|
"or have explicit permission to test."
|
|
)
|
|
)
|
|
parser.add_argument("--host", required=True, help="authorized IPv4, IPv6, or hostname")
|
|
parser.add_argument("--start-port", type=int, default=1, help="first port (default: 1)")
|
|
parser.add_argument("--end-port", type=int, default=65535, help="last port (default: 65535)")
|
|
parser.add_argument("--threads", type=int, default=1, help="maximum concurrent probes, 1-64 (default: 1)")
|
|
parser.add_argument(
|
|
"--delay-ms",
|
|
type=int,
|
|
default=200,
|
|
help="global delay between probe starts, 0-30000 ms (default: 200)",
|
|
)
|
|
parser.add_argument("--timeout", type=float, default=2.5, help="per-port timeout in seconds (default: 2.5)")
|
|
parser.add_argument("--show-open", action="store_true", help="also print open ports that do not speak BP")
|
|
parser.add_argument("--progress-every", type=int, default=1000, help="progress interval; 0 disables")
|
|
parser.add_argument("--output", type=Path, help="optional CSV output for BP and other open ports")
|
|
args = parser.parse_args()
|
|
|
|
if not 1 <= args.start_port <= 65535:
|
|
parser.error("--start-port must be between 1 and 65535")
|
|
if not 1 <= args.end_port <= 65535:
|
|
parser.error("--end-port must be between 1 and 65535")
|
|
if args.start_port > args.end_port:
|
|
parser.error("--start-port must not exceed --end-port")
|
|
if not 1 <= args.threads <= 64:
|
|
parser.error("--threads must be between 1 and 64")
|
|
if not 0 <= args.delay_ms <= 30000:
|
|
parser.error("--delay-ms must be between 0 and 30000")
|
|
if not 0.05 <= args.timeout <= 120:
|
|
parser.error("--timeout must be between 0.05 and 120 seconds")
|
|
if args.progress_every < 0:
|
|
parser.error("--progress-every must be 0 or greater")
|
|
return args
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
total = args.end_port - args.start_port + 1
|
|
delay = args.delay_ms / 1000.0
|
|
completed = 0
|
|
results: list[Result] = []
|
|
started = time.monotonic()
|
|
print(
|
|
f"BP scan host={args.host} ports={args.start_port}-{args.end_port} "
|
|
f"threads={args.threads} delay={args.delay_ms}ms timeout={args.timeout:g}s"
|
|
)
|
|
|
|
print_lock = threading.Lock()
|
|
|
|
def consume(result: Result) -> None:
|
|
nonlocal completed
|
|
completed += 1
|
|
if result.state != "closed":
|
|
results.append(result)
|
|
with print_lock:
|
|
if result.state == "bp":
|
|
print(f"BP {args.host}:{result.port} {result.elapsed_ms}ms")
|
|
elif result.state == "open" and args.show_open:
|
|
suffix = f" {result.detail}" if result.detail else ""
|
|
print(f"OPEN {args.host}:{result.port} {result.elapsed_ms}ms{suffix}")
|
|
if args.progress_every and completed % args.progress_every == 0:
|
|
elapsed = time.monotonic() - started
|
|
print(f"progress {completed}/{total} elapsed={elapsed:.1f}s")
|
|
|
|
pending: set[concurrent.futures.Future[Result]] = set()
|
|
try:
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=args.threads) as pool:
|
|
for port in range(args.start_port, args.end_port + 1):
|
|
while len(pending) >= args.threads:
|
|
done, pending = concurrent.futures.wait(
|
|
pending,
|
|
return_when=concurrent.futures.FIRST_COMPLETED,
|
|
)
|
|
for future in done:
|
|
consume(future.result())
|
|
pending.add(pool.submit(test_port, args.host, port, args.timeout))
|
|
if delay:
|
|
time.sleep(delay)
|
|
for future in concurrent.futures.as_completed(pending):
|
|
consume(future.result())
|
|
except KeyboardInterrupt:
|
|
print("\nInterrupted; partial results follow.", file=sys.stderr)
|
|
|
|
results.sort(key=lambda item: item.port)
|
|
bp_ports = [item.port for item in results if item.state == "bp"]
|
|
elapsed = time.monotonic() - started
|
|
print(f"completed={completed}/{total} elapsed={elapsed:.1f}s")
|
|
print("BP ports: " + (", ".join(map(str, bp_ports)) if bp_ports else "none"))
|
|
|
|
if args.output:
|
|
with args.output.open("w", newline="", encoding="utf-8") as handle:
|
|
writer = csv.writer(handle)
|
|
writer.writerow(("host", "port", "state", "elapsed_ms", "detail"))
|
|
for result in results:
|
|
writer.writerow((args.host, result.port, result.state, result.elapsed_ms, result.detail))
|
|
print(f"wrote {args.output}")
|
|
|
|
return 0 if bp_ports else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|