advanced-connection-test

advanced-connection-test

advanced-connection-test is a Python package for advanced network connectivity diagnostics. It allows you to accurately distinguish between various network states — no connection, LAN only, captive portal, mandatory proxy, SSL errors, and working connection — in a robust, safe, and transparent way.

Purpose

Provide a reliable tool to:

• Diagnose network issues in corporate, public, or home environments.
• Quickly identify the cause of a failed Internet connection.
• Adapt an application's behavior based on the actual connectivity status.
• Detect mandatory proxies, captive portals, and outdated configurations.

Installation

pip install advanced-connection-test

Or, for local development with test dependencies:

git clone https://github.com/raffaellof/connection-test.git
cd connection-test
pip install -e ".[dev]"

Requirements: Python 3.7+, aiohttp >= 3.8.0

---

Quick Usage

import asyncio
from connection_test import enhanced_connection_test, ConnectionStatus

result = asyncio.run(enhanced_connection_test())

if result.status == ConnectionStatus.CONNECTED_DIRECT:
    print(f"Online in {result.test_duration_ms}ms")
elif result.status == ConnectionStatus.CAPTIVE_PORTAL:
    print(f"Captive portal detected: {result.captive_portal_url}")
elif result.status == ConnectionStatus.PROXY_REQUIRED:
    print(f"Proxy required: {result.detected_proxy_url}")
elif result.status == ConnectionStatus.PROXY_AUTH_FAILED:
    print("Proxy credentials incorrect or missing")
elif result.status == ConnectionStatus.SSL_ERROR:
    print("SSL error — check system date/time")
elif result.status == ConnectionStatus.LAN_ONLY:
    print("Local network OK, but Internet unreachable")
elif result.status == ConnectionStatus.NO_CONNECTION:
    print("No network detected")

With Custom URLs

Useful on networks where proxies block some sites but not others: pass the critical URLs for your application instead of relying on the default list.

from connection_test import enhanced_connection_test, ConnectionTestConfig

config = ConnectionTestConfig(
    test_urls=["https://my-api.company.com", "https://www.google.com"],
    timeout=10,
    global_timeout=30,
)
result = asyncio.run(enhanced_connection_test(config=config))
print(result.status.value, result.message)

Diagnostic Mode

Tests all URLs in the list (instead of exiting on the first success) and returns details for each:

result = asyncio.run(enhanced_connection_test(test_all_urls=True))
for url_result in result.details.get("results_per_url", []):
    print(f"{url_result['url']}: {'OK' if url_result['success'] else 'FAIL'}")

---

Possible States (ConnectionStatus)

