mirror of
https://github.com/MayersScott/rkn-block-checker.git
synced 2026-07-09 17:29:01 +00:00
feat: privacy-friendly default User-Agent, --identify for honest UA
This commit is contained in:
parent
492ce5ea41
commit
7628ca7d4a
3 changed files with 164 additions and 10 deletions
|
|
@ -27,7 +27,12 @@ def get_self_info(timeout: float = 5.0) -> dict:
|
|||
return {}
|
||||
|
||||
|
||||
def check_url(name: str, url: str, timeout: float = 5.0) -> CheckResult:
|
||||
def check_url(
|
||||
name: str,
|
||||
url: str,
|
||||
timeout: float = 5.0,
|
||||
identify: bool = False,
|
||||
) -> CheckResult:
|
||||
host = urlparse(url).hostname or url
|
||||
res = CheckResult(name=name, url=url)
|
||||
|
||||
|
|
@ -118,7 +123,7 @@ def check_url(name: str, url: str, timeout: float = 5.0) -> CheckResult:
|
|||
res.notes.append(f"TLS error: {res.tls_error}")
|
||||
return res
|
||||
|
||||
probe = http_mod.fetch(url, timeout=timeout)
|
||||
probe = http_mod.fetch(url, timeout=timeout, identify=identify)
|
||||
res.status_code = probe.status_code
|
||||
res.plt_ms = probe.elapsed_ms
|
||||
res.http_error = probe.error
|
||||
|
|
@ -155,6 +160,7 @@ def iter_check_urls(
|
|||
urls: dict[str, str],
|
||||
max_workers: int = DEFAULT_WORKERS,
|
||||
timeout: float = 5.0,
|
||||
identify: bool = False,
|
||||
) -> Iterator[CheckResult]:
|
||||
"""Yield CheckResult objects as soon as each probe finishes.
|
||||
|
||||
|
|
@ -168,7 +174,7 @@ def iter_check_urls(
|
|||
with ThreadPoolExecutor(max_workers=max_workers) as pool:
|
||||
futures_map: dict = {}
|
||||
for name, url in urls.items():
|
||||
fut = pool.submit(check_url, name, url, timeout)
|
||||
fut = pool.submit(check_url, name, url, timeout, identify)
|
||||
futures_map[fut] = name
|
||||
for fut in as_completed(futures_map):
|
||||
try:
|
||||
|
|
@ -186,6 +192,7 @@ def check_urls_parallel(
|
|||
urls: dict[str, str],
|
||||
max_workers: int = DEFAULT_WORKERS,
|
||||
timeout: float = 5.0,
|
||||
identify: bool = False,
|
||||
) -> list[CheckResult]:
|
||||
"""Run all probes in parallel and return results in the original input order.
|
||||
|
||||
|
|
@ -196,6 +203,8 @@ def check_urls_parallel(
|
|||
name_order = list(urls.keys())
|
||||
by_name = {
|
||||
r.name: r
|
||||
for r in iter_check_urls(urls, max_workers=max_workers, timeout=timeout)
|
||||
for r in iter_check_urls(
|
||||
urls, max_workers=max_workers, timeout=timeout, identify=identify,
|
||||
)
|
||||
}
|
||||
return [by_name[name] for name in name_order if name in by_name]
|
||||
|
|
|
|||
|
|
@ -6,14 +6,52 @@ from typing import Optional
|
|||
|
||||
import requests
|
||||
|
||||
from . import __version__
|
||||
from .targets import STUB_MARKERS
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_TIMEOUT = 5.0
|
||||
DEFAULT_USER_AGENT = "Mozilla/5.0 (RKN-Checker)"
|
||||
BODY_SNIPPET_LEN = 2000
|
||||
|
||||
# By default we send headers indistinguishable from a typical Chrome browser.
|
||||
# This is a privacy choice for the *user*, not an attempt to hide from the
|
||||
# servers being probed: an honest "RKN-Checker/<ver>" UA used to be the
|
||||
# default, but it leaves a unique fingerprint in any logs along the path —
|
||||
# including logs at VPN providers that, in some jurisdictions, hand data to
|
||||
# regulators. A generic UA blends the probe in with normal traffic and
|
||||
# minimizes the risk for users diagnosing their own connection. Operators
|
||||
# who *want* to be seen as diagnostic tooling (e.g. probing infrastructure
|
||||
# they own) can opt in via the --identify CLI flag, which switches to a
|
||||
# self-identifying UA.
|
||||
GENERIC_USER_AGENT = (
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||||
"Chrome/147.0.0.0 Safari/537.36"
|
||||
)
|
||||
HONEST_USER_AGENT = (
|
||||
f"rkn-block-checker/{__version__} "
|
||||
"(+https://github.com/MayersScott/rkn-block-checker)"
|
||||
)
|
||||
|
||||
# Headers Chrome actually sends. Without these, requests still stand out
|
||||
# even with the right UA — many sites and middleboxes fingerprint on the
|
||||
# header *set*, not just User-Agent.
|
||||
GENERIC_HEADERS: dict[str, str] = {
|
||||
"User-Agent": GENERIC_USER_AGENT,
|
||||
"Accept": (
|
||||
"text/html,application/xhtml+xml,application/xml;q=0.9,"
|
||||
"image/avif,image/webp,image/apng,*/*;q=0.8"
|
||||
),
|
||||
"Accept-Language": "en-US,en;q=0.9",
|
||||
"Accept-Encoding": "gzip, deflate, br",
|
||||
"Sec-Fetch-Dest": "document",
|
||||
"Sec-Fetch-Mode": "navigate",
|
||||
"Sec-Fetch-Site": "none",
|
||||
"Sec-Fetch-User": "?1",
|
||||
"Upgrade-Insecure-Requests": "1",
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class HttpProbe:
|
||||
|
|
@ -24,13 +62,33 @@ class HttpProbe:
|
|||
timed_out: bool = False
|
||||
|
||||
|
||||
def fetch(url: str, timeout: float = DEFAULT_TIMEOUT) -> HttpProbe:
|
||||
def build_headers(identify: bool = False) -> dict[str, str]:
|
||||
"""Return the header set to use for a probe.
|
||||
|
||||
`identify=False` (default): generic Chrome-like headers — minimizes the
|
||||
fingerprint left in logs. This is what end users running the tool to
|
||||
check their own connection should send.
|
||||
|
||||
`identify=True`: honest, self-identifying UA so that operators of probed
|
||||
infrastructure can distinguish this tool from anonymous traffic. Use
|
||||
this when you control or have permission to probe the target.
|
||||
"""
|
||||
if identify:
|
||||
return {**GENERIC_HEADERS, "User-Agent": HONEST_USER_AGENT}
|
||||
return dict(GENERIC_HEADERS)
|
||||
|
||||
|
||||
def fetch(
|
||||
url: str,
|
||||
timeout: float = DEFAULT_TIMEOUT,
|
||||
identify: bool = False,
|
||||
) -> HttpProbe:
|
||||
try:
|
||||
r = requests.get(
|
||||
url,
|
||||
timeout=timeout,
|
||||
allow_redirects=True,
|
||||
headers={"User-Agent": DEFAULT_USER_AGENT},
|
||||
headers=build_headers(identify=identify),
|
||||
)
|
||||
return HttpProbe(
|
||||
status_code=r.status_code,
|
||||
|
|
|
|||
|
|
@ -1,6 +1,12 @@
|
|||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from rkn_checker.http import fetch, looks_like_stub
|
||||
from rkn_checker.http import (
|
||||
GENERIC_USER_AGENT,
|
||||
HONEST_USER_AGENT,
|
||||
build_headers,
|
||||
fetch,
|
||||
looks_like_stub,
|
||||
)
|
||||
|
||||
|
||||
class TestFetch:
|
||||
|
|
@ -15,13 +21,15 @@ class TestFetch:
|
|||
assert probe.status_code == 200
|
||||
assert probe.error is None
|
||||
|
||||
@patch("rkn_checker.http.requests.get", side_effect=__import__("requests").exceptions.Timeout("t"))
|
||||
@patch("rkn_checker.http.requests.get",
|
||||
side_effect=__import__("requests").exceptions.Timeout("t"))
|
||||
def test_timeout_returns_timed_out_probe(self, mock_get):
|
||||
probe = fetch("https://example.com")
|
||||
assert probe.timed_out is True
|
||||
assert probe.error == "timeout"
|
||||
|
||||
@patch("rkn_checker.http.requests.get", side_effect=__import__("requests").exceptions.RequestException("e"))
|
||||
@patch("rkn_checker.http.requests.get",
|
||||
side_effect=__import__("requests").exceptions.RequestException("e"))
|
||||
def test_generic_error_returns_error_probe(self, mock_get):
|
||||
probe = fetch("https://example.com")
|
||||
assert probe.error is not None
|
||||
|
|
@ -39,6 +47,85 @@ class TestFetch:
|
|||
assert mock_get.call_args[1]["timeout"] == 2.0
|
||||
|
||||
|
||||
class TestUserAgentDefault:
|
||||
"""The default UA must be generic to avoid leaving a unique fingerprint
|
||||
in network logs along the path. This is the user-protection behaviour;
|
||||
any regression here would re-expose users to log-correlation risk."""
|
||||
|
||||
@patch("rkn_checker.http.requests.get")
|
||||
def test_default_ua_is_generic_chrome(self, mock_get):
|
||||
resp = MagicMock()
|
||||
resp.status_code = 200
|
||||
resp.elapsed.total_seconds.return_value = 0.05
|
||||
resp.text = "ok"
|
||||
mock_get.return_value = resp
|
||||
|
||||
fetch("https://example.com")
|
||||
|
||||
sent_headers = mock_get.call_args[1]["headers"]
|
||||
assert sent_headers["User-Agent"] == GENERIC_USER_AGENT
|
||||
# Sanity: it shouldn't include the project name in any header by default.
|
||||
assert "rkn" not in sent_headers["User-Agent"].lower()
|
||||
|
||||
@patch("rkn_checker.http.requests.get")
|
||||
def test_default_sends_chrome_like_header_set(self, mock_get):
|
||||
# A correct UA alone isn't enough — Chrome sends a specific *set* of
|
||||
# headers; missing ones (Sec-Fetch-*, Accept-Language, etc) make the
|
||||
# request stand out even if UA looks right.
|
||||
resp = MagicMock()
|
||||
resp.status_code = 200
|
||||
resp.elapsed.total_seconds.return_value = 0.05
|
||||
resp.text = "ok"
|
||||
mock_get.return_value = resp
|
||||
|
||||
fetch("https://example.com")
|
||||
|
||||
sent_headers = mock_get.call_args[1]["headers"]
|
||||
for h in ("Accept", "Accept-Language", "Accept-Encoding",
|
||||
"Sec-Fetch-Dest", "Sec-Fetch-Mode", "Sec-Fetch-Site",
|
||||
"Upgrade-Insecure-Requests"):
|
||||
assert h in sent_headers, f"missing browser-like header: {h}"
|
||||
|
||||
|
||||
class TestUserAgentIdentify:
|
||||
"""When --identify is on, the request *should* be self-identifying so
|
||||
operators of probed infrastructure can distinguish this tool from
|
||||
anonymous scraping or attacks."""
|
||||
|
||||
@patch("rkn_checker.http.requests.get")
|
||||
def test_identify_flag_uses_honest_ua(self, mock_get):
|
||||
resp = MagicMock()
|
||||
resp.status_code = 200
|
||||
resp.elapsed.total_seconds.return_value = 0.05
|
||||
resp.text = "ok"
|
||||
mock_get.return_value = resp
|
||||
|
||||
fetch("https://example.com", identify=True)
|
||||
|
||||
sent_headers = mock_get.call_args[1]["headers"]
|
||||
assert sent_headers["User-Agent"] == HONEST_USER_AGENT
|
||||
assert "rkn-block-checker" in sent_headers["User-Agent"]
|
||||
|
||||
def test_build_headers_default_does_not_identify(self):
|
||||
h = build_headers(identify=False)
|
||||
assert "rkn" not in h["User-Agent"].lower()
|
||||
|
||||
def test_build_headers_identify_keeps_other_headers(self):
|
||||
# Switching to identify mode must not strip the rest of the
|
||||
# browser-like header set — only swap UA.
|
||||
default = build_headers(identify=False)
|
||||
identifying = build_headers(identify=True)
|
||||
assert set(default.keys()) == set(identifying.keys())
|
||||
assert identifying["User-Agent"] != default["User-Agent"]
|
||||
|
||||
def test_build_headers_returns_independent_copies(self):
|
||||
# Mutating the returned dict must not affect future callers.
|
||||
a = build_headers()
|
||||
a["X-Test"] = "1"
|
||||
b = build_headers()
|
||||
assert "X-Test" not in b
|
||||
|
||||
|
||||
class TestLooksLikeStubNegative:
|
||||
def test_generic_blocked_by_does_not_match(self):
|
||||
assert looks_like_stub("this resource is blocked by your provider") is False
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue