Something went wrong. Try again.
💀 A small mitmproxy interceptor blocking your favourite telemetry providers.
mitmproxy
Something went wrong. Try again.
simple-proxy addon.py
4.9 kB · 148 lines
Python
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149import sys
from mitmproxy import http, tcp
# Line-buffer stdout so logs survive a crash even when piped to a file.sys.stdout.reconfigure(line_buffering=True)
# Hosts dedicated to telemetry ingest. Suffix match, any method. Safe to# blanket-mock - none of these serve a marketing site or dashboard UI.INGEST_HOSTS = ( # Datadog browser RUM (regional) "browser-intake-datadoghq.com", "browser-intake-datadoghq.eu", "browser-intake-datadoghq.us", "browser-intake-us3-datadoghq.com", "browser-intake-us5-datadoghq.com", "browser-intake-ap1-datadoghq.com",
# Google Analytics / GTM / DoubleClick (use GET pixels - must be here) "google-analytics.com", "analytics.google.com", "googletagmanager.com", "stats.g.doubleclick.net", "app-measurement.com",
"firebaselogging.googleapis.com", "firebase-settings.crashlytics.com",
# Product analytics - dedicated API subdomains only, never the apex. "api.mixpanel.com", "api-eu.mixpanel.com", "api.amplitude.com", "api2.amplitude.com", "api.eu.amplitude.com", "api.segment.io",
"notify.bugsnag.com", "sessions.bugsnag.com", "api.rollbar.com",
"collector.newrelic.com", "nr-data.net", "dc.services.visualstudio.com", "applicationinsights.azure.com", "in.appcenter.ms",
"lr-ingest.io", "vitals.vercel-insights.com",
"mc.yandex.ru", "mc.yandex.com", "telemetr.ee",)
# Vendor apex domains that ALSO host marketing sites or customer dashboards.# Mock only non-GET requests so browsing the site (and loading static assets)# still works, while POST/PUT/PATCH event uploads get absorbed.VENDOR_SUFFIXES_NON_GET = ( "datadoghq.com", "datadoghq.eu", "datadoghq.us", "mixpanel.com", "amplitude.com", "newrelic.com", "appsflyer.com", "adjust.com", "singular.net", "kochava.com", "pendo.io", "fullstory.com", "hotjar.com", "hotjar.io", "smartlook.com", "inspectlet.com", "mouseflow.com", "logrocket.io", "crashlytics.com", "heapanalytics.com", "telemetree.io", "splunkcloud.com", "sumologic.com", "loggly.com", "vercel-analytics.com",)
# Hosts that mix event ingest with feature flags / remote config. Mock only# the event paths so the app's flag/config calls still resolve.## Deliberately NOT mocked:# - Sentry (sentry.io, *.ingest.sentry.io) - crash reports stay on# - Cloudflare Insights - kept on by request# - Branch (api.branch.io) - also resolves deep links# - Firebase Remote Config / Installations - apps depend on these to bootMOCK_PATH_PREFIXES = ( ("i.posthog.com", ("/e/", "/batch/", "/capture/", "/s/")), ("app.posthog.com", ("/e/", "/batch/", "/capture/", "/s/")), ("events.statsigapi.net", ("/",)), ("events.launchdarkly.com", ("/",)), ("mobile.launchdarkly.com", ("/mobile/events",)),)
def _host_matches(host: str, suffix: str) -> bool: return host == suffix or host.endswith("." + suffix)
def _mock_reason(method: str, host: str, path: str) -> str | None: host = host.lower() path_only = path.split("?", 1)[0] if any(_host_matches(host, h) for h in INGEST_HOSTS): return "ingest" if method != "GET" and any(_host_matches(host, h) for h in VENDOR_SUFFIXES_NON_GET): return "vendor-post" for h, prefixes in MOCK_PATH_PREFIXES: if _host_matches(host, h) and any(path_only.startswith(p) for p in prefixes): return "path" return None
def _safe_url(req: http.Request) -> str: # Strip query string (OAuth tokens, session IDs, signed URLs) and any # control / DEL chars (defends terminal against ANSI-escape injection). url = req.pretty_url.split("?", 1)[0] return "".join(c if c == "\t" or (" " <= c < "\x7f") else "?" for c in url)
def request(flow: http.HTTPFlow) -> None: safe = _safe_url(flow.request) print(f"HTTP {flow.request.method} {safe}") reason = _mock_reason( flow.request.method, flow.request.pretty_host, flow.request.path ) if reason: # No X-Mocked-By header - paranoid SDKs could detect the marker # and change behavior (retry, refuse to function, surface a warning). flow.response = http.Response.make( 202, b'{"status":"ok"}', {"Content-Type": "application/json"}, ) print(f" ⊘ mocked [{reason}] → {safe}")
def response(flow: http.HTTPFlow) -> None: if flow.response is None: return print(f" → {flow.response.status_code} {_safe_url(flow.request)}")
def tcp_start(flow: tcp.TCPFlow) -> None: addr = flow.server_conn.address if addr: print(f"TCP connect → {addr[0]}:{addr[1]}")
def tcp_end(flow: tcp.TCPFlow) -> None: addr = flow.server_conn.address if addr: print(f"TCP close → {addr[0]}:{addr[1]}")