"""Check literal HTTP(S) URLs in repository files. Scans git-tracked and untracked-but-not-ignored text files for literal ``http://`` and ``https://`` URLs. Placeholder/template URLs and reserved local/example hosts are skipped, as are ``.onion`` addresses, which no clearnet probe can reach without a Tor SOCKS proxy this suite does not use. Per-line suppression uses the unified ``# nocheck: url`` marker. See ``docs/contributing/actions/testing/suppression.md``. This is an external test because it performs live HTTP requests against the referenced third-party URLs. HTTP ``401`` (Unauthorized), ``403`` (Forbidden), ``405`` (Method Not Allowed), ``406`` (Not Acceptable) and ``415`` (Unsupported Media Type) are treated as reachable (server is alive but auth-gated, method-restricted, or rejecting the probe's headers). HTTP ``418`` (I'm a teapot), ``429`` (Too Many Requests), ``451`` (Unavailable For Legal Reasons) and every ``5xx`` server response emit warning annotations rather than failing: the server answered, so the URL exists. All other ``4xx`` codes fail once a second probe after a pause answers the same way. A timeout or connection error is a third outcome, reported as an unverified warning: the server never answered, so the URL was not checked at all. Such a probe is indeterminate rather than negative -- a host may be geo-blocked, be a rendered template placeholder, or simply be firewalled off from this runner. That tolerance is what let a sandbox with no route to the registry report a green run over 244 unanswered probes, among them 110 lockfile URLs that were in fact HTTP 404. The guard against it is ``_CONNECTIVITY_CANARIES``: before any probing, the suite checks the infrastructure hosts the repository actually depends on, and fails outright when they are unreachable. A run that cannot reach its subject does not pass; it does not run. Probing runs ``INFINITO_WORKER_FETCH`` requests wide, and nothing stops that pool from aiming most of them at one host, so ``_MAX_REQUESTS_PER_HOST`` bounds how many requests may be in flight against one hostname and leaves total parallelism alone. It sits below the six connections per host Chrome and Firefox allow, because every probe here opens a fresh connection instead of reusing a kept-alive one. The cost is paid by whichever host holds the most URLs, which becomes the pass's critical path. What this suite measures is not always the upstream. Inside the test container the package cache pins many of these hosts in ``/etc/hosts``, so a probe reaches a local Nexus proxy that answers in the upstream's name. Run 38482429514 failed on a ``huggingface.co`` model URL reported as HTTP 404 while the file was live: the proxy had rejected it for having no detectable MIME type. Both false negatives and false positives are possible for a pinned host, and ``test_apt_ubuntu_mirrors._pinned_hosts`` is the existing way to tell which ones those are. """ from __future__ import annotations import concurrent.futures import ipaddress import os import random import re import subprocess import sys import threading import time import unittest import unittest.mock from collections import defaultdict from pathlib import Path from typing import NamedTuple from urllib.parse import urlsplit, urlunsplit import requests from utils.annotations.message import error, warning from utils.annotations.suppress import is_suppressed_at from utils.cache.files import iter_non_ignored_files, read_text from . import PROJECT_ROOT from ._host_gate import host_gate _REPO_ROOT = PROJECT_ROOT _URL_RE = re.compile(r"https?://[^\s<>'\"`\]]+") _TEMPLATE_MARKERS = ("${", "{{", "}}", "{%", "%}") _PUBLIC_HOST_RE = re.compile(r"^[A-Za-z0-9.-]+$") _CONNECTIVITY_CANARIES = ( "https://github.com/", "https://pypi.org/simple/ansible/", "https://registry.npmjs.org/", ) _RESERVED_HOSTS = { "example", "example.com", "example.net", "example.org", "localhost", "domain-example.com", } _RESERVED_HOST_SUFFIXES = ( ".example", ".example.com", ".example.net", ".example.org", ".invalid", ".local", ".onion", ".localhost", ".localdomain", ".test", ".tld", ) _OK_STATUS_CODES = {401, 403, 405, 406, 415} _WARNING_STATUS_CODES = {418, 429, 451} _REQUEST_TIMEOUT_SECONDS = 10 _FAIL_RETRY_DELAY_SECONDS = 5 _USER_AGENT = ( "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36" ) _MAX_WORKERS = int(os.environ["INFINITO_WORKER_FETCH"]) _MAX_REQUESTS_PER_HOST = 4 _GLOBAL_DEADLINE_SECONDS = 5 * 3600 + 30 * 60 _PROGRESS_INTERVAL = 50 class UrlOccurrence(NamedTuple): file: Path line: int url: str class ProbeOutcome(NamedTuple): kind: str detail: str def _repo_files(root: Path) -> list[Path]: """Return git-tracked and untracked-but-not-ignored files.""" try: out = subprocess.check_output( [ "git", "-C", str(root), "ls-files", "-z", "--cached", "--others", "--exclude-standard", ], stderr=subprocess.STDOUT, ) except Exception: # pragma: no cover - git is expected in tests return _repo_files_without_git_metadata(root) rel_paths = [p for p in out.decode("utf-8", errors="replace").split("\0") if p] return sorted(root / rel for rel in rel_paths if (root / rel).is_file()) def _repo_files_without_git_metadata(root: Path) -> list[Path]: """Fallback scan for environments where the checkout omits .git metadata. Routes through `utils.cache.files.iter_non_ignored_files` so the walk + .gitignore matching are shared with every other lint test.""" return sorted(Path(p) for p in iter_non_ignored_files(root=str(root))) def _is_probably_text_file(path: Path) -> bool: """Skip binary files so the scan stays fast and relevant.""" try: with path.open("rb") as handle: sample = handle.read(4096) except OSError: return False return b"\0" not in sample def _strip_unbalanced_suffix(url: str, opener: str, closer: str) -> str: """Trim a trailing closer that belongs to surrounding markup, not the URL.""" while url.endswith(closer) and url.count(opener) < url.count(closer): url = url[:-1] return url def _truncate_at_second_scheme(url: str) -> str: """Cut off concatenated URLs like ``foo:https://bar`` after the first URL.""" positions = [ pos for pos in (url.find("http://", 1), url.find("https://", 1)) if pos >= 0 ] if not positions: return url return url[: min(positions)].rstrip(",:") def _truncate_at_markup_boundary(url: str) -> str: """Stop at the first unmatched markdown/prose closer after the URL.""" result: list[str] = [] paren_depth = 0 for char in url: if char == ")": if paren_depth == 0: break paren_depth -= 1 elif char == "]": break elif char == "(": paren_depth += 1 result.append(char) return "".join(result) def _normalize_url(raw: str) -> str: """Trim common trailing punctuation added by prose or markup.""" url = _truncate_at_second_scheme(raw) url = _truncate_at_markup_boundary(url) url = url.rstrip(".,;:") for escaped in (r"\n", r"\r"): while url.endswith(escaped): url = url[: -len(escaped)] for opener, closer in (("(", ")"), ("[", "]"), ("{", "}")): url = _strip_unbalanced_suffix(url, opener, closer) return url def _is_live_literal_url(url: str) -> bool: """Return True when *url* is a literal public HTTP(S) URL worth probing.""" if "…" in url: return False if any(marker in url for marker in _TEMPLATE_MARKERS): return False if any(char in url for char in "{}$"): return False try: parsed = urlsplit(url) except ValueError: return False if parsed.scheme not in {"http", "https"} or not parsed.netloc: return False host = (parsed.hostname or "").lower() if not host: return False if "." not in host: return False if not _PUBLIC_HOST_RE.match(host): return False if host.split(".", 1)[0] == "example": return False if host in _RESERVED_HOSTS: return False if any(host.endswith(suffix) for suffix in _RESERVED_HOST_SUFFIXES): return False try: ip = ipaddress.ip_address(host) except ValueError: return True return not ( ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_private or ip.is_reserved or ip.is_unspecified ) def _extract_urls(path: Path) -> list[UrlOccurrence]: """Return URL occurrences from *path* that should be probed live.""" if not _is_probably_text_file(path): return [] try: lines = read_text(str(path)).splitlines() except (OSError, UnicodeDecodeError): return [] placement = "block-above" if path.suffix == ".po" else "same-or-above" occurrences: list[UrlOccurrence] = [] for line_no, line in enumerate(lines, start=1): if is_suppressed_at(lines, line_no, "url", mode=placement): continue for match in _URL_RE.finditer(line): url = _normalize_url(match.group(0)) if _is_live_literal_url(url): occurrences.append(UrlOccurrence(path, line_no, url)) return occurrences def _probe_key(url: str) -> str: """Deduplicate URLs by dropping fragments, which do not affect HTTP reachability.""" parsed = urlsplit(url) return urlunsplit(parsed._replace(fragment="")) def _probe_url(url: str) -> ProbeOutcome: """Probe one URL and classify the result for external-test stability.""" try: with host_gate(url, _MAX_REQUESTS_PER_HOST): response = requests.get( url, allow_redirects=False, headers={"User-Agent": _USER_AGENT}, stream=True, timeout=_REQUEST_TIMEOUT_SECONDS, ) try: status = response.status_code finally: response.close() except requests.Timeout as exc: return ProbeOutcome("unverified", f"Timeout: {exc}") except requests.ConnectionError as exc: return ProbeOutcome("unverified", f"ConnectionError: {exc}") except requests.RequestException as exc: return ProbeOutcome("fail", f"{type(exc).__name__}: {exc}") except Exception as exc: # pragma: no cover - defensive safety net return ProbeOutcome("fail", f"{type(exc).__name__}: {exc}") if status < 400 or status in _OK_STATUS_CODES: return ProbeOutcome("ok", f"HTTP {status}") if status in _WARNING_STATUS_CODES or status >= 500: return ProbeOutcome("warn", f"HTTP {status}") return ProbeOutcome("fail", f"HTTP {status}") def _probe_canary(url: str) -> ProbeOutcome: """Probe a connectivity canary, retrying once so one blip cannot fail the run.""" outcome = _probe_url(url) if outcome.kind == "unverified": outcome = _probe_url(url) return outcome def _probe_with_retry(url: str) -> ProbeOutcome: """Probe one URL; a failing HTTP status is confirmed by a second probe after a pause.""" outcome = _probe_url(url) if outcome.kind == "fail" and outcome.detail.startswith("HTTP "): time.sleep(_FAIL_RETRY_DELAY_SECONDS) outcome = _probe_url(url) return outcome def _collect_occurrences(root: Path) -> dict[str, list[UrlOccurrence]]: """Collect probe-worthy URLs from all non-ignored repository files.""" by_url: dict[str, list[UrlOccurrence]] = defaultdict(list) for path in _repo_files(root): for occurrence in _extract_urls(path): by_url[_probe_key(occurrence.url)].append(occurrence) return dict(by_url) class TestUrlsReachable(unittest.TestCase): """Literal public HTTP(S) URLs in repository files should stay reachable.""" def test_public_http_urls_are_reachable(self) -> None: occurrences_by_url = _collect_occurrences(_REPO_ROOT) self.assertTrue( occurrences_by_url, "No probe-worthy public HTTP(S) URLs found in repository files.", ) unreachable_canaries = [ (url, outcome.detail) for url, outcome in ((c, _probe_canary(c)) for c in _CONNECTIVITY_CANARIES) if outcome.kind == "unverified" ] if unreachable_canaries: self.fail( "This host cannot reach the infrastructure the repository depends" " on, so nothing below would actually be verified:\n" + "\n".join( f" {url} -> {detail}" for url, detail in unreachable_canaries ) + "\n\n Every other probe would fail the same way and be reported" " as a warning, turning an unchecked run green. Run this suite from" " a host with a route to the internet." ) failing_found: list[tuple[str, UrlOccurrence, str, int]] = [] warnings_found: list[tuple[str, UrlOccurrence, str, int]] = [] unverified_found: list[tuple[str, UrlOccurrence, str, int]] = [] total = len(occurrences_by_url) print( f"Probing {total} URLs " f"(workers={_MAX_WORKERS}, " f"per-probe={_REQUEST_TIMEOUT_SECONDS}s, " f"global={_GLOBAL_DEADLINE_SECONDS}s)...", file=sys.stderr, flush=True, ) executor = concurrent.futures.ThreadPoolExecutor(max_workers=_MAX_WORKERS) submission_order = sorted(occurrences_by_url) random.shuffle(submission_order) future_to_url = { executor.submit(_probe_with_retry, url): url for url in submission_order } completed = 0 start = time.monotonic() try: for future in concurrent.futures.as_completed( future_to_url, timeout=_GLOBAL_DEADLINE_SECONDS ): probe_url = future_to_url[future] outcome = future.result() occurrences = occurrences_by_url[probe_url] completed += 1 if completed % _PROGRESS_INTERVAL == 0 or completed == total: elapsed = time.monotonic() - start print( f" [{completed}/{total}] elapsed={elapsed:.1f}s", file=sys.stderr, flush=True, ) if outcome.kind == "fail": failing_found.append( ( "External URL failure", occurrences[0], f"{probe_url} -> {outcome.detail}", len(occurrences), ) ) continue if outcome.kind == "unverified": unverified_found.append( ( "External URL unverified", occurrences[0], f"{probe_url} -> {outcome.detail}", len(occurrences), ) ) continue if outcome.kind == "warn": warnings_found.append( ( "External URL reachability", occurrences[0], f"{probe_url} -> {outcome.detail}", len(occurrences), ) ) except concurrent.futures.TimeoutError: elapsed = time.monotonic() - start unfinished_urls = [u for f, u in future_to_url.items() if not f.done()] for probe_url in unfinished_urls: occurrences = occurrences_by_url[probe_url] unverified_found.append( ( "External URL unverified", occurrences[0], ( f"{probe_url} -> Timeout: global deadline " f"{_GLOBAL_DEADLINE_SECONDS}s exceeded" ), len(occurrences), ) ) print( f" global deadline reached at {elapsed:.1f}s; " f"{len(unfinished_urls)} probes still running " f"(counted as unverified, not waited for)", file=sys.stderr, flush=True, ) finally: executor.shutdown(wait=False, cancel_futures=True) for title, occurrence, message, count in sorted( failing_found, key=lambda item: (item[1].file.as_posix(), item[1].line, item[2]), ): rel = occurrence.file.relative_to(_REPO_ROOT).as_posix() suffix = "" if count == 1 else f" ({count} occurrences)" error( f"{message}{suffix}", title=title, file=rel, line=occurrence.line, ) for title, occurrence, message, count in sorted( warnings_found, key=lambda item: (item[1].file.as_posix(), item[1].line, item[2]), ): rel = occurrence.file.relative_to(_REPO_ROOT).as_posix() suffix = "" if count == 1 else f" ({count} occurrences)" warning( f"{message}{suffix}", title=title, file=rel, line=occurrence.line, ) for title, occurrence, message, count in sorted( unverified_found, key=lambda item: (item[1].file.as_posix(), item[1].line, item[2]), ): rel = occurrence.file.relative_to(_REPO_ROOT).as_posix() suffix = "" if count == 1 else f" ({count} occurrences)" warning( f"{message}{suffix}", title=title, file=rel, line=occurrence.line, ) if not failing_found: return lines = [ f"Failing HTTP(S) URLs found ({len(failing_found)}):", "", " Fix the URL, remove it, or adjust the reference.", " 401/403/405/406/415 = server alive (auth/method/headers). 418/429/451 + all 5xx = warning. Other 4xx = fail.", "", ] for _title, occurrence, message, count in sorted( failing_found, key=lambda item: (item[1].file.as_posix(), item[1].line, item[2]), ): rel = occurrence.file.relative_to(_REPO_ROOT).as_posix() suffix = "" if count == 1 else f" ({count} occurrences)" lines.append(f" {rel}:{occurrence.line}: {message}{suffix}") self.fail("\n".join(lines)) class _FakeResponse: """Stand-in for the streamed response ``_probe_url`` reads and closes.""" status_code = 200 def close(self) -> None: return class _ConcurrencyRecorder: """``requests.get`` replacement recording the peak in-flight count per host.""" def __init__(self, hold_seconds: float) -> None: self._hold_seconds = hold_seconds self._lock = threading.Lock() self._in_flight: dict[str, int] = defaultdict(int) self.peak: dict[str, int] = defaultdict(int) def __call__(self, url: str, **_kwargs: object) -> _FakeResponse: host = (urlsplit(url).hostname or "").lower() with self._lock: self._in_flight[host] += 1 self.peak[host] = max(self.peak[host], self._in_flight[host]) time.sleep(self._hold_seconds) with self._lock: self._in_flight[host] -= 1 return _FakeResponse() class TestHostConcurrencyCap(unittest.TestCase): """One host absorbs at most ``_MAX_REQUESTS_PER_HOST`` probes at a time.""" def test_each_host_gets_its_own_capped_budget(self) -> None: hosts = ("cap-probe-one.invalid", "cap-probe-two.invalid") urls = [ f"https://{host}/{index}" for host in hosts for index in range(_MAX_REQUESTS_PER_HOST * 3) ] recorder = _ConcurrencyRecorder(hold_seconds=0.05) with ( unittest.mock.patch.object(requests, "get", recorder), concurrent.futures.ThreadPoolExecutor(len(urls)) as pool, ): outcomes = list(pool.map(_probe_url, urls)) self.assertEqual({outcome.kind for outcome in outcomes}, {"ok"}) for host in hosts: self.assertEqual( recorder.peak[host], _MAX_REQUESTS_PER_HOST, f"{host} peaked at {recorder.peak[host]} simultaneous probes " f"against a cap of {_MAX_REQUESTS_PER_HOST}: above it the gate is" " missing, below it the two hosts are sharing one gate", ) if __name__ == "__main__": unittest.main()