Note: detected_proxy_url in the results always contains the proxy URL with masked credentials (http://***@host:port), never in clear text.

---

Architecture — Test Phases

The test runs up to 6 sequential phases (Phase 0–5) with early-exit on the first conclusive result.

---

Phase 0 — Proxy Variable Pre-check

What it does: Before starting any network test, it reads the environment variables HTTP_PROXY, HTTPS_PROXY (and lowercase variants) to know if a proxy is already configured.

How: Reads os.getenv() and computes safe_proxy_url (masked URL) for logging. Does not perform any network request. The collected information is used in phases 3 and 4.

Why: Separating env var reading from execution ensures safe_proxy_url is always available for logs even in case of global timeout (the partial_state is updated immediately).

Outcome: None — preparatory phase, always proceeds.

---

Phase 1 — Connectivity Baseline (TCP Socket + DNS)

What it does: First checks if at least one network interface is active by attempting a TCP connection to 8.8.8.8:53 (Google's public DNS server), then validates DNS resolution.

How:

1. Creates a SOCK_STREAM (TCP) socket with a 1-second timeout.
2. Resolves 3 public domains (www.google.com, github.com, cloudflare.com) with a 2-second timeout per domain.
3. Requires at least 2 resolutions with public IPs (not RFC 1918, not loopback, not link-local).

Why: This quickly separates transport-level failure from Internet reachability. Socket failure indicates physical/link issues (unplugged cable, Wi-Fi off, broken driver). DNS failure with socket success indicates LAN-only/split-horizon conditions. UDP (SOCK_DGRAM) is not used because socket.connect() with UDP does not send data nor verify reachability, always returning success even without a network.

Outcomes: NO_CONNECTION (socket fail) or LAN_ONLY (DNS fail).

---

Phase 2 — Direct HTTP

What it does: Performs HTTPS requests to the configured URLs without a proxy, checking for 2xx status and domain match between requested URL and final response URL.

How: Uses aiohttp.ClientSession without trust_env=True (default behavior), so proxy env vars are ignored by aiohttp. unset_proxy_env_async() is kept as a defensive safeguard. Supports two modes: performance (early-exit on first success) and diagnostic (test_all_urls=True, tests all URLs).

Why: Verifies real application connectivity. Domain match detects cross-domain redirects typical of captive portals. Separate SSL error counting allows distinguishing SSL_ERROR from other failures. If direct connectivity works while a proxy is configured in env vars, that proxy configuration is considered stale.

Outcomes: CONNECTED_DIRECT (success without configured proxy), PROXY_STALE (direct works with configured proxy), SSL_ERROR (all attempted URLs returned SSL errors — in performance mode, only the actually attempted URLs are considered), or proceeds to the next phase.

---

Phase 3 — Configured Proxy

What it does: If HTTP_PROXY/HTTPS_PROXY are configured, tests that proxy explicitly.

How: Sends HTTPS requests through the configured proxy (proxy=... in aiohttp request).

• HTTP 407 returns PROXY_AUTH_FAILED.
• Successful proxy requests return CONNECTED_PROXY.
• If proxy fails and direct re-test succeeds, returns PROXY_STALE.
• If both proxy and direct re-test fail, captive portal detection is executed:
  • captive portal detected → CAPTIVE_PORTAL_PROXY
  • captive portal not detected → PROXY_REQUIRED

Why: This phase distinguishes invalid credentials, stale proxy settings, proxy-dependent networks, and captive portal scenarios in proxy-configured environments.

Outcomes: CONNECTED_PROXY, PROXY_AUTH_FAILED, PROXY_STALE, CAPTIVE_PORTAL_PROXY, PROXY_REQUIRED.

---

Phase 4 — Proxy Scan

What it does: If no configured proxy solved the issue, scans local ports 8080, 3128, and 8888 looking for an undeclared proxy.

How: Uses asyncio.open_connection() with a 0.5s timeout per port. Each open port is validated with a real HTTP request through it.

• HTTP 407 from detected proxy returns PROXY_AUTH_FAILED.
• Working detected proxy returns PROXY_REQUIRED.
• Non-proxy/open unrelated ports are ignored.

Outcome: PROXY_AUTH_FAILED, PROXY_REQUIRED, or proceeds.

---

Phase 5 — Captive Portal

What it does: Queries 3 dedicated HTTP endpoints to detect the presence of a captive portal using majority vote.

How: Sends HTTP requests (not HTTPS, deliberately interceptable) to:

• Google: connectivitycheck.gstatic.com/generate_204 → expected HTTP 204
• Microsoft: msftconnecttest.com/connecttest.txt → expected body "Microsoft Connect Test"
• Firefox: detectportal.firefox.com/success.txt → expected body "success"

If ≥50% of conclusive tests indicate interception, the captive portal is confirmed.

Why: A single endpoint may be temporarily unreachable (CDN down, corporate firewall) causing false positives. Three independent vendors with majority vote drastically reduce this possibility. Requests use HTTP because captive portals only intercept cleartext traffic — HTTPS cannot be altered without the certificate revealing the interception.

Outcome: CAPTIVE_PORTAL if confirmed, UNKNOWN_ERROR as final fallback.

---

Security Features

• No credentials in logs — all proxy URLs are masked via _mask_proxy_credentials() before any logging output.
• SSL certificate verification enabled — all HTTPS requests use certificate verification by default.
• Timeout on every operation — socket (1s), DNS (2s/domain), HTTP (5s, configurable), global (60s, configurable). No operation can hang.
• Async lock on os.environ — prevents race conditions in concurrent contexts that simultaneously modify system proxy variables.

---

Reference API

enhanced_connection_test()

async def enhanced_connection_test(
    config: Optional[ConnectionTestConfig] = None,
    test_urls: Optional[List[str]] = None,
    timeout: int = 5,
    test_all_urls: bool = False,
    global_timeout: int = 60,
) -> ConnectionTestResult

ConnectionTestConfig

from connection_test import ConnectionTestConfig

config = ConnectionTestConfig(
    test_urls=["https://example.com"],
    timeout=10,
    test_all_urls=False,
    global_timeout=30,
)

ConnectionTestResult

The result contains:

• status — value of ConnectionStatus
• message — description in English for the end user
• details — dictionary with technical information (tested URL, duration, error type, etc.)
• requires_action — True if user action is required
• suggested_route — suggested path (e.g., '/proxy_login', '/auth/captive_portal')
• detected_proxy_url — detected proxy URL (masked credentials)
• captive_portal_url — intercepted captive portal URL
• test_duration_ms — total duration in milliseconds

---

Glossary

• LAN (Local Area Network): Local network, typically limited to a building or office.
• Captive portal: System that blocks Internet access until the user authenticates via a dedicated web page (common in hotels, airports, universities).
• Authenticated proxy: Proxy requiring username and password for access (HTTP 407).
• Transparent proxy: Proxy that intercepts traffic without the client being configured to use it.
• Split-horizon DNS: DNS configuration that returns different responses based on the network origin of the query (internal vs. external).
• Majority vote: Consensus technique requiring agreement from at least half of participants to make a decision — used for captive portal detection.
• Timeout: Maximum wait time for a response before considering the operation failed.

---

License

MIT — see LICENSE