diff --git a/src/netsleuth/probes/tls_rtt.py b/src/netsleuth/probes/tls_rtt.py index cb03fbe..7932df6 100644 --- a/src/netsleuth/probes/tls_rtt.py +++ b/src/netsleuth/probes/tls_rtt.py @@ -4,6 +4,7 @@ import asyncio import hashlib import ssl import time +import warnings from datetime import UTC, datetime from netsleuth.models import TlsResult @@ -19,7 +20,9 @@ def split_timings( return tcp_rtt_ms, handshake_ms, ttfb_ms -def cpu_bound_ratio(tls_handshake_ms: float | None, tcp_rtt_ms: float | None) -> float | None: +def cpu_bound_ratio( + tls_handshake_ms: float | None, tcp_rtt_ms: float | None +) -> float | None: if tls_handshake_ms is None or not tcp_rtt_ms: return None return tls_handshake_ms / tcp_rtt_ms @@ -28,13 +31,20 @@ def cpu_bound_ratio(tls_handshake_ms: float | None, tcp_rtt_ms: float | None) -> def tls_context(verify: bool = True) -> ssl.SSLContext: ctx = ssl.create_default_context() if not verify: + warnings.warn( + "TLS certificate verification is disabled. This connection is insecure.", + RuntimeWarning, + stacklevel=2, + ) ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE ctx.set_alpn_protocols(["h2", "http/1.1"]) return ctx -def fingerprint_verdict(host: str, actual_sha256: str | None, pins: dict[str, str]) -> str: +def fingerprint_verdict( + host: str, actual_sha256: str | None, pins: dict[str, str] +) -> str: pin = pins.get(host) if not pin: return "unpinned" @@ -76,7 +86,9 @@ async def measure_tls( result = TlsResult(label=label, host=host, port=port) pins = pins or {} try: - tcp_rtt_ms = await tcp_connect_rtt(host, port=port, timeout=timeout, source_ip=source_ip) + tcp_rtt_ms = await tcp_connect_rtt( + host, port=port, timeout=timeout, source_ip=source_ip + ) result.tcp_rtt_ms = tcp_rtt_ms local_addr = (source_ip, 0) if source_ip else None @@ -84,7 +96,11 @@ async def measure_tls( async def _connect(do_verify: bool): return await asyncio.wait_for( asyncio.open_connection( - host, port, ssl=tls_context(verify=do_verify), server_hostname=host, local_addr=local_addr + host, + port, + ssl=tls_context(verify=do_verify), + server_hostname=host, + local_addr=local_addr, ), timeout=timeout, ) @@ -124,7 +140,9 @@ async def measure_tls( result.cert_not_after = parsed.get("notAfter") result.cert_days_remaining = days_remaining(result.cert_not_after) - request = f"HEAD / HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n".encode() + request = ( + f"HEAD / HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n".encode() + ) began_ttfb = time.perf_counter() writer.write(request) await writer.drain() @@ -137,7 +155,9 @@ async def measure_tls( except OSError: pass - _tcp_rtt_ms, handshake_ms, ttfb_ms = split_timings(tcp_rtt_ms, tls_total_ms, ttfb_ms) + _tcp_rtt_ms, handshake_ms, ttfb_ms = split_timings( + tcp_rtt_ms, tls_total_ms, ttfb_ms + ) result.tls_handshake_ms = handshake_ms result.ttfb_ms = ttfb_ms except (OSError, TimeoutError, ssl.SSLError) as exc: @@ -160,7 +180,15 @@ async def tls_fanout( async def _bounded(label: str, host: str) -> TlsResult: async with semaphore: return await measure_tls( - label, host, port=port, timeout=timeout, source_ip=source_ip, verify=verify, pins=pins + label, + host, + port=port, + timeout=timeout, + source_ip=source_ip, + verify=verify, + pins=pins, ) - return list(await asyncio.gather(*(_bounded(label, host) for label, host in targets))) + return list( + await asyncio.gather(*(_bounded(label, host) for label, host in targets)) + ) diff --git a/tests/test_cli.py b/tests/test_cli.py index 0e72067..6b601b8 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,5 +1,4 @@ -import os -from pathlib import Path, PurePosixPath +from pathlib import PurePosixPath from unittest.mock import MagicMock, patch from netsleuth.cli import _dedupe, _os_timezone, parse_target diff --git a/tests/test_netinfo.py b/tests/test_netinfo.py index 7134cd9..7aa16da 100644 --- a/tests/test_netinfo.py +++ b/tests/test_netinfo.py @@ -111,7 +111,6 @@ def test_degradation_note_on_windows_mentions_the_api_rather_than_sysctl(): import pytest -from unittest.mock import patch from netsleuth.netinfo import iface_for_ip, is_tunnel_iface, mtu_anomaly, primary_interface_ip @@ -244,7 +243,6 @@ class MockStats: from unittest.mock import patch from netsleuth.netinfo import collect_local_net -from netsleuth.models import LocalNet @patch("netsleuth.netinfo.primary_interface_ip") @patch("netsleuth.netinfo.psutil.net_if_addrs") diff --git a/tests/test_tls_rtt.py b/tests/test_tls_rtt.py index 5ef6ae7..8feec9e 100644 --- a/tests/test_tls_rtt.py +++ b/tests/test_tls_rtt.py @@ -4,6 +4,8 @@ import socket import ssl from datetime import UTC, datetime +import pytest + from netsleuth.probes.tls_rtt import ( cert_name, cpu_bound_ratio, @@ -63,7 +65,8 @@ def test_tls_context_verifies_by_default(): def test_tls_context_disables_verification_when_asked(): - ctx = tls_context(verify=False) + with pytest.warns(RuntimeWarning, match="TLS certificate verification is disabled"): + ctx = tls_context(verify=False) assert ctx.check_hostname is False assert ctx.verify_mode == ssl.CERT_NONE @@ -83,7 +86,10 @@ def test_fingerprint_verdict_unpinned_when_host_has_no_pin(): def test_fingerprint_verdict_mismatch_when_actual_is_none(): - assert fingerprint_verdict("example.com", None, {"example.com": "aabbccdd"}) == "mismatch" + assert ( + fingerprint_verdict("example.com", None, {"example.com": "aabbccdd"}) + == "mismatch" + ) def test_cert_name_flattens_the_rdn_sequence():