mirror of
https://github.com/Horizontal42/netsleuth.git
synced 2026-11-11 02:12:04 +03:00
verify TLS certificates and fingerprint them for MITM detection
tls_context() now defaults to real certificate verification instead of the
tool's original always-CERT_NONE posture; measure_tls() only falls back to
unverified on ssl.SSLCertVerificationError, retiming the retry since the
failed attempt's duration isn't representative. This is also the only way
to get subject/issuer/notAfter at all - getpeercert() returns {} for a
chain that never validated.
Every TLS result now carries cert_sha256 (SHA-256 of the full DER cert, not
the SPKI - that would need the cryptography package, a dependency this
doesn't need), cert_issuer, cert_days_remaining, and the long-dead
cert_verified field finally gets written. New tls.pinned_fingerprints
config (host -> expected hash, capped and hex-validated) drives three new
findings: a pin mismatch is crit, an unpinned validation failure is warn,
an expiring cert is warn.
Live-tested with --tls against real Cloudflare/Google endpoints - verified
handshake, correct issuer and expiry both times.
Part of plans/zesty-zooming-wilkinson.md (gitignored).
This commit is contained in:
Andrew Pimenov
parent
84c3dffbce
commit
99876103f8
14 changed files with 301 additions and 15 deletions
|
|
@ -141,6 +141,10 @@ The AS prefix benchmark (`probes/prefix_benchmark.py`) gets the same treatment f
|
|||
|
||||
**Speedtest server geo-sanity compares countries, not coordinates.** The original design called for a haversine distance in km between the client and the speedtest server, which needs both ends' lat/lon. Live inspection of the bundled Ookla CLI's own JSON output (`speedtest.exe --format=json`) showed its `server` object carries `location` (a city name) and `country`, never `lat`/`lon` — that's the documented schema, not a version quirk. `speed_findings()` compares `SpeedResult.server_country` against the client's own geolocated country instead: coarser than a distance, but real, always available when a tier reports a country, and it still catches the case that matters (the test hit a server in another country entirely).
|
||||
|
||||
**TLS certificate handshake tries verified first, only degrades to `CERT_NONE` on a validation failure.** `probes/tls_rtt.py::tls_context(verify: bool)` defaults to a real, unmodified `create_default_context()` — a change from the tool's original always-`CERT_NONE` posture. `measure_tls()` attempts a verified connection first; only on `ssl.SSLCertVerificationError` does it retry unverified (re-timing the retry, since the failed attempt's duration isn't representative of a real connection). This is the only way to get `subject`/`issuer`/`notAfter` at all: Python's `getpeercert()` returns `{}` for an unvalidated chain, populating the parsed dict only when the certificate actually verified — `getpeercert(binary_form=True)` (the raw DER, hashed into `cert_sha256`) is the one thing available either way. The fingerprint is the SHA-256 of the **full DER certificate, not the SPKI** — extracting just the subject-public-key-info needs the `cryptography` package, a new dependency this feature doesn't need; the report footnote tells a user comparing against an externally-pinned value to compute it the same way (`openssl x509 -outform DER | sha256sum`).
|
||||
|
||||
**Pinning is configured, not flagged.** `tls.pinned_fingerprints` (host → expected SHA-256, capped at 32 entries, each normalized and hex-validated by a pydantic validator) follows the same posture as `dpi_check.ports`: raising the list requires editing and redeploying the config, not a `--pin` CLI argument on a diagnostic tool that anyone could point at anything.
|
||||
|
||||
**Python 3.14 is the floor, not 3.10.** The project moved off `>=3.10` when this batch of probes was added; none of the new code actually depends on a 3.11+-only construct (`ssl.SSLObject.start_tls()`, `asyncio.timeout()`, `datetime.UTC`) — the two-sequential-connection TLS design above was chosen independently of what the interpreter allows. The version floor moved because there was no longer a reason to hold it back, not because a specific new feature required it.
|
||||
|
||||
## Storage
|
||||
|
|
@ -159,4 +163,4 @@ Five sections back the L4-L7 probes, all opt-in at the CLI level and all with a
|
|||
|
||||
## Tests
|
||||
|
||||
`uv run pytest -q`. Roughly 758 tests covering the parsers, normalizers, verdict engine, DNSBL decoding, throughput math, ping statistics, jitter percentiles, serialization and the report diff. Glue is deliberately untested: live HTTP, real ICMP, Typer wiring, `psutil`/`ctypes` calls and the `--watch` timing loop. Traceroute fixtures live in `tests/fixtures/traceroute/{windows,linux,darwin}/` and include a cp866-encoded Russian Windows sample, because that encoding is exactly what breaks naive parsers. DoH response fixtures (RFC 8484 JSON form) live in `tests/fixtures/api/doh_*.json` alongside the other provider fixtures.
|
||||
`uv run pytest -q`. Roughly 777 tests covering the parsers, normalizers, verdict engine, DNSBL decoding, throughput math, ping statistics, jitter percentiles, serialization and the report diff. Glue is deliberately untested: live HTTP, real ICMP, Typer wiring, `psutil`/`ctypes` calls and the `--watch` timing loop. Traceroute fixtures live in `tests/fixtures/traceroute/{windows,linux,darwin}/` and include a cp866-encoded Russian Windows sample, because that encoding is exactly what breaks naive parsers. DoH response fixtures (RFC 8484 JSON form) live in `tests/fixtures/api/doh_*.json` alongside the other provider fixtures.
|
||||
|
|
|
|||
|
|
@ -141,6 +141,10 @@ src/netsleuth/
|
|||
|
||||
**Geo-sanity сервера speedtest сравнивает страны, а не координаты.** Изначальный дизайн предполагал расстояние по haversine в км между клиентом и сервером speedtest, для чего нужны lat/lon обеих сторон. Живая проверка собственного JSON-вывода бандл-бинарника Ookla CLI (`speedtest.exe --format=json`) показала, что его объект `server` несёт `location` (название города) и `country`, но никогда `lat`/`lon` — это документированная схема, а не особенность версии. `speed_findings()` вместо этого сравнивает `SpeedResult.server_country` с собственной геолоцированной страной клиента: грубее расстояния, но реально, всегда доступно, когда уровень каскада сообщает страну, и всё равно ловит именно тот случай, что важен — тест попал на сервер совсем в другой стране.
|
||||
|
||||
**TLS-рукопожатие сертификата сначала пробует проверенное соединение, деградирует до `CERT_NONE` только при ошибке валидации.** `probes/tls_rtt.py::tls_context(verify: bool)` по умолчанию — настоящий, немодифицированный `create_default_context()`, что меняет изначальную позицию инструмента «всегда `CERT_NONE`». `measure_tls()` сначала пробует проверенное соединение; только при `ssl.SSLCertVerificationError` повторяет без проверки (перезамеряя время повторной попытки, поскольку длительность неудачной попытки не отражает реальное соединение). Это единственный способ вообще получить `subject`/`issuer`/`notAfter`: `getpeercert()` в Python возвращает `{}` для непровалидированной цепочки, заполняя разобранный словарь, только если сертификат действительно прошёл проверку — `getpeercert(binary_form=True)` (сырой DER, хешируемый в `cert_sha256`) доступен в любом случае. Отпечаток — это SHA-256 **полного DER-сертификата, не SPKI** — извлечение только subject-public-key-info требует пакет `cryptography`, новую зависимость, которая этой фиче не нужна; сноска в отчёте говорит пользователю, сравнивающему с внешне закреплённым значением, считать его так же (`openssl x509 -outform DER | sha256sum`).
|
||||
|
||||
**Пиннинг конфигурируется, а не флагуется.** `tls.pinned_fingerprints` (хост → ожидаемый SHA-256, capped на 32 записи, каждая нормализована и провалидирована как hex pydantic-валидатором) следует той же позиции, что `dpi_check.ports`: поднять список можно только правкой и передеплоем конфига, а не аргументом `--pin` в диагностическом инструменте, который кто угодно может направить куда угодно.
|
||||
|
||||
**Python 3.14 — минимальная версия, не 3.10.** Проект ушёл с `>=3.10` одновременно с этим набором проб; ни один новый код на деле не зависит от конструкций только для 3.11+ (`ssl.SSLObject.start_tls()`, `asyncio.timeout()`, `datetime.UTC`) — схема TLS с двумя последовательными соединениями выше выбрана независимо от того, что позволяет интерпретатор. Планка версии сдвинулась потому, что не осталось причин её удерживать, а не потому, что это потребовалось для конкретной новой функции.
|
||||
|
||||
## Хранение
|
||||
|
|
@ -159,4 +163,4 @@ src/netsleuth/
|
|||
|
||||
## Тесты
|
||||
|
||||
`uv run pytest -q`. Около 758 тестов: парсеры, нормализаторы, движок вердиктов, декодирование DNSBL, математика пропускной способности, статистика пингов, перцентили джиттера, сериализация и сравнение отчётов. Клей намеренно не тестируется: живой HTTP, настоящий ICMP, обвязка Typer, вызовы `psutil`/`ctypes` и цикл `--watch`. Фикстуры трассировок лежат в `tests/fixtures/traceroute/{windows,linux,darwin}/` и включают образец русской Windows в cp866 — именно эта кодировка ломает наивные парсеры. Фикстуры DoH-ответов (формат RFC 8484 JSON) лежат в `tests/fixtures/api/doh_*.json` рядом с фикстурами других провайдеров.
|
||||
`uv run pytest -q`. Около 777 тестов: парсеры, нормализаторы, движок вердиктов, декодирование DNSBL, математика пропускной способности, статистика пингов, перцентили джиттера, сериализация и сравнение отчётов. Клей намеренно не тестируется: живой HTTP, настоящий ICMP, обвязка Typer, вызовы `psutil`/`ctypes` и цикл `--watch`. Фикстуры трассировок лежат в `tests/fixtures/traceroute/{windows,linux,darwin}/` и включают образец русской Windows в cp866 — именно эта кодировка ломает наивные парсеры. Фикстуры DoH-ответов (формат RFC 8484 JSON) лежат в `tests/fixtures/api/doh_*.json` рядом с фикстурами других провайдеров.
|
||||
|
|
|
|||
|
|
@ -65,7 +65,7 @@ None of this is required — netsleuth degrades gracefully without any of it —
|
|||
- **Reputation** — locally cached FireHOL blocklists (no query leaves your machine), Shodan InternetDB exposure, and optional classic DNSBL zones.
|
||||
- **Latency & path** — per-host ping statistics with real jitter/loss/mdev/p95/p99, and a traceroute that never needs administrator rights. The first hop is diagnosed separately from the rest of the route, so loss or slowness there is correctly attributed to your own router/Wi-Fi rather than the ISP. Every hop is annotated with its ASN, AS name and country, and flags a route that detours through a third country on the way to the destination.
|
||||
- **Bandwidth** — a cascading speedtest (Ookla binary → Cloudflare → fast.com → optional M-Lab NDT7) with separate download and upload bufferbloat grades, and a flag when the test server landed in a different country than you.
|
||||
- **TLS handshake RTT** (`--tls`) — TCP RTT, TLS 1.3 handshake time and time-to-first-byte to reference services, flagging servers whose handshake is disproportionately slow relative to the network (CPU-bound TLS termination, not a network problem).
|
||||
- **TLS handshake RTT** (`--tls`) — TCP RTT, TLS 1.3 handshake time and time-to-first-byte to reference services, flagging servers whose handshake is disproportionately slow relative to the network (CPU-bound TLS termination, not a network problem). Also captures each certificate's SHA-256 fingerprint, issuer and days until expiry, flags a validation failure, and checks against an optional pinned fingerprint (`tls.pinned_fingerprints` in `config.yaml`) to catch a corporate/ISP TLS-intercepting middlebox or a genuine MITM.
|
||||
- **DNS: system vs DoH** (`--dns-advanced`) — compares your system resolver's answers and latency against Cloudflare/Google/Quad9 DoH, and probes a bogus resolver IP to detect a transparent DNS proxy intercepting port 53.
|
||||
- **Path diversity / Anycast** (`--path-diversity`) — reads Cloudflare's `CF-RAY` edge code to find which PoP is actually serving your traffic, and flags an international routing loop when it's in a different country than you are.
|
||||
- **AS prefix benchmark** (`--prefix-bench`) — pings the first host of a handful of your own AS's announced prefixes to find the lowest-latency PoP. Capped at 256 prefixes and off by default — this samples your own network, it does not scan it.
|
||||
|
|
|
|||
|
|
@ -65,7 +65,7 @@ netsleuth
|
|||
- **Репутация** — локально кэшируемые списки FireHOL (ни один запрос не покидает машину), экспозиция по Shodan InternetDB и опционально классические зоны DNSBL.
|
||||
- **Задержки и маршрут** — статистика пингов по каждому хосту с настоящими джиттером/потерями/mdev/p95/p99 и трассировка, которой никогда не нужны права администратора. Первый хоп диагностируется отдельно от остального маршрута, поэтому потери или медлительность там правильно относятся к вашему роутеру/Wi-Fi, а не к провайдеру. Каждый хоп размечен своим ASN, именем AS и страной, а маршрут, уходящий крюком через третью страну, отмечается отдельно.
|
||||
- **Скорость** — каскадный замер (бинарник Ookla → Cloudflare → fast.com → опционально M-Lab NDT7) с раздельными оценками bufferbloat на приём и передачу и отметкой, если сервер замера оказался в другой стране.
|
||||
- **TLS handshake RTT** (`--tls`) — TCP RTT, время TLS 1.3-рукопожатия и время до первого байта к эталонным сервисам; отдельно отмечает сервер, чьё рукопожатие непропорционально медленнее сети (упирается в CPU TLS-терминатора, а не в сеть).
|
||||
- **TLS handshake RTT** (`--tls`) — TCP RTT, время TLS 1.3-рукопожатия и время до первого байта к эталонным сервисам; отдельно отмечает сервер, чьё рукопожатие непропорционально медленнее сети (упирается в CPU TLS-терминатора, а не в сеть). Также фиксирует SHA-256-отпечаток сертификата, издателя и дни до истечения, отмечает провал валидации и сверяет с опциональным закреплённым отпечатком (`tls.pinned_fingerprints` в `config.yaml`), чтобы поймать корпоративный/провайдерский TLS-перехватывающий middlebox или настоящий MITM.
|
||||
- **DNS: системный vs DoH** (`--dns-advanced`) — сравнивает ответы и задержку системного резолвера с Cloudflare/Google/Quad9 DoH, а также проверяет фиктивный IP резолвера на признаки прозрачного DNS-прокси, перехватывающего 53 порт.
|
||||
- **Диверсификация маршрута / Anycast** (`--path-diversity`) — по заголовку Cloudflare `CF-RAY` определяет, какая точка (PoP) реально обслуживает трафик, и отмечает международную петлю маршрута, если она в другой стране, чем клиент.
|
||||
- **Бенчмарк префиксов AS** (`--prefix-bench`) — пингует первый хост нескольких анонсируемых префиксов вашей собственной AS, чтобы найти точку с наименьшей задержкой. Ограничено 256 префиксами и выключено по умолчанию — это выборочная проба своей же сети, а не её сканирование.
|
||||
|
|
|
|||
|
|
@ -82,6 +82,10 @@ tls:
|
|||
port: 443
|
||||
concurrency: 4
|
||||
ttfb_path: "/"
|
||||
verify: true
|
||||
# host -> expected SHA-256 of the full DER certificate (openssl x509 -outform DER
|
||||
# | sha256sum). Colons/case are normalized. Capped at 32 entries in code.
|
||||
pinned_fingerprints: {}
|
||||
|
||||
# Opt-in via --prefix-bench. Pings the first host of a handful of your own AS's
|
||||
# announced prefixes to find the lowest-latency PoP. max_prefixes is capped at 256
|
||||
|
|
|
|||
|
|
@ -292,6 +292,8 @@ async def _tls_section(settings: Settings, options: Options) -> ModuleResult:
|
|||
timeout=settings.timeouts.http_seconds,
|
||||
concurrency=cfg.concurrency,
|
||||
source_ip=options.bind.ipv4 if options.bind else None,
|
||||
verify=cfg.verify,
|
||||
pins=cfg.pinned_fingerprints,
|
||||
)
|
||||
return ModuleResult(name="tls", status="ok", data=results)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
|
@ -129,6 +130,9 @@ class Thresholds(BaseModel):
|
|||
first_hop_ms: Band = Band(good=5.0, warn=25.0)
|
||||
|
||||
|
||||
_SHA256_HEX_RE = re.compile(r"^[0-9a-f]{64}$")
|
||||
|
||||
|
||||
class TlsConfig(BaseModel):
|
||||
targets: list[HostSpec] = Field(
|
||||
default_factory=lambda: [
|
||||
|
|
@ -139,6 +143,23 @@ class TlsConfig(BaseModel):
|
|||
port: int = 443
|
||||
concurrency: int = 4
|
||||
ttfb_path: str = "/"
|
||||
verify: bool = True
|
||||
pinned_fingerprints: dict[str, str] = Field(default_factory=dict)
|
||||
|
||||
@field_validator("pinned_fingerprints")
|
||||
@classmethod
|
||||
def _normalize_pins(cls, value: dict[str, str]) -> dict[str, str]:
|
||||
if len(value) > 32:
|
||||
raise ValueError("tls.pinned_fingerprints may list at most 32 entries")
|
||||
normalized: dict[str, str] = {}
|
||||
for host, fingerprint in value.items():
|
||||
cleaned = fingerprint.replace(":", "").strip().lower()
|
||||
if not _SHA256_HEX_RE.match(cleaned):
|
||||
raise ValueError(
|
||||
f"tls.pinned_fingerprints[{host!r}] is not a 64-character hex SHA-256 digest: {fingerprint!r}"
|
||||
)
|
||||
normalized[host] = cleaned
|
||||
return normalized
|
||||
|
||||
|
||||
class PrefixBenchConfig(BaseModel):
|
||||
|
|
|
|||
|
|
@ -730,6 +730,9 @@ def _tls(report: dict, lang: str = "en") -> list[str]:
|
|||
_t(lang, "TTFB (ms)", "TTFB (мс)"),
|
||||
_t(lang, "Version", "Версия"),
|
||||
_t(lang, "Cipher", "Шифр"),
|
||||
_t(lang, "Cert SHA-256", "Отпечаток SHA-256"),
|
||||
_t(lang, "Issuer", "Издатель"),
|
||||
_t(lang, "Expires in (days)", "Истекает через (дней)"),
|
||||
],
|
||||
[
|
||||
[
|
||||
|
|
@ -740,17 +743,27 @@ def _tls(report: dict, lang: str = "en") -> list[str]:
|
|||
_num(r.get("ttfb_ms")),
|
||||
r.get("tls_version") or "—",
|
||||
r.get("cipher") or "—",
|
||||
(r.get("cert_sha256") or "—")[:16] + ("…" if r.get("cert_sha256") else ""),
|
||||
r.get("cert_issuer") or "—",
|
||||
_num(r.get("cert_days_remaining")),
|
||||
]
|
||||
for r in results
|
||||
],
|
||||
lang,
|
||||
)
|
||||
mismatches = [r for r in results if r.get("pin_verdict") == "mismatch"]
|
||||
if mismatches:
|
||||
lines += [""] + [
|
||||
f"- ⚠️ {r.get('label')}: {_t(lang, 'pinned fingerprint mismatch', 'отпечаток не совпадает с закреплённым')}"
|
||||
for r in mismatches
|
||||
]
|
||||
errored = [r for r in results if r.get("error")]
|
||||
if errored:
|
||||
lines += [""] + [f"- {r.get('label')}: {r.get('error')}" for r in errored]
|
||||
lines += [
|
||||
"",
|
||||
f"> {_t(lang, 'TLS handshake time is derived by subtracting a separate TCP-only baseline connection from a second, full TLS connection.', 'Время TLS-рукопожатия получено вычитанием отдельного базового TCP-соединения из времени второго, полного TLS-соединения.')}",
|
||||
f"> {_t(lang, 'The certificate fingerprint is the SHA-256 of the full DER certificate, not the SPKI — compute it as `openssl x509 -outform DER | sha256sum` to compare against an externally-pinned value.', 'Отпечаток сертификата — это SHA-256 полного DER-сертификата, а не SPKI — считайте его как `openssl x509 -outform DER | sha256sum`, чтобы сравнить с внешне закреплённым значением.')}",
|
||||
]
|
||||
return lines + [""]
|
||||
|
||||
|
|
|
|||
|
|
@ -615,6 +615,56 @@ def tls_findings(results: list[TlsResult], t: Thresholds) -> list[Finding]:
|
|||
advice_ru="Медленный TTFB при быстром рукопожатии указывает на обработку на стороне сервера, а не на сеть.",
|
||||
)
|
||||
)
|
||||
if r.pin_verdict == "mismatch":
|
||||
findings.append(
|
||||
Finding(
|
||||
id=f"tls.cert_pin_mismatch.{r.label}",
|
||||
severity="crit",
|
||||
title=f"Certificate for {r.host} does not match the pinned fingerprint",
|
||||
detail=f"Observed SHA-256 {r.cert_sha256}, which does not match the configured pin.",
|
||||
metric="cert_sha256",
|
||||
value=r.cert_sha256,
|
||||
advice="A corporate/ISP TLS-intercepting middlebox or a genuine MITM; "
|
||||
"compare against the fingerprint seen from another network before assuming the worst.",
|
||||
title_ru=f"Сертификат {r.host} не совпадает с закреплённым отпечатком",
|
||||
detail_ru=f"Наблюдаемый SHA-256 {r.cert_sha256} не совпадает с настроенным отпечатком.",
|
||||
advice_ru="Корпоративный/провайдерский TLS-перехватывающий middlebox или настоящий MITM; "
|
||||
"сравните с отпечатком, увиденным из другой сети, прежде чем делать выводы.",
|
||||
)
|
||||
)
|
||||
elif r.cert_verified is False and r.pin_verdict == "unpinned":
|
||||
findings.append(
|
||||
Finding(
|
||||
id=f"tls.cert_unverified.{r.label}",
|
||||
severity="warn",
|
||||
title=f"Certificate chain for {r.host} did not validate",
|
||||
detail=f"Issuer: {r.cert_issuer or 'unknown'}.",
|
||||
metric="cert_issuer",
|
||||
value=r.cert_issuer,
|
||||
advice="Often the issuer of an interception appliance rather than the real service's CA; "
|
||||
"pin the expected fingerprint in tls.pinned_fingerprints to escalate this to a hard alert.",
|
||||
title_ru=f"Цепочка сертификатов {r.host} не прошла проверку",
|
||||
detail_ru=f"Издатель: {r.cert_issuer or 'неизвестен'}.",
|
||||
advice_ru="Часто это издатель перехватывающего устройства, а не настоящий CA сервиса; "
|
||||
"закрепите ожидаемый отпечаток в tls.pinned_fingerprints, чтобы превратить это в жёсткий алерт.",
|
||||
)
|
||||
)
|
||||
if r.cert_days_remaining is not None and 0 <= r.cert_days_remaining < 14:
|
||||
findings.append(
|
||||
Finding(
|
||||
id=f"tls.cert_expiring.{r.label}",
|
||||
severity="warn",
|
||||
title=f"Certificate for {r.host} expires soon",
|
||||
detail=f"{r.cert_days_remaining} day(s) remaining ({r.cert_not_after}).",
|
||||
metric="cert_days_remaining",
|
||||
value=r.cert_days_remaining,
|
||||
threshold=14,
|
||||
advice="Renew before it lapses; an expired certificate breaks the service for every client.",
|
||||
title_ru=f"Сертификат {r.host} скоро истекает",
|
||||
detail_ru=f"Осталось {r.cert_days_remaining} дн. (до {r.cert_not_after}).",
|
||||
advice_ru="Продлите до истечения; просроченный сертификат сломает сервис для всех клиентов.",
|
||||
)
|
||||
)
|
||||
return findings
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -364,6 +364,12 @@ class TlsResult:
|
|||
cipher: str | None = None
|
||||
alpn: str | None = None
|
||||
cert_verified: bool | None = None
|
||||
cert_sha256: str | None = None
|
||||
cert_subject: str | None = None
|
||||
cert_issuer: str | None = None
|
||||
cert_not_after: str | None = None
|
||||
cert_days_remaining: int | None = None
|
||||
pin_verdict: str = "unpinned"
|
||||
error: str | None = None
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,8 +1,10 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
import ssl
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from netsleuth.models import TlsResult
|
||||
from netsleuth.probes.latency import tcp_connect_rtt
|
||||
|
|
@ -23,14 +25,44 @@ def cpu_bound_ratio(tls_handshake_ms: float | None, tcp_rtt_ms: float | None) ->
|
|||
return tls_handshake_ms / tcp_rtt_ms
|
||||
|
||||
|
||||
def tls_context() -> ssl.SSLContext:
|
||||
def tls_context(verify: bool = True) -> ssl.SSLContext:
|
||||
ctx = ssl.create_default_context()
|
||||
if not verify:
|
||||
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:
|
||||
pin = pins.get(host)
|
||||
if not pin:
|
||||
return "unpinned"
|
||||
if actual_sha256 is None:
|
||||
return "mismatch"
|
||||
normalized_pin = pin.replace(":", "").strip().lower()
|
||||
normalized_actual = actual_sha256.replace(":", "").strip().lower()
|
||||
return "match" if normalized_pin == normalized_actual else "mismatch"
|
||||
|
||||
|
||||
def cert_name(rdn_sequence: tuple | None) -> str | None:
|
||||
if not rdn_sequence:
|
||||
return None
|
||||
parts = [f"{key}={value}" for rdn in rdn_sequence for key, value in rdn]
|
||||
return ", ".join(parts) or None
|
||||
|
||||
|
||||
def days_remaining(not_after: str | None, now: datetime | None = None) -> int | None:
|
||||
if not not_after:
|
||||
return None
|
||||
try:
|
||||
expires_ts = ssl.cert_time_to_seconds(not_after)
|
||||
except ValueError:
|
||||
return None
|
||||
expires = datetime.fromtimestamp(expires_ts, tz=timezone.utc)
|
||||
return (expires - (now or datetime.now(timezone.utc))).days
|
||||
|
||||
|
||||
async def measure_tls(
|
||||
label: str,
|
||||
host: str,
|
||||
|
|
@ -38,21 +70,40 @@ async def measure_tls(
|
|||
port: int = 443,
|
||||
timeout: float = 3.0,
|
||||
source_ip: str | None = None,
|
||||
verify: bool = True,
|
||||
pins: dict[str, str] | None = None,
|
||||
) -> TlsResult:
|
||||
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)
|
||||
result.tcp_rtt_ms = tcp_rtt_ms
|
||||
|
||||
local_addr = (source_ip, 0) if source_ip else None
|
||||
began = time.perf_counter()
|
||||
reader, writer = await asyncio.wait_for(
|
||||
|
||||
async def _connect(do_verify: bool):
|
||||
return await asyncio.wait_for(
|
||||
asyncio.open_connection(
|
||||
host, port, ssl=tls_context(), server_hostname=host, local_addr=local_addr
|
||||
host, port, ssl=tls_context(verify=do_verify), server_hostname=host, local_addr=local_addr
|
||||
),
|
||||
timeout=timeout,
|
||||
)
|
||||
|
||||
began = time.perf_counter()
|
||||
cert_verified = verify
|
||||
try:
|
||||
reader, writer = await _connect(verify)
|
||||
except ssl.SSLCertVerificationError:
|
||||
if not verify:
|
||||
raise
|
||||
# The failed handshake's time is not representative of a real connection;
|
||||
# only the successful (unverified) attempt's duration feeds tls_handshake_ms.
|
||||
began = time.perf_counter()
|
||||
reader, writer = await _connect(False)
|
||||
cert_verified = False
|
||||
tls_total_ms = (time.perf_counter() - began) * 1000.0
|
||||
result.cert_verified = cert_verified
|
||||
result.resolved_ip = (writer.get_extra_info("peername") or (None,))[0]
|
||||
|
||||
ssl_object = writer.get_extra_info("ssl_object")
|
||||
if ssl_object is not None:
|
||||
|
|
@ -60,6 +111,18 @@ async def measure_tls(
|
|||
cipher = ssl_object.cipher()
|
||||
result.cipher = cipher[0] if cipher else None
|
||||
result.alpn = ssl_object.selected_alpn_protocol()
|
||||
der = ssl_object.getpeercert(binary_form=True)
|
||||
if der:
|
||||
result.cert_sha256 = hashlib.sha256(der).hexdigest()
|
||||
result.pin_verdict = fingerprint_verdict(host, result.cert_sha256, pins)
|
||||
if cert_verified:
|
||||
# getpeercert() only returns the parsed dict for a chain that
|
||||
# actually validated -- under CERT_NONE it is always {}.
|
||||
parsed = ssl_object.getpeercert() or {}
|
||||
result.cert_subject = cert_name(parsed.get("subject"))
|
||||
result.cert_issuer = cert_name(parsed.get("issuer"))
|
||||
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()
|
||||
began_ttfb = time.perf_counter()
|
||||
|
|
@ -89,11 +152,15 @@ async def tls_fanout(
|
|||
timeout: float,
|
||||
concurrency: int,
|
||||
source_ip: str | None = None,
|
||||
verify: bool = True,
|
||||
pins: dict[str, str] | None = None,
|
||||
) -> list[TlsResult]:
|
||||
semaphore = asyncio.Semaphore(concurrency)
|
||||
|
||||
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)
|
||||
return await measure_tls(
|
||||
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)))
|
||||
|
|
|
|||
|
|
@ -191,6 +191,39 @@ def test_dpi_check_concurrency_is_capped_as_a_rate_limit(tmp_path):
|
|||
load_settings(config_path=path, env_file=tmp_path / "nope.env")
|
||||
|
||||
|
||||
def test_tls_pinned_fingerprints_accepts_valid_hex_and_normalizes_case_and_colons(tmp_path):
|
||||
path = tmp_path / "config.yaml"
|
||||
fingerprint_with_colons = ":".join(["AB"] * 32)
|
||||
path.write_text(
|
||||
f'tls:\n pinned_fingerprints: {{"example.com": "{fingerprint_with_colons}"}}\n',
|
||||
encoding="utf-8",
|
||||
)
|
||||
s = load_settings(config_path=path, env_file=tmp_path / "nope.env")
|
||||
assert s.tls.pinned_fingerprints["example.com"] == "ab" * 32
|
||||
|
||||
|
||||
def test_tls_pinned_fingerprints_rejects_a_short_value(tmp_path):
|
||||
path = tmp_path / "config.yaml"
|
||||
path.write_text('tls:\n pinned_fingerprints: {"example.com": "aabbcc"}\n', encoding="utf-8")
|
||||
with pytest.raises(ValidationError, match="64-character hex"):
|
||||
load_settings(config_path=path, env_file=tmp_path / "nope.env")
|
||||
|
||||
|
||||
def test_tls_pinned_fingerprints_rejects_non_hex_characters(tmp_path):
|
||||
path = tmp_path / "config.yaml"
|
||||
path.write_text('tls:\n pinned_fingerprints: {"example.com": "' + "zz" * 32 + '"}\n', encoding="utf-8")
|
||||
with pytest.raises(ValidationError, match="64-character hex"):
|
||||
load_settings(config_path=path, env_file=tmp_path / "nope.env")
|
||||
|
||||
|
||||
def test_tls_pinned_fingerprints_are_capped_at_32_entries(tmp_path):
|
||||
path = tmp_path / "config.yaml"
|
||||
entries = ", ".join(f'"host{i}.example.com": "{"aa" * 32}"' for i in range(33))
|
||||
path.write_text("tls:\n pinned_fingerprints: {" + entries + "}\n", encoding="utf-8")
|
||||
with pytest.raises(ValidationError, match="at most 32"):
|
||||
load_settings(config_path=path, env_file=tmp_path / "nope.env")
|
||||
|
||||
|
||||
def test_captive_portal_check_urls_are_capped(tmp_path):
|
||||
path = tmp_path / "config.yaml"
|
||||
path.write_text(
|
||||
|
|
|
|||
|
|
@ -99,6 +99,37 @@ def test_tls_findings_do_not_crash_on_a_result_with_no_measurements(thresholds):
|
|||
assert tls_findings([tls(tcp_rtt_ms=None, tls_handshake_ms=None, ttfb_ms=None)], thresholds) == []
|
||||
|
||||
|
||||
def test_pin_mismatch_fires_a_crit_finding(thresholds):
|
||||
findings = tls_findings([tls(pin_verdict="mismatch", cert_sha256="a" * 64)], thresholds)
|
||||
assert [f.id for f in findings] == ["tls.cert_pin_mismatch.cloudflare"]
|
||||
assert findings[0].severity == "crit"
|
||||
|
||||
|
||||
def test_unverified_cert_without_a_pin_fires_a_warn_finding(thresholds):
|
||||
findings = tls_findings([tls(cert_verified=False, cert_issuer="Corp Intercept CA")], thresholds)
|
||||
assert [f.id for f in findings] == ["tls.cert_unverified.cloudflare"]
|
||||
assert findings[0].severity == "warn"
|
||||
assert "Corp Intercept CA" in findings[0].detail
|
||||
|
||||
|
||||
def test_a_verified_unpinned_cert_produces_neither_pin_nor_unverified_findings(thresholds):
|
||||
findings = tls_findings([tls(cert_verified=True, pin_verdict="unpinned")], thresholds)
|
||||
ids = [f.id for f in findings]
|
||||
assert "tls.cert_pin_mismatch.cloudflare" not in ids
|
||||
assert "tls.cert_unverified.cloudflare" not in ids
|
||||
|
||||
|
||||
def test_expiring_cert_fires_a_warn_finding(thresholds):
|
||||
findings = tls_findings([tls(cert_days_remaining=5, cert_not_after="Jun 1 12:00:00 2026 GMT")], thresholds)
|
||||
assert [f.id for f in findings] == ["tls.cert_expiring.cloudflare"]
|
||||
assert findings[0].severity == "warn"
|
||||
|
||||
|
||||
def test_cert_with_plenty_of_days_left_does_not_fire(thresholds):
|
||||
findings = tls_findings([tls(cert_days_remaining=120)], thresholds)
|
||||
assert "tls.cert_expiring.cloudflare" not in [f.id for f in findings]
|
||||
|
||||
|
||||
def prefix_probe(**kw) -> PrefixProbe:
|
||||
base = dict(prefix="203.0.113.0/24", probe_ip="203.0.113.1", avg_ms=10.0, reachable=True)
|
||||
base.update(kw)
|
||||
|
|
|
|||
|
|
@ -2,9 +2,13 @@ from __future__ import annotations
|
|||
|
||||
import socket
|
||||
import ssl
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from netsleuth.probes.tls_rtt import (
|
||||
cert_name,
|
||||
cpu_bound_ratio,
|
||||
days_remaining,
|
||||
fingerprint_verdict,
|
||||
measure_tls,
|
||||
split_timings,
|
||||
tls_context,
|
||||
|
|
@ -51,13 +55,60 @@ def test_cpu_bound_ratio_is_none_when_tcp_rtt_is_zero():
|
|||
assert cpu_bound_ratio(30.0, 0.0) is None
|
||||
|
||||
|
||||
def test_tls_context_disables_hostname_checking_for_diagnostics():
|
||||
def test_tls_context_verifies_by_default():
|
||||
ctx = tls_context()
|
||||
assert isinstance(ctx, ssl.SSLContext)
|
||||
assert ctx.check_hostname is True
|
||||
assert ctx.verify_mode == ssl.CERT_REQUIRED
|
||||
|
||||
|
||||
def test_tls_context_disables_verification_when_asked():
|
||||
ctx = tls_context(verify=False)
|
||||
assert ctx.check_hostname is False
|
||||
assert ctx.verify_mode == ssl.CERT_NONE
|
||||
|
||||
|
||||
def test_fingerprint_verdict_matches_case_and_colon_insensitively():
|
||||
pins = {"example.com": "AA:BB:CC:DD"}
|
||||
assert fingerprint_verdict("example.com", "aabbccdd", pins) == "match"
|
||||
|
||||
|
||||
def test_fingerprint_verdict_mismatch():
|
||||
pins = {"example.com": "aabbccdd"}
|
||||
assert fingerprint_verdict("example.com", "11223344", pins) == "mismatch"
|
||||
|
||||
|
||||
def test_fingerprint_verdict_unpinned_when_host_has_no_pin():
|
||||
assert fingerprint_verdict("example.com", "aabbccdd", {}) == "unpinned"
|
||||
|
||||
|
||||
def test_fingerprint_verdict_mismatch_when_actual_is_none():
|
||||
assert fingerprint_verdict("example.com", None, {"example.com": "aabbccdd"}) == "mismatch"
|
||||
|
||||
|
||||
def test_cert_name_flattens_the_rdn_sequence():
|
||||
subject = ((("commonName", "example.com"),), (("organizationName", "Example Inc"),))
|
||||
assert cert_name(subject) == "commonName=example.com, organizationName=Example Inc"
|
||||
|
||||
|
||||
def test_cert_name_handles_empty_input():
|
||||
assert cert_name(None) is None
|
||||
assert cert_name(()) is None
|
||||
|
||||
|
||||
def test_days_remaining_computes_from_a_fixed_now():
|
||||
now = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
assert days_remaining("Jan 11 00:00:00 2026 GMT", now=now) == 10
|
||||
|
||||
|
||||
def test_days_remaining_handles_a_malformed_string():
|
||||
assert days_remaining("not a date") is None
|
||||
|
||||
|
||||
def test_days_remaining_handles_missing_input():
|
||||
assert days_remaining(None) is None
|
||||
|
||||
|
||||
async def test_measure_tls_reports_an_error_result_for_a_closed_port_without_raising():
|
||||
port = _closed_port()
|
||||
result = await measure_tls("dead", "127.0.0.1", port=port, timeout=1.0)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue