春江暮客

春江暮客的个人学习分享网站

Probe TCP Ports with Python: Distinguish Refusal, Timeout, and System Errors

2026-10-18 Technology
Probe TCP Ports with Python: Distinguish Refusal, Timeout, and System Errors

A background script cannot reach a local service, but its log only says “connection failed.” Is the port refusing connections, did the attempt time out, or did another system error occur? This tutorial builds a small Python probe that emits JSON and explicit exit codes, then reproduces success and failure locally.

The examples suit beginner-to-intermediate readers and use only the Python standard library. Each run probes one IPv4 address and TCP port, sends no application data, and performs no automatic retry.

1. Method: define the scope

  1. ok means this TCP connection was established.
  2. refused means the system reported a refusal, commonly because no service is listening; an intermediary can also actively reject it.
  3. timeout means the connection operation timed out. It does not identify a firewall as the cause.
  4. error preserves other system failures, such as an unreachable network or a permission restriction.

We deliberately validate a numeric IPv4 address with IPv4Address. The tool rejects localhost, domain names, IPv6, and URLs containing http://, keeping DNS resolution outside this experiment. Use the actual destination address; 0.0.0.0 is commonly a server bind address and should not be used as a remote service destination.

After TCP succeeds, check the application protocol. For a website deployment, continue with curl status and page-content checks; a connectable port alone does not confirm a successful release.

2. Setup: download the scripts and start a local service

Use Linux, Python 3.9 or newer, and curl for downloading. These examples were tested on Linux with Python 3.11.4. Commands work in bash or zsh and change no firewall or scheduler settings.

  1. In terminal A, run these commands in a fresh directory:
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

The demo service listens only on loopback and serves only the new empty directory. Python’s http.server --bind and --directory documentation describes these options; we use it only as a local test service. Leave terminal A running, and press Ctrl+C when finished.

  1. In terminal B, enter the same tcp-probe-demo directory. If downloads are restricted, save the complete implementation below as check_tcp.py and the section 5 code as test_tcp.py, skipping both curl commands.

If the server reports Address already in use, choose 8766 and replace 8765 in all subsequent commands too. There is no need to terminate another program occupying the port.

3. Complete implementation: one connection, one JSON line

"""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())

socket.settimeout() controls the connection operation’s wait. The argument must be finite, greater than 0, and no more than 60 seconds; it is not a hard wall-clock deadline for the whole process. The context closes the socket after the probe.

Elapsed time uses time.monotonic(), avoiding system clock adjustments. elapsed_ms is an observation rounded to three decimal places. The system network stack can also return a timeout earlier.

4. Usage: success, refusal, and exit codes

  1. Keep the service in terminal A running and execute this in terminal B:
python3 check_tcp.py 127.0.0.1 8765 --timeout 1

Output resembles this; elapsed time varies:

{"ip": "127.0.0.1", "port": 8765, "status": "ok", "errno": null, "error": "", "timeout_seconds": 1.0, "elapsed_ms": 0.21}
  1. Press Ctrl+C in terminal A to stop the demo service you started, then return to terminal B:
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

If no other program has taken the port, expect status: "refused" and Probe failed: exit=1. Error numbers and messages come from the system; do not hardcode a platform’s numeric errno or English message when classifying failures.

Exit code Meaning
0 This TCP connection succeeded.
1 Refusal, timeout, or another system error; stdout still contains JSON.
2 Invalid arguments; the explanation goes to stderr, with no success JSON.

A refusal and a timeout are different observations. Neither completely establishes the remote application’s state. The next section verifies classification with controlled experiments.

5. Validation: dynamic ports and injected timeouts

Save this as test_tcp.py in the current directory. The downloadable version contains the same code:

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")

Run:

python3 test_tcp.py

Expected output:

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

The first two tests use real local sockets: one listens, while the other binds without listening. The system allocates each port dynamically, and the fixture keeps it reserved throughout the test, avoiding the race between finding a free port and using it. The listening fixture handles no HTTP requests yet returns ok, demonstrating that TCP success does not establish application health.

Timeout and unreachable-network failures use unittest.mock.patch to inject exceptions. They verify classification and context exit, not real network timeout timing. No external target or firewall adjustment is needed.

6. Automation and Linux troubleshooting

  1. Restart the section 2 demo service, then use && in terminal B to gate the next check:
python3 check_tcp.py 127.0.0.1 8765 --timeout 1 \
  && printf 'Continue with an application-level check\n'

The next command runs only after a successful probe; a failed probe preserves a nonzero exit code. In production, replace the message with a real application check. For JSON endpoints, follow Python HTTP JSON requests and bounded retries. Success is an observation at one moment; a later connection can still fail, so application clients need their own error handling.

  1. On the Linux host or container running the service, use an existing ss command to inspect listeners:
ss -ltnp 'sport = :8765'

-ltnp selects listening sockets, TCP, numeric addresses, and process information. Permissions may hide another user’s process details. If ss is absent, the Python probe and self-check still work. Do not blindly open ports or stop existing services just to make a probe pass.

Symptom Concrete next check
refused Confirm the destination and port, the service listener, and any active rejection rules.
timeout Compare the listener, client route, and network access rules; confirm the target before extending the timeout.
error Read errno and error in the JSON and address the specific network or permission failure.
Works locally, fails inside a container 127.0.0.1 inside a container points to that container; use a destination reachable from the actual caller.
TCP works, HTTP fails Check TLS, status codes, and response content; the probe reads none of them.
Works interactively, fails on a schedule Use absolute interpreter and script paths, then follow scheduled-job path and environment troubleshooting to check the account and working directory.

7. Summary

Separating success, refusal, timeout, and other errors gives automation useful failure details and a clear exit code for subsequent steps. Validate the tool with local dynamic ports and injected exceptions, then probe from the actual caller’s environment. Once TCP connects, continue with application responses and business-content checks.

标签

1024 ablang adsense agents.md ai ai-agent ai-agents ai-seo algorithm amp antibodies apparmor argparse automation background backup bash batch-processing binarycif bioinformatics biopython blockchain boltz bootstrapping boxes bubblewrap c-index cca cdn checkpoint cli cloudflare codex cofoldarena copy cpu monitoring cron csv cuda curl data-leakage data-processing data-quality data-validation datascience datavisualization datetime debugging decimal deployment desktop-app devtools disk-space disown dns docker dovecot download economics electron encoding esm esm2 esm3 esmc esmfold2 fable faceswap fasta fastmcp faulthandler ffmpeg file-descriptors file-io flashppi flask folium frontend game generator git github-actions google google-research grep gzip harness hls html http hugo imports indexnow javascript jev json jsonl just k-means kaggle langfuse leecode linux list litellm llm llms.txt logging logs lollipop lsof m3u8 machine-learning macos manacher matplotlib mcp mirror mmcif model-evaluation monitoring mp3 mp4 mpnn multiomics mutation mysql nanobert nanobodies networkx nginx normalize numpy ollama omegatherm pandas password pathlib pdb pep-723 phaser pillow pip postfix preprocessing print probability protein-alignment protein-design protein-embeddings protein-interactions protein-language-models protein-stability protein-structure proxy pydantic pyecharts pyqt python python3 r raincloud reproducibility requests reservoir-sampling rfoptimization rg ripgrep rosettafold3 roundcube rrsi rsi rsync s-tui sampling scale scikit-learn scrapy screen seaborn security seo sequence-identity sha256 shell signals sklearn socket solana somaticsignatures spl sqlite ssh standardize static-site subprocess symlink sysadmin system-one tarfile tcp tensorflow tkinter tls toml trade tron tronpy troubleshooting turtle typesafe-ai ubuntu unicode usdt utf-8 uv vhhbert vite wallet webp wordcloud wordpress workflow yaml zipfile zoneinfo 迅雷解析

友情链接

其它