feat: calibrated confidence levels in verdicts and summary

This commit is contained in:
MayersScott 2026-05-06 11:17:45 +03:00
parent 30c3f47c7a
commit 9776ace195
5 changed files with 203 additions and 78 deletions

View file

@ -10,7 +10,7 @@ import requests
from . import dns as dns_mod
from . import http as http_mod
from . import network
from .models import CheckResult, Verdict
from .models import CheckResult, Confidence, Verdict
logger = logging.getLogger(__name__)
@ -36,19 +36,29 @@ def check_url(name: str, url: str, timeout: float = 5.0) -> CheckResult:
if res.sys_ip is None and res.doh_ip is not None:
res.verdict = Verdict.DNS_BLOCK
res.confidence = Confidence.HIGH
res.dns_error = "system resolver failed, DoH succeeded"
res.notes.append("system DNS doesn't resolve, DoH does — DNS poisoning")
res.notes.append(
"system DNS doesn't resolve, DoH does — consistent with DNS poisoning"
)
return res
if res.sys_ip is None and res.doh_ip is None:
res.verdict = Verdict.DOWN
res.confidence = Confidence.LOW
res.dns_error = "domain not resolved anywhere"
res.notes.append("domain doesn't resolve via system DNS or DoH")
res.notes.append(
"domain doesn't resolve via system DNS or DoH — could be NXDOMAIN, "
"downed authoritative server, or DNS-level block"
)
return res
if res.sys_ip and res.doh_ip and res.sys_ip != res.doh_ip:
res.dns_mismatch = True
res.notes.append(f"DNS mismatch: sys={res.sys_ip} vs doh={res.doh_ip}")
res.notes.append(
f"DNS mismatch: sys={res.sys_ip} vs doh={res.doh_ip} "
"(may indicate transparent DNS rewriting)"
)
res.tcp_ok, res.tcp_time_ms, res.tcp_error = network.check_tcp(
host, timeout=timeout
@ -56,12 +66,21 @@ def check_url(name: str, url: str, timeout: float = 5.0) -> CheckResult:
if not res.tcp_ok:
if res.tcp_error == "timeout":
res.verdict = Verdict.TIMEOUT
res.notes.append("TCP timeout — port 443 unreachable")
res.confidence = Confidence.LOW
res.notes.append(
"TCP timeout on port 443 — could be IP block, route loss, "
"or upstream congestion"
)
elif "reset" in (res.tcp_error or ""):
res.verdict = Verdict.TCP_RESET
res.notes.append("TCP RST — IP-level block")
res.confidence = Confidence.MEDIUM
res.notes.append(
"TCP RST received — pattern matches RST injection by a "
"middlebox, but a busy server can also send RST"
)
else:
res.verdict = Verdict.DOWN
res.confidence = Confidence.LOW
res.notes.append(f"TCP failed: {res.tcp_error}")
return res
@ -75,12 +94,21 @@ def check_url(name: str, url: str, timeout: float = 5.0) -> CheckResult:
err = (res.tls_error or "").lower()
if "reset" in err:
res.verdict = Verdict.TLS_BLOCK
res.notes.append("TLS reset — DPI cutting on SNI (typical RKN/TSPU)")
res.confidence = Confidence.MEDIUM
res.notes.append(
"TLS reset right after ClientHello — consistent with SNI-based "
"DPI filtering (typical TSPU/RKN signature), not proof"
)
elif "timeout" in err:
res.verdict = Verdict.TLS_BLOCK
res.notes.append("TLS timeout — silent drop after ClientHello")
res.confidence = Confidence.MEDIUM
res.notes.append(
"TLS handshake silently dropped — consistent with DPI filtering "
"by ClientHello, but could be a flaky path"
)
else:
res.verdict = Verdict.TLS_BLOCK
res.confidence = Confidence.LOW
res.notes.append(f"TLS error: {res.tls_error}")
return res
@ -91,22 +119,29 @@ def check_url(name: str, url: str, timeout: float = 5.0) -> CheckResult:
if probe.timed_out:
res.verdict = Verdict.TIMEOUT
res.confidence = Confidence.LOW
return res
if probe.error:
res.verdict = Verdict.DOWN
res.confidence = Confidence.LOW
return res
if res.status_code == 451:
res.verdict = Verdict.HTTP_STUB
res.notes.append("HTTP 451 — Unavailable For Legal Reasons")
res.confidence = Confidence.HIGH
res.notes.append("HTTP 451 — Unavailable For Legal Reasons (explicit)")
return res
if http_mod.looks_like_stub(probe.body_snippet):
res.verdict = Verdict.HTTP_STUB
res.notes.append("response body matches an ISP stub-page marker")
res.confidence = Confidence.HIGH
res.notes.append(
"response body matches a known ISP stub-page marker"
)
return res
res.verdict = Verdict.OK
res.confidence = Confidence.HIGH
return res

View file

@ -16,6 +16,23 @@ class Verdict(str, Enum):
UNKNOWN = "UNKNOWN"
class Confidence(str, Enum):
"""How confident we are in a verdict.
HIGH - two independent signals confirm the diagnosis (e.g. DNS poisoning
confirmed by DoH returning a different IP, or an explicit HTTP 451,
or a known stub-page marker in the body)
MEDIUM - a known censorship pattern matches, but a single signal can't
rule out a server-side issue or a flaky network (e.g. TCP RST,
TLS handshake aborted on ClientHello)
LOW - symptom is ambiguous (timeout, generic failure) and could be
caused by anything from DPI to a flaky uplink
"""
HIGH = "HIGH"
MEDIUM = "MEDIUM"
LOW = "LOW"
BLOCKED_VERDICTS: frozenset[Verdict] = frozenset({
Verdict.DNS_BLOCK,
Verdict.TCP_RESET,
@ -31,6 +48,7 @@ class CheckResult:
url: str
verdict: Verdict = Verdict.UNKNOWN
confidence: Confidence = Confidence.LOW
notes: list[str] = field(default_factory=list)
sys_ip: Optional[str] = None
@ -54,4 +72,5 @@ class CheckResult:
def to_dict(self) -> dict[str, Any]:
d = asdict(self)
d["verdict"] = self.verdict.value
return d
d["confidence"] = self.confidence.value
return d

View file

@ -4,7 +4,7 @@ import os
import sys
from collections import Counter
from .models import BLOCKED_VERDICTS, CheckResult, Verdict
from .models import BLOCKED_VERDICTS, CheckResult, Confidence, Verdict
class C:
@ -29,16 +29,27 @@ if not _colors_enabled():
setattr(C, _attr, "")
VERDICT_STYLE: dict[Verdict, tuple[str, str]] = {
Verdict.OK: (C.GREEN, "✓ OK"),
Verdict.DNS_BLOCK: (C.RED, "✗ DNS BLOCK"),
Verdict.TCP_RESET: (C.RED, "✗ TCP RESET"),
Verdict.TLS_BLOCK: (C.RED, "✗ TLS BLOCK"),
Verdict.HTTP_STUB: (C.RED, "✗ HTTP STUB"),
Verdict.TIMEOUT: (C.YELLOW, "⚠ TIMEOUT"),
Verdict.DOWN: (C.GRAY, "· DOWN"),
Verdict.UNKNOWN: (C.GRAY, "? UNKNOWN"),
}
def _label_for(verdict: Verdict, confidence: Confidence) -> tuple[str, str]:
if verdict == Verdict.OK:
return C.GREEN, "✓ OK"
if verdict == Verdict.DOWN:
return C.GRAY, "· DOWN"
if verdict == Verdict.UNKNOWN:
return C.GRAY, "? UNKNOWN"
base = {
Verdict.DNS_BLOCK: "DNS",
Verdict.TCP_RESET: "TCP RESET",
Verdict.TLS_BLOCK: "TLS DPI",
Verdict.HTTP_STUB: "HTTP STUB",
Verdict.TIMEOUT: "TIMEOUT",
}.get(verdict, verdict.value)
if confidence == Confidence.HIGH:
return C.RED, f"{base}"
if confidence == Confidence.MEDIUM:
return C.YELLOW, f"~ LIKELY {base}"
return C.GRAY, f"? {base}?"
def print_header(info: dict) -> None:
@ -58,23 +69,23 @@ def print_header(info: dict) -> None:
def print_section(title: str) -> None:
print(f"\n{C.BOLD}{title}{C.RESET}")
print(
f" {C.DIM}{'name':<14}{'verdict':<14}"
f" {C.DIM}{'name':<14}{'verdict':<22}"
f"{'TCP':>8}{'TLS':>8}{'PLT':>8} {'status':<6}{C.RESET}"
)
print(f" {C.DIM}{'-' * 60}{C.RESET}")
print(f" {C.DIM}{'-' * 68}{C.RESET}")
def print_result(r: CheckResult) -> None:
color, label = VERDICT_STYLE.get(r.verdict, (C.GRAY, r.verdict.value))
color, label = _label_for(r.verdict, r.confidence)
status = str(r.status_code) if r.status_code else ""
tcp = f"{r.tcp_time_ms:.0f}ms" if r.tcp_time_ms is not None else ""
tls = f"{r.tls_time_ms:.0f}ms" if r.tls_time_ms is not None else ""
plt = f"{r.plt_ms:.0f}ms" if r.plt_ms is not None else ""
status = str(r.status_code) if r.status_code else "-"
tcp = f"{r.tcp_time_ms:.0f}ms" if r.tcp_time_ms is not None else "-"
tls = f"{r.tls_time_ms:.0f}ms" if r.tls_time_ms is not None else "-"
plt = f"{r.plt_ms:.0f}ms" if r.plt_ms is not None else "-"
print(
f" {r.name:<14}"
f"{color}{label:<14}{C.RESET}"
f"{color}{label:<22}{C.RESET}"
f"{tcp:>8}{tls:>8}{plt:>8} "
f"{status:<6}"
)
@ -86,6 +97,10 @@ def print_summary(white: list[CheckResult], black: list[CheckResult]) -> None:
white_ok = sum(1 for r in white if r.verdict == Verdict.OK)
black_ok = sum(1 for r in black if r.verdict == Verdict.OK)
black_blocked = sum(1 for r in black if r.verdict in BLOCKED_VERDICTS)
black_high_conf = sum(
1 for r in black
if r.verdict in BLOCKED_VERDICTS and r.confidence == Confidence.HIGH
)
print(f"\n{C.BOLD}{C.CYAN}{'=' * 70}{C.RESET}")
print(f"{C.BOLD} Summary{C.RESET}")
@ -96,18 +111,19 @@ def print_summary(white: list[CheckResult], black: list[CheckResult]) -> None:
f"{black_blocked}/{len(black)} blocked"
)
color, verdict = _summary_verdict(
white_ok, len(white), black_ok, black_blocked, len(black)
color, verdict, conf_note = _summary_verdict(
white_ok, len(white), black_ok, black_blocked, len(black),
black_high_conf,
)
print(f"\n {color}{C.BOLD}{verdict}{C.RESET}")
if conf_note:
print(f" {C.DIM} {conf_note}{C.RESET}")
types = Counter(r.verdict for r in black if r.verdict in BLOCKED_VERDICTS)
if types:
print(f"\n {C.DIM}Block types in the blacklist:{C.RESET}")
for verdict_type, count in types.most_common():
type_color, label = VERDICT_STYLE.get(
verdict_type, (C.GRAY, verdict_type.value)
)
type_color, label = _label_for(verdict_type, Confidence.HIGH)
print(f" {type_color}{label}{C.RESET}: {count}")
print(f"{C.BOLD}{C.CYAN}{'=' * 70}{C.RESET}\n")
@ -119,11 +135,50 @@ def _summary_verdict(
black_ok: int,
black_blocked: int,
black_total: int,
) -> tuple[str, str]:
if white_ok < white_total / 2:
return C.YELLOW, "Connectivity is degraded — even the whitelist barely loads."
black_high_conf: int = 0,
) -> tuple[str, str, str]:
"""Return (color, verdict line, confidence note).
The confidence note explains *why* we report what we report - and in
particular whether the whitelist control is healthy enough for the
blacklist signal to be trustworthy.
"""
if white_total > 0 and white_ok < white_total / 2:
return (
C.YELLOW,
"Inconclusive - control whitelist is also failing.",
"Can't separate censorship from a broken uplink without a working "
"baseline. Try a different network, or check the local connection.",
)
if black_blocked == 0 and black_ok == black_total:
return C.GREEN, "You're NOT in an RKN-blocked zone (or you're using a VPN)."
return (
C.GREEN,
"Likely NOT in an RKN-blocked zone (or VPN is masking it).",
"All blacklisted sites loaded - either you're outside the blocked "
"zone, or your VPN/proxy is intercepting the traffic.",
)
if black_blocked >= black_total * 0.7:
return C.RED, "You ARE in an RKN-blocked zone."
return C.YELLOW, "Partial blocks — some blacklisted sites still load."
if black_total > 0 and black_high_conf >= black_total * 0.5:
return (
C.RED,
"Likely in an RKN-blocked zone (high confidence).",
f"{black_high_conf}/{black_total} blacklist failures match "
"high-confidence patterns (DNS poisoning confirmed by DoH, "
"HTTP 451, known stub-page markers).",
)
return (
C.RED,
"Likely in an RKN-blocked zone (medium confidence).",
"Most blacklist failures match censorship patterns (TLS DPI, "
"TCP RST), but those signals can also be caused by server-side "
"issues. A control vantage point would confirm.",
)
return (
C.YELLOW,
"Partial blocks - some blacklisted sites still load.",
"Mixed signals. May indicate selective filtering, a mix of real "
"blocks and unrelated server issues, or a CDN flake.",
)

View file

@ -2,7 +2,7 @@ from unittest.mock import patch
from rkn_checker.core import check_url
from rkn_checker.http import HttpProbe
from rkn_checker.models import Verdict
from rkn_checker.models import Confidence, Verdict
def _patches(
@ -38,18 +38,17 @@ class TestVerdictPath:
def test_happy_path_yields_ok(self):
r = _run_with(_patches())
assert r.verdict == Verdict.OK
assert r.tcp_ok is True
assert r.tls_ok is True
assert r.status_code == 200
assert r.confidence == Confidence.HIGH
def test_dns_block_when_system_fails_but_doh_works(self):
r = _run_with(_patches(sys_ip=None, doh_ip="1.2.3.4"))
assert r.verdict == Verdict.DNS_BLOCK
assert r.dns_error is not None
assert r.confidence == Confidence.HIGH
def test_down_when_neither_resolver_finds_domain(self):
r = _run_with(_patches(sys_ip=None, doh_ip=None))
assert r.verdict == Verdict.DOWN
assert r.confidence == Confidence.LOW
def test_dns_mismatch_is_flagged_but_not_fatal(self):
r = _run_with(_patches(sys_ip="1.1.1.1", doh_ip="2.2.2.2"))
@ -59,38 +58,39 @@ class TestVerdictPath:
def test_tcp_timeout_yields_timeout_verdict(self):
r = _run_with(_patches(tcp=(False, None, "timeout")))
assert r.verdict == Verdict.TIMEOUT
assert r.confidence == Confidence.LOW
def test_tcp_reset_yields_tcp_reset_verdict(self):
r = _run_with(_patches(tcp=(False, None, "connection reset")))
assert r.verdict == Verdict.TCP_RESET
assert r.confidence == Confidence.MEDIUM
def test_tcp_other_failure_yields_down(self):
r = _run_with(_patches(tcp=(False, None, "OSError: no route to host")))
assert r.verdict == Verdict.DOWN
def test_tls_reset_is_classified_as_tls_block(self):
r = _run_with(
_patches(tls=(False, None, None, "connection reset during TLS"))
)
r = _run_with(_patches(tls=(False, None, None, "connection reset during TLS")))
assert r.verdict == Verdict.TLS_BLOCK
assert r.confidence == Confidence.MEDIUM
def test_tls_timeout_is_also_a_tls_block(self):
r = _run_with(_patches(tls=(False, None, None, "timeout")))
assert r.verdict == Verdict.TLS_BLOCK
assert r.confidence == Confidence.MEDIUM
def test_http_451_is_an_http_stub(self):
probe = HttpProbe(status_code=451, elapsed_ms=50.0, body_snippet="")
r = _run_with(_patches(http=probe))
assert r.verdict == Verdict.HTTP_STUB
assert r.confidence == Confidence.HIGH
def test_http_stub_marker_in_body(self):
probe = HttpProbe(
status_code=200,
elapsed_ms=50.0,
body_snippet="доступ ограничен по решению",
)
probe = HttpProbe(status_code=200, elapsed_ms=50.0,
body_snippet="доступ ограничен по решению")
r = _run_with(_patches(http=probe))
assert r.verdict == Verdict.HTTP_STUB
assert r.confidence == Confidence.HIGH
def test_http_timeout_yields_timeout(self):
probe = HttpProbe(error="timeout", timed_out=True)
@ -98,10 +98,8 @@ class TestVerdictPath:
assert r.verdict == Verdict.TIMEOUT
def test_normal_200_is_ok(self):
probe = HttpProbe(
status_code=200,
elapsed_ms=50.0,
body_snippet="<html><body>welcome</body></html>",
)
probe = HttpProbe(status_code=200, elapsed_ms=50.0,
body_snippet="<html><body>welcome</body></html>")
r = _run_with(_patches(http=probe))
assert r.verdict == Verdict.OK
assert r.confidence == Confidence.HIGH

View file

@ -2,39 +2,57 @@ from rkn_checker.output import _summary_verdict
class TestSummaryVerdict:
def test_degraded_connection_when_whitelist_mostly_fails(self):
_, msg = _summary_verdict(
white_ok=5, white_total=20, black_ok=0, black_blocked=15, black_total=15
def test_inconclusive_when_whitelist_mostly_fails(self):
_, msg, note = _summary_verdict(
white_ok=5, white_total=20,
black_ok=0, black_blocked=15, black_total=15,
)
assert "degraded" in msg.lower()
assert "inconclusive" in msg.lower()
def test_no_blocks_when_blacklist_fully_open(self):
_, msg = _summary_verdict(
white_ok=20, white_total=20, black_ok=15, black_blocked=0, black_total=15
_, msg, _ = _summary_verdict(
white_ok=20, white_total=20,
black_ok=15, black_blocked=0, black_total=15,
)
assert "not in" in msg.lower()
def test_blocked_zone_when_majority_of_blacklist_fails(self):
_, msg = _summary_verdict(
white_ok=20, white_total=20, black_ok=3, black_blocked=12, black_total=15
def test_blocked_zone_high_confidence(self):
_, msg, note = _summary_verdict(
white_ok=20, white_total=20,
black_ok=3, black_blocked=12, black_total=15,
black_high_conf=10,
)
assert "are in" in msg.lower()
assert "blocked zone" in msg.lower()
assert "high confidence" in msg.lower()
def test_blocked_zone_medium_confidence(self):
_, msg, _ = _summary_verdict(
white_ok=20, white_total=20,
black_ok=3, black_blocked=12, black_total=15,
black_high_conf=2,
)
assert "blocked zone" in msg.lower()
assert "medium confidence" in msg.lower()
def test_partial_blocks_when_some_blacklist_loads(self):
_, msg = _summary_verdict(
white_ok=20, white_total=20, black_ok=10, black_blocked=5, black_total=15
_, msg, _ = _summary_verdict(
white_ok=20, white_total=20,
black_ok=10, black_blocked=5, black_total=15,
)
assert "partial" in msg.lower()
def test_partial_blocks_at_threshold_boundary(self):
# 11/15 = ~73% — should already be over the 70% threshold.
_, msg = _summary_verdict(
white_ok=20, white_total=20, black_ok=3, black_blocked=11, black_total=15
_, msg, _ = _summary_verdict(
white_ok=20, white_total=20,
black_ok=3, black_blocked=11, black_total=15,
black_high_conf=8,
)
assert "are in" in msg.lower()
assert "blocked zone" in msg.lower()
def test_whitelist_check_takes_priority(self):
_, msg = _summary_verdict(
white_ok=2, white_total=20, black_ok=0, black_blocked=15, black_total=15
_, msg, _ = _summary_verdict(
white_ok=2, white_total=20,
black_ok=0, black_blocked=15, black_total=15,
black_high_conf=15,
)
assert "degraded" in msg.lower()
assert "inconclusive" in msg.lower()