diff --git a/.github/workflows/backend.yml b/.github/workflows/backend.yml index 84c4514e6496..809fa7393f70 100644 --- a/.github/workflows/backend.yml +++ b/.github/workflows/backend.yml @@ -254,6 +254,9 @@ jobs: PYTHONHASHSEED: '0' XDIST_PER_WORKER_SNUBA: '1' XDIST_WORKERS: '4' + # Record tests that reach hosts outside the runner, without failing them. Only on + # master and manual runs, so pull requests pay nothing. See network_guard.py. + SENTRY_NETWORK_GUARD: ${{ github.event_name == 'pull_request' && 'off' || 'report' }} steps: - uses: actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332 # v4.1.7 @@ -404,6 +407,16 @@ jobs: .artifacts/resource-usage-summary.json retention-days: 30 + - name: Upload network guard report + if: ${{ !cancelled() && env.SENTRY_NETWORK_GUARD != 'off' }} + continue-on-error: true + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 + with: + name: network-guard-${{ github.run_id }}-${{ matrix.instance }} + path: .artifacts/network-guard.json + if-no-files-found: ignore + retention-days: 30 + - name: Report failures if: ${{ !cancelled() && github.event_name == 'pull_request' }} continue-on-error: true diff --git a/src/sentry/testutils/pytest/__init__.py b/src/sentry/testutils/pytest/__init__.py index 1d98423e8551..168ab930b3cf 100644 --- a/src/sentry/testutils/pytest/__init__.py +++ b/src/sentry/testutils/pytest/__init__.py @@ -14,6 +14,7 @@ "sentry.testutils.pytest.show_flaky_failures", "sentry.testutils.thread_leaks.pytest", "sentry.testutils.pytest.timeout_info", + "sentry.testutils.pytest.network_guard", ] if os.environ.get("SENTRY_SKIP_SELENIUM_PLUGIN") != "1": diff --git a/src/sentry/testutils/pytest/network_guard.py b/src/sentry/testutils/pytest/network_guard.py new file mode 100644 index 000000000000..47b24dff9a43 --- /dev/null +++ b/src/sentry/testutils/pytest/network_guard.py @@ -0,0 +1,580 @@ +"""Guard tests against reaching the outside world over the network. + +Tests that talk to real external hosts are slow and flaky, and they can leak +data. This plugin patches the Python-level socket entry points that establish +outbound traffic and checks each destination against an allowlist of local +targets: loopback, the unspecified address, the local hostname, the docker +bridge gateway that CI binds the live server to, and the hosts of the DSNs that +test infrastructure itself reports to. + +``SENTRY_NETWORK_GUARD`` selects the mode: + +- ``off``: nothing is patched. The default. +- ``report``: blocked attempts are recorded and still allowed, then written to + ``.artifacts/network-guard.json`` and summarized at the end of the session, + including the GitHub Actions job summary. CI uses it on master. +- ``strict``: blocked attempts raise :class:`NetworkAccessBlocked`, except in + known offenders. + +``SENTRY_NETWORK_GUARD_ALLOW`` takes extra comma-separated hosts or IPs to allow, +for local setups that run services somewhere other than loopback. + +Mark a test with ``@pytest.mark.enable_network(reason=...)`` when it really +needs the network. + +``tests/network_guard_known_offenders.txt`` (relative to the pytest rootdir, or +``SENTRY_NETWORK_GUARD_KNOWN_OFFENDERS``) lists tests that are known to reach +the network and still need a fix. One entry per line: a test file, a class, or +a test nodeid (which covers all of its parametrizations). Their attempts are +recorded but never blocked, and the report separates them from new offenders. +Remove an entry once its tests are fixed. + +Only connection establishment is checked (``connect``, ``connect_ex``, and +``sendto``/``sendmsg`` with an explicit address) plus the resolver functions. +``send``/``recv`` are left alone: a socket can only reach a peer it connected +to, and wrapping them would add a Python call to every redis or HTTP round-trip. +Clients implemented in C (psycopg2/libpq, confluent-kafka/librdkafka, grpc core, +c-ares) open their own sockets and are invisible to this guard. +""" + +from __future__ import annotations + +import ipaddress +import os +import socket +import sys +import sysconfig +import threading +from collections.abc import Callable, Generator +from pathlib import Path +from typing import Any +from urllib.parse import urlsplit + +import orjson +import pytest + +MODE_ENV = "SENTRY_NETWORK_GUARD" +ALLOW_ENV = "SENTRY_NETWORK_GUARD_ALLOW" +REPORT_ENV = "SENTRY_NETWORK_GUARD_REPORT" +KNOWN_OFFENDERS_ENV = "SENTRY_NETWORK_GUARD_KNOWN_OFFENDERS" +MARKER = "enable_network" +MODES = ("off", "report", "strict") + +_MAX_RECORDS = 5000 +_MAX_SUMMARY_ROWS = 100 +_WORKEROUTPUT_KEY = "sentry_network_guard" +_INET_FAMILIES = frozenset((socket.AF_INET, socket.AF_INET6)) +_STDLIB_DIR = sysconfig.get_paths()["stdlib"] +_THIS_FILE = __file__ + + +class NetworkAccessBlocked(PermissionError): + pass + + +def _get_mode() -> str: + mode = os.environ.get(MODE_ENV, "off").strip().lower() or "off" + if mode not in MODES: + raise pytest.UsageError(f"{MODE_ENV} must be one of {', '.join(MODES)}, got {mode!r}") + return mode + + +_mode = "off" +# The nodeid of the test whose setup, call, or teardown is running, or None +# when nothing is being guarded. Read by every thread so that work a test hands +# to a thread pool is still attributed to it. +_current_nodeid: str | None = None +# The known offenders entry that covers the running test, if any. +_current_known: str | None = None +_known_offenders: frozenset[str] = frozenset() +# Known offenders entries whose tests ran in this process, to find fixed ones. +_known_ran: set[str] = set() +_allowed_hostnames: set[str] = set() +_allowed_ips: set[str] = set() +# Hosts already checked and allowed. Saves the ipaddress parse on repeat connects. +_allowed_cache: set[str] = set() +_names_by_ip: dict[str, str] = {} +_records: dict[tuple[str, str, str], dict[str, Any]] = {} +_records_lock = threading.Lock() +_originals: list[tuple[Any, str, Any]] = [] +_merged_key = pytest.StashKey[list[dict[str, Any]]]() +_known_ran_key = pytest.StashKey[set[str]]() + + +def _known_offenders_path(config: pytest.Config) -> Path: + if path := os.environ.get(KNOWN_OFFENDERS_ENV): + return Path(path) + return config.rootpath / "tests" / "network_guard_known_offenders.txt" + + +def _load_known_offenders(path: Path) -> frozenset[str]: + try: + lines = path.read_text().splitlines() + except FileNotFoundError: + return frozenset() + entries = (line.split("#", 1)[0].strip() for line in lines) + return frozenset(entry for entry in entries if entry) + + +def _known_entry(nodeid: str) -> str | None: + """Return the known offenders entry that covers ``nodeid``, if any.""" + if nodeid in _known_offenders: + return nodeid + parts = nodeid.split("[", 1)[0].split("::") + for end in range(len(parts), 0, -1): + prefix = "::".join(parts[:end]) + if prefix in _known_offenders: + return prefix + return None + + +def _dsn_host(dsn: str | None) -> str | None: + if not dsn: + return None + return urlsplit(dsn).hostname + + +def _live_server_host() -> str | None: + # pytest-django accepts "host", "host:port", or "host:port-range". + address = os.environ.get("DJANGO_LIVE_TEST_SERVER_ADDRESS", "").strip() + if not address: + return None + if address.count(":") == 1: + address = address.split(":", 1)[0] + return address.strip("[]") + + +def _build_allowlist() -> None: + from sentry.testutils.thread_leaks.sentry import THREAD_LEAK_DSN + + hosts = {"localhost", "localhost.localdomain"} + try: + hostname = socket.gethostname() + except OSError: + hostname = "" + if hostname: + hosts.add(hostname) + hosts.add(hostname.split(".", 1)[0]) + + # Test infrastructure reports failures and thread leaks to Sentry itself. + for dsn in (os.environ.get("PYTEST_SENTRY_DSN"), THREAD_LEAK_DSN): + if dsn_host := _dsn_host(dsn): + hosts.add(dsn_host) + + # CI binds the live server to the docker bridge gateway so that containers + # such as Relay and Symbolicator can call back into it. + if live_server_host := _live_server_host(): + hosts.add(live_server_host) + + for extra in os.environ.get(ALLOW_ENV, "").split(","): + if extra := extra.strip(): + hosts.add(extra) + + for host in hosts: + folded = host.casefold().rstrip(".") + try: + _allowed_ips.add(str(_parse_ip(folded))) + except ValueError: + _allowed_hostnames.add(folded) + + +def _parse_ip(host: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address: + ip = ipaddress.ip_address(host.split("%", 1)[0]) + if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None: + return ip.ipv4_mapped + return ip + + +def _host_allowed(host: str) -> bool: + if host in _allowed_cache: + return True + folded = host.casefold().rstrip(".") + if folded in _allowed_hostnames: + allowed = True + else: + try: + ip = _parse_ip(folded) + except ValueError: + allowed = False + else: + # Connecting to the unspecified address reaches the local host on + # Linux and macOS. The redis-cluster test config relies on it. + allowed = ip.is_loopback or ip.is_unspecified or str(ip) in _allowed_ips + if allowed: + _allowed_cache.add(host) + return allowed + + +def _as_host(value: Any) -> str | None: + if isinstance(value, bytes): + try: + return value.decode("ascii") + except UnicodeDecodeError: + return None + if isinstance(value, str): + return value + return None + + +def _origin() -> list[str]: + """Return the innermost project frames plus the library that opened the socket.""" + frames: list[str] = [] + library: str | None = None + frame: Any = sys._getframe(1) + while frame is not None and len(frames) < 4: + filename = frame.f_code.co_filename + if filename == _THIS_FILE: + pass + elif "site-packages" in filename: + if library is None: + library = filename.split("site-packages/", 1)[1] + library = f"{library}:{frame.f_lineno}" + elif not filename.startswith(_STDLIB_DIR) and not filename.startswith("<"): + frames.append(f"{_relpath(filename)}:{frame.f_lineno} in {frame.f_code.co_name}") + frame = frame.f_back + if library is not None: + frames.append(f"via {library}") + return frames + + +def _relpath(filename: str) -> str: + for marker in ("/src/sentry/", "/tests/", "/getsentry/"): + index = filename.rfind(marker) + if index != -1: + return filename[index + 1 :] + return filename + + +def _blocked(nodeid: str, kind: str, target: str) -> None: + known = _current_known is not None + key = (nodeid, kind, target) + with _records_lock: + record = _records.get(key) + if record is not None: + record["count"] += 1 + elif len(_records) < _MAX_RECORDS: + _records[key] = { + "nodeid": nodeid, + "kind": kind, + "target": target, + "count": 1, + "known": known, + "thread": threading.current_thread().name, + "origin": _origin(), + } + if _mode == "strict" and not known: + raise NetworkAccessBlocked( + f"Network access to {target} ({kind}) is blocked during tests. " + "Mock it (for example with `responses`), or mark the test with " + f"@pytest.mark.{MARKER}(reason=...) if it really needs the network." + ) + + +def _check_address(sock: socket.socket, address: Any, kind: str) -> None: + nodeid = _current_nodeid + if nodeid is None or type(address) is not tuple or not address: + return + host = address[0] + if not isinstance(host, str) or _host_allowed(host): + return + if sock.family not in _INET_FAMILIES: + return + port = address[1] if len(address) > 1 else "?" + name = _names_by_ip.get(host) + target = f"{name} ({host}):{port}" if name else f"{host}:{port}" + _blocked(nodeid, kind, target) + + +def _guard_connect(original: Callable[..., Any]) -> Callable[..., Any]: + def connect(self: socket.socket, address: Any) -> Any: + _check_address(self, address, "connect") + return original(self, address) + + return connect + + +def _guard_sendto(original: Callable[..., Any]) -> Callable[..., Any]: + def sendto(self: socket.socket, *args: Any) -> Any: + if len(args) >= 2: + _check_address(self, args[-1], "sendto") + return original(self, *args) + + return sendto + + +def _guard_sendmsg(original: Callable[..., Any]) -> Callable[..., Any]: + def sendmsg(self: socket.socket, *args: Any) -> Any: + if len(args) >= 4: + _check_address(self, args[3], "sendmsg") + return original(self, *args) + + return sendmsg + + +def _is_numeric_host(host: str) -> bool: + try: + _parse_ip(host) + except ValueError: + pass + else: + return True + # The resolver parses legacy IPv4 forms such as "2130706433" or + # "0177.0.0.1" with inet_aton and sends no DNS query for them. + try: + socket.inet_aton(host) + except (OSError, ValueError): + return False + return True + + +def _check_resolve(kind: str, host: str | None) -> bool: + """Return whether the lookup is for a non-local host that was let through.""" + nodeid = _current_nodeid + if nodeid is None or not host or _host_allowed(host): + return False + if kind != "gethostbyaddr" and kind != "getnameinfo" and _is_numeric_host(host): + return False # a numeric address needs no lookup; the connect is checked + _blocked(nodeid, kind, host) + return True + + +def _remember(host: str | None, infos: Any) -> None: + # Name the IPs of reported lookups, so the connect that follows is readable. + if not host: + return + for info in infos: + ip = info[4][0] + if isinstance(ip, str): + _names_by_ip.setdefault(ip, host) + + +def _guard_getaddrinfo(original: Callable[..., Any]) -> Callable[..., Any]: + def getaddrinfo(host: Any, *args: Any, **kwargs: Any) -> Any: + name = _as_host(host) + reported = _check_resolve("getaddrinfo", name) + infos = original(host, *args, **kwargs) + if name and _current_nodeid is not None: + folded = name.casefold().rstrip(".") + if folded in _allowed_hostnames: + for info in infos: + if isinstance(info[4][0], str): + _allowed_ips.add(str(_parse_ip(info[4][0]))) + elif reported: + _remember(name, infos) + return infos + + return getaddrinfo + + +def _guard_host_lookup(kind: str, original: Callable[..., Any]) -> Callable[..., Any]: + def lookup(host: Any, *args: Any, **kwargs: Any) -> Any: + _check_resolve(kind, _as_host(host)) + return original(host, *args, **kwargs) + + return lookup + + +def _guard_getnameinfo(original: Callable[..., Any]) -> Callable[..., Any]: + def getnameinfo(sockaddr: Any, flags: int) -> Any: + if not flags & socket.NI_NUMERICHOST and type(sockaddr) is tuple and sockaddr: + _check_resolve("getnameinfo", _as_host(sockaddr[0])) + return original(sockaddr, flags) + + return getnameinfo + + +def _patch(owner: Any, name: str, wrapper: Callable[..., Any]) -> None: + original = getattr(owner, name, None) + if original is None: + return + _originals.append((owner, name, original)) + setattr(owner, name, wrapper) + + +def install() -> None: + _build_allowlist() + _patch(socket.socket, "connect", _guard_connect(socket.socket.connect)) + _patch(socket.socket, "connect_ex", _guard_connect(socket.socket.connect_ex)) + _patch(socket.socket, "sendto", _guard_sendto(socket.socket.sendto)) + if hasattr(socket.socket, "sendmsg"): + _patch(socket.socket, "sendmsg", _guard_sendmsg(socket.socket.sendmsg)) + _patch(socket, "getaddrinfo", _guard_getaddrinfo(socket.getaddrinfo)) + for name in ("gethostbyname", "gethostbyname_ex", "gethostbyaddr"): + _patch(socket, name, _guard_host_lookup(name, getattr(socket, name))) + _patch(socket, "getnameinfo", _guard_getnameinfo(socket.getnameinfo)) + + +def uninstall() -> None: + while _originals: + owner, name, original = _originals.pop() + setattr(owner, name, original) + + +def get_records() -> list[dict[str, Any]]: + with _records_lock: + return [dict(record) for record in _records.values()] + + +def pytest_configure(config: pytest.Config) -> None: + global _mode, _known_offenders + config.addinivalue_line( + "markers", f"{MARKER}(reason): allow this test to reach hosts outside the local machine" + ) + _mode = _get_mode() + if _mode != "off": + _known_offenders = _load_known_offenders(_known_offenders_path(config)) + install() + config.stash[_merged_key] = [] + config.stash[_known_ran_key] = set() + + +def pytest_unconfigure(config: pytest.Config) -> None: + uninstall() + + +@pytest.hookimpl(wrapper=True) +def pytest_collection(session: pytest.Session) -> Generator[None, object, object]: + global _current_nodeid + if _mode == "off": + return (yield) + _current_nodeid = "" + try: + return (yield) + finally: + _current_nodeid = None + + +@pytest.hookimpl(wrapper=True) +def pytest_runtest_protocol( + item: pytest.Item, nextitem: pytest.Item | None +) -> Generator[None, object, object]: + global _current_nodeid, _current_known + if _mode == "off" or item.get_closest_marker(MARKER) is not None: + return (yield) + _current_nodeid = item.nodeid + _current_known = _known_entry(item.nodeid) + if _current_known is not None: + _known_ran.add(_current_known) + try: + return (yield) + finally: + _current_nodeid = _current_known = None + + +@pytest.hookimpl(optionalhook=True) +def pytest_testnodedown(node: Any, error: object) -> None: + workeroutput = getattr(node, "workeroutput", None) or {} + output = workeroutput.get(_WORKEROUTPUT_KEY) + if output: + node.config.stash[_merged_key].extend(output["records"]) + node.config.stash[_known_ran_key].update(output["known_ran"]) + + +def _report_path(config: pytest.Config) -> Path: + if path := os.environ.get(REPORT_ENV): + return Path(path) + return config.rootpath / ".artifacts" / "network-guard.json" + + +def _merged(config: pytest.Config) -> list[dict[str, Any]]: + return sorted( + config.stash[_merged_key] + get_records(), + key=lambda r: (r["target"], r["nodeid"], r["kind"]), + ) + + +def _fixed_known_offenders(config: pytest.Config, records: list[dict[str, Any]]) -> list[str]: + """Known offenders entries whose tests ran in this session without an attempt.""" + offending = {_known_entry(r["nodeid"]) for r in records if r["known"]} + return sorted((config.stash[_known_ran_key] | _known_ran) - offending) + + +def _md(value: str) -> str: + return value.replace("|", "\\|") + + +def _job_summary( + config: pytest.Config, records: list[dict[str, Any]], fixed: list[str] +) -> list[str]: + new = [r for r in records if not r["known"]] + known_tests = {r["nodeid"] for r in records if r["known"]} + new_tests = {r["nodeid"] for r in new} + known_path = _known_offenders_path(config) + try: + known_path = known_path.relative_to(config.rootpath) + except ValueError: + pass + lines = [f"### Network guard ({_mode} mode)", ""] + if new: + lines += [ + f"**{len(new_tests)} test(s) not in `{known_path}` reached hosts outside the " + "runner.** Mock the call (for example with `responses`), or mark the test with " + f"`@pytest.mark.{MARKER}(reason=...)` if it really needs the network.", + "", + "| Test | Attempt | Origin |", + "| --- | --- | --- |", + ] + for r in new[:_MAX_SUMMARY_ROWS]: + origin = r["origin"][0] if r["origin"] else "?" + lines.append( + f"| `{_md(r['nodeid'])}` | {r['kind']} `{_md(r['target'])}` x{r['count']} " + f"| `{_md(origin)}` |" + ) + if len(new) > _MAX_SUMMARY_ROWS: + lines.append(f"\n{len(new) - _MAX_SUMMARY_ROWS} more in the network-guard artifact.") + else: + lines.append("No new tests reached hosts outside the runner.") + if known_tests: + lines += ["", f"{len(known_tests)} known offender test(s) still reach the network."] + if fixed: + lines += [ + "", + f"These `{known_path}` entries ran without a network attempt in this shard. " + "If no other shard reports them, remove them:", + "", + *(f"- `{entry}`" for entry in fixed), + ] + return lines + + +@pytest.hookimpl(trylast=True) +def pytest_sessionfinish(session: pytest.Session) -> None: + if _mode == "off": + return + config = session.config + workeroutput = getattr(config, "workeroutput", None) + if workeroutput is not None: + workeroutput[_WORKEROUTPUT_KEY] = { + "records": get_records(), + "known_ran": sorted(_known_ran), + } + return + records = _merged(config) + fixed = _fixed_known_offenders(config, records) + path = _report_path(config) + path.parent.mkdir(parents=True, exist_ok=True) + report = {"mode": _mode, "records": records, "fixed_known_offenders": fixed} + path.write_bytes(orjson.dumps(report, option=orjson.OPT_INDENT_2)) + if summary_path := os.environ.get("GITHUB_STEP_SUMMARY"): + with open(summary_path, "a") as f: + f.write("\n".join(_job_summary(config, records, fixed)) + "\n") + + +def pytest_terminal_summary(terminalreporter: Any, config: pytest.Config) -> None: + if _mode == "off" or hasattr(config, "workerinput"): + return + records = _merged(config) + if not records: + return + new = [r for r in records if not r["known"]] + terminalreporter.section("network guard") + terminalreporter.write_line( + f"{len(records)} blocked network attempt(s) ({_mode} mode), {len(new)} outside known " + f"offenders, see {_report_path(config)}" + ) + for record in new[:200]: + origin = record["origin"][0] if record["origin"] else "?" + terminalreporter.write_line( + f"NETWORK-GUARD {record['kind']} {record['target']} x{record['count']} " + f"[{record['thread']}] {record['nodeid']} @ {origin}" + ) diff --git a/src/sentry/testutils/thread_leaks/sentry.py b/src/sentry/testutils/thread_leaks/sentry.py index d22231ca2e90..aca9adfc7437 100644 --- a/src/sentry/testutils/thread_leaks/sentry.py +++ b/src/sentry/testutils/thread_leaks/sentry.py @@ -18,6 +18,9 @@ from sentry_sdk._types import Event as SentryEvent from sentry_sdk._types import LogLevelStr +# proj-thread-leaks +THREAD_LEAK_DSN = "https://447e81e71c1aa0da0d52f3eaba37a703@o1.ingest.us.sentry.io/4509798820085760" + @functools.cache def get_scope() -> sentry_sdk.scope.Scope: @@ -44,8 +47,7 @@ def get_scope() -> sentry_sdk.scope.Scope: environment = "local" client = sentry_sdk.Client( - # proj-thread-leaks - dsn="https://447e81e71c1aa0da0d52f3eaba37a703@o1.ingest.us.sentry.io/4509798820085760", + dsn=THREAD_LEAK_DSN, environment=environment, ) diff --git a/tests/network_guard_known_offenders.txt b/tests/network_guard_known_offenders.txt new file mode 100644 index 000000000000..68bd8ec0563e --- /dev/null +++ b/tests/network_guard_known_offenders.txt @@ -0,0 +1,4 @@ +# Tests that are known to reach hosts outside the CI runner and still need a fix. +# See src/sentry/testutils/pytest/network_guard.py. One entry per line: a test file, +# a class, or a test nodeid. The network guard records these tests separately from +# new offenders, and strict mode does not block them. Remove an entry once it's fixed. diff --git a/tests/sentry/testutils/pytest/test_network_guard.py b/tests/sentry/testutils/pytest/test_network_guard.py new file mode 100644 index 000000000000..bbdb9d314268 --- /dev/null +++ b/tests/sentry/testutils/pytest/test_network_guard.py @@ -0,0 +1,232 @@ +import os +import socket +from collections.abc import Generator +from pathlib import Path +from unittest import mock + +import pytest + +from sentry.testutils.pytest import network_guard + +# TEST-NET-3 (RFC 5737) is reserved for documentation and never routed. +EXTERNAL_IP = "203.0.113.7" + + +@pytest.fixture +def guard() -> Generator[None]: + installed = bool(network_guard._originals) + if not installed: + network_guard.install() + try: + with ( + mock.patch.object(network_guard, "_mode", "strict"), + mock.patch.object(network_guard, "_current_nodeid", "test-node"), + mock.patch.object(network_guard, "_records", {}), + ): + yield + finally: + if not installed: + network_guard.uninstall() + + +@pytest.mark.usefixtures("guard") +def test_blocks_external_connect() -> None: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + with pytest.raises(network_guard.NetworkAccessBlocked): + sock.connect((EXTERNAL_IP, 443)) + with pytest.raises(network_guard.NetworkAccessBlocked): + sock.connect_ex((EXTERNAL_IP, 443)) + + [record] = network_guard.get_records() + assert record["nodeid"] == "test-node" + assert record["kind"] == "connect" + assert record["target"] == f"{EXTERNAL_IP}:443" + assert record["count"] == 2 + assert record["origin"][0].startswith("tests/sentry/testutils/pytest/test_network_guard.py:") + + +@pytest.mark.usefixtures("guard") +def test_blocks_external_sendto() -> None: + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + with pytest.raises(network_guard.NetworkAccessBlocked): + sock.sendto(b"x", (EXTERNAL_IP, 53)) + with pytest.raises(network_guard.NetworkAccessBlocked): + sock.sendto(b"x", 0, (EXTERNAL_IP, 53)) + + +@pytest.mark.usefixtures("guard") +def test_blocks_external_resolution() -> None: + with pytest.raises(network_guard.NetworkAccessBlocked): + socket.getaddrinfo("example.com", 443) + with pytest.raises(network_guard.NetworkAccessBlocked): + socket.gethostbyname("example.com") + + +@pytest.mark.usefixtures("guard") +@pytest.mark.parametrize("host", [EXTERNAL_IP, "2130706433", "0177.0000.0000.0001"]) +def test_numeric_hosts_resolve_without_a_record(host: str) -> None: + assert socket.getaddrinfo(host, 80) + assert network_guard.get_records() == [] + + +@pytest.mark.usefixtures("guard") +def test_allows_local_targets() -> None: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server: + server.bind(("127.0.0.1", 0)) + server.listen(8) + port = server.getsockname()[1] + for host in ("127.0.0.1", "localhost", "0.0.0.0"): + with socket.create_connection((host, port), timeout=1): + pass + + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.sendto(b"x", ("127.0.0.1", 9)) + + assert socket.getaddrinfo(None, 80) + assert network_guard.get_records() == [] + + +@pytest.mark.parametrize( + "host", + [ + "127.0.0.1", + "127.1.2.3", + "::1", + "::ffff:127.0.0.1", + "0.0.0.0", + "::", + "LOCALHOST", + "localhost.", + ], +) +def test_local_hosts_allowed(host: str) -> None: + assert network_guard._host_allowed(host) + + +@pytest.mark.parametrize("host", [EXTERNAL_IP, "10.0.0.1", "169.254.169.254", "example.com"]) +def test_external_hosts_not_allowed(host: str) -> None: + assert not network_guard._host_allowed(host) + + +@pytest.mark.usefixtures("guard") +def test_report_mode_records_without_raising() -> None: + with ( + mock.patch.object(network_guard, "_mode", "report"), + socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock, + ): + sock.setblocking(False) + try: + sock.sendto(b"x", (EXTERNAL_IP, 9)) + except OSError: + pass # the sandbox may have no route; the attempt is what matters + + [record] = network_guard.get_records() + assert record["kind"] == "sendto" + assert record["target"] == f"{EXTERNAL_IP}:9" + + +@pytest.mark.usefixtures("guard") +def test_inactive_outside_tests() -> None: + with ( + mock.patch.object(network_guard, "_current_nodeid", None), + socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock, + ): + network_guard._check_address(sock, (EXTERNAL_IP, 443), "connect") + assert network_guard.get_records() == [] + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("172.17.0.1", "172.17.0.1"), + ("172.17.0.1:8000", "172.17.0.1"), + ("172.17.0.1:8000-8010", "172.17.0.1"), + ("", None), + ], +) +def test_live_server_host(value: str, expected: str | None) -> None: + with mock.patch.dict(os.environ, {"DJANGO_LIVE_TEST_SERVER_ADDRESS": value}): + assert network_guard._live_server_host() == expected + + +@pytest.mark.enable_network(reason="checks that the marker turns the guard off") +def test_enable_network_marker() -> None: + assert network_guard._current_nodeid is None + + +@pytest.mark.parametrize( + ("nodeid", "expected"), + [ + ("tests/a/test_x.py::test_one", "tests/a/test_x.py::test_one"), + ("tests/a/test_x.py::test_one[case-1]", "tests/a/test_x.py::test_one"), + ("tests/a/test_y.py::TestY::test_two", "tests/a/test_y.py"), + ("tests/a/test_z.py::TestZ::test_three", "tests/a/test_z.py::TestZ"), + ("tests/a/test_x.py::test_one_more", None), + ("tests/a/test_x.py::test_other", None), + ], +) +def test_known_entry(nodeid: str, expected: str | None) -> None: + known = frozenset( + ["tests/a/test_x.py::test_one", "tests/a/test_y.py", "tests/a/test_z.py::TestZ"] + ) + with mock.patch.object(network_guard, "_known_offenders", known): + assert network_guard._known_entry(nodeid) == expected + + +def test_load_known_offenders(tmp_path: Path) -> None: + path = tmp_path / "known.txt" + path.write_text("# header\n\ntests/a/test_x.py::test_one # stripe\ntests/a/test_y.py\n") + assert network_guard._load_known_offenders(path) == { + "tests/a/test_x.py::test_one", + "tests/a/test_y.py", + } + assert network_guard._load_known_offenders(tmp_path / "missing.txt") == frozenset() + + +@pytest.mark.usefixtures("guard") +def test_strict_mode_records_known_offenders_without_raising() -> None: + with ( + mock.patch.object(network_guard, "_current_known", "test-node"), + socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock, + ): + sock.setblocking(False) + try: + sock.sendto(b"x", (EXTERNAL_IP, 9)) + except OSError: + pass # the sandbox may have no route; the attempt is what matters + + [record] = network_guard.get_records() + assert record["known"] is True + + +def test_job_summary_separates_new_and_known_offenders(pytestconfig: pytest.Config) -> None: + records = [ + { + "nodeid": "tests/a/test_x.py::test_new[a|b]", + "kind": "connect", + "target": f"{EXTERNAL_IP}:443", + "count": 2, + "known": False, + "thread": "MainThread", + "origin": ["src/sentry/x.py:1 in f"], + }, + { + "nodeid": "tests/a/test_y.py::test_known", + "kind": "getaddrinfo", + "target": "api.example.com", + "count": 1, + "known": True, + "thread": "MainThread", + "origin": [], + }, + ] + with mock.patch.object(network_guard, "_mode", "report"): + summary = "\n".join( + network_guard._job_summary(pytestconfig, records, ["tests/a/test_z.py"]) + ) + + assert "**1 test(s) not in `tests/network_guard_known_offenders.txt`" in summary + assert f"| `tests/a/test_x.py::test_new[a\\|b]` | connect `{EXTERNAL_IP}:443` x2 |" in summary + assert "api.example.com" not in summary + assert "1 known offender test(s) still reach the network." in summary + assert "- `tests/a/test_z.py`" in summary