用 Python 探测 TCP 端口：区分连接拒绝、超时与系统错误

python3 --version
mkdir tcp-probe-demo
cd tcp-probe-demo
curl -fS https://www.bobobk.com/downloads/tcp-probe/check_tcp.py -o check_tcp.py
curl -fS https://www.bobobk.com/downloads/tcp-probe/test_tcp.py -o test_tcp.py
python3 check_tcp.py --help
mkdir served
python3 -m http.server 8765 --bind 127.0.0.1 --directory served

"""Probe one numeric IPv4 TCP endpoint. Python 3.9+, standard library."""
import argparse
import errno
import ipaddress
import json
import math
import socket
import time


def ipv4(value):
    try:
        return str(ipaddress.IPv4Address(value))
    except ipaddress.AddressValueError as exc:
        raise argparse.ArgumentTypeError("use a numeric IPv4 address") from exc


def probe(ip, port, timeout):
    result = {"ip": ip, "port": port, "status": "ok", "errno": None,
              "error": "", "timeout_seconds": timeout}
    started = time.monotonic()
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as connection:
            connection.settimeout(timeout)
            connection.connect((ip, port))
    except socket.timeout as exc:
        result.update(status="timeout", errno=exc.errno, error=str(exc))
    except OSError as exc:
        status = "refused" if exc.errno == errno.ECONNREFUSED else "error"
        result.update(status=status, errno=exc.errno, error=str(exc))
    result["elapsed_ms"] = round((time.monotonic() - started) * 1000, 3)
    return result


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("ip", type=ipv4)
    parser.add_argument("port", type=int)
    parser.add_argument("--timeout", type=float, default=2.0,
                        help="connection timeout in seconds, 0 < value <= 60")
    args = parser.parse_args()
    if not 1 <= args.port <= 65535:
        parser.error("port must be between 1 and 65535")
    if not math.isfinite(args.timeout) or not 0 < args.timeout <= 60:
        parser.error("--timeout must be finite and between 0 (exclusive) and 60")
    report = probe(args.ip, args.port, args.timeout)
    print(json.dumps(report, ensure_ascii=False))
    return 0 if report["status"] == "ok" else 1


if __name__ == "__main__":
    raise SystemExit(main())

python3 check_tcp.py 127.0.0.1 8765 --timeout 1

{"ip": "127.0.0.1", "port": 8765, "status": "ok", "errno": null, "error": "", "timeout_seconds": 1.0, "elapsed_ms": 0.21}

if python3 check_tcp.py 127.0.0.1 8765 --timeout 1; then
  printf 'TCP connected\n'
else
  probe_exit=$?
  printf 'Probe failed: exit=%s\n' "$probe_exit"
fi

import errno
import json
import socket
import subprocess
import sys
from unittest.mock import patch
import check_tcp


def run(arguments):
    return subprocess.run(
        [sys.executable, "check_tcp.py", *arguments],
        capture_output=True, text=True, timeout=5, check=False,
    )


# Port 0 asks Linux for an available port; keep the socket open during each test.
for listening, expected in [(True, "ok"), (False, "refused")]:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as fixture:
        fixture.bind(("127.0.0.1", 0))
        if listening:
            fixture.listen(1)
        port = fixture.getsockname()[1]
        completed = run(["127.0.0.1", str(port), "--timeout", "0.5"])
        report = json.loads(completed.stdout)
        assert report["status"] == expected, report
        assert completed.returncode == (0 if listening else 1), completed
        assert completed.stderr == "", completed
        assert report["ip"] == "127.0.0.1" and report["port"] == port
        assert report["elapsed_ms"] >= 0
        if not listening:
            assert report["errno"] == errno.ECONNREFUSED, report

for arguments in [
    ["localhost", "80"], ["::1", "80"], ["127.0.0.1", "0"],
    ["127.0.0.1", "65536"], ["127.0.0.1", "80", "--timeout", "0"],
    ["127.0.0.1", "80", "--timeout", "nan"],
    ["127.0.0.1", "80", "--timeout", "inf"],
    ["127.0.0.1", "80", "--timeout", "61"],
]:
    completed = run(arguments)
    assert completed.returncode == 2 and not completed.stdout, completed
    assert completed.stderr, completed

# Inject failures: no firewall changes, external packets, or real timeout wait.
for failure, expected in [
    (socket.timeout("test timeout"), "timeout"),
    (OSError(errno.ENETUNREACH, "test unreachable"), "error"),
]:
    with patch("check_tcp.socket.socket") as factory:
        connection = factory.return_value.__enter__.return_value
        connection.connect.side_effect = failure
        report = check_tcp.probe("127.0.0.1", 12345, 0.5)
        assert report["status"] == expected, report
        connection.settimeout.assert_called_once_with(0.5)
        connection.connect.assert_called_once_with(("127.0.0.1", 12345))
        factory.return_value.__exit__.assert_called_once()

print("OK: listening, refused, invalid arguments, injected timeout and network error")

python3 test_tcp.py

OK: listening, refused, invalid arguments, injected timeout and network error

python3 check_tcp.py 127.0.0.1 8765 --timeout 1 \
  && printf 'Continue with an application-level check\n'

ss -ltnp 'sport = :8765'