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
okmeans this TCP connection was established.refusedmeans the system reported a refusal, commonly because no service is listening; an intermediary can also actively reject it.timeoutmeans the connection operation timed out. It does not identify a firewall as the cause.errorpreserves 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.
- 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.
- In terminal B, enter the same
tcp-probe-demodirectory. If downloads are restricted, save the complete implementation below ascheck_tcp.pyand the section 5 code astest_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
- 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}
- 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
- 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.
- On the Linux host or container running the service, use an existing
sscommand 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.
- 原文作者:春江暮客
- 原文链接:https://www.bobobk.com/en/python-tcp-port-check.html
- 版权声明:本作品采用 知识共享署名-非商业性使用-禁止演绎 4.0 国际许可协议 进行许可,非商业转载请注明出处(作者,原文链接),商业转载请联系作者获得授权。
相关文章
- Check ZIP Backups with Python: CRC, Duplicate Members, and Missing Files
- Check Disk Space with Python Before Starting a Batch Job
- Find Broken Symlinks with Python: Separate Missing Targets, Permission Errors, and Loops
- Fix Python SQLite database is locked: Timeouts and Short Transactions
- Check Scheduled Output Freshness with Python: File Age, Empty Files, and Exit Codes