Spoof-when-possible browser personas, Tier-2 telemetry kill + artifact wipe, UI polish
Browser: identity persona dropdown (blend Windows-Chrome / Firefox / mac-Safari / hardened / custom) + cookie policy dropdown. Blend personas relax RFP/FPI/strict-TP so the machine looks normal while still hiding behind the chain. Persona TZ applied via env var; UA / language / locale / screen via user.js. WebRTC still forced off (it leaks real IP). Privacy: Telemetry kill toggle (DiagTrack, Activity History, ad ID, Cortana, scheduled tasks) with full snapshot/restore. One-button forensic artifact wipe (TEMP, Recent, Jump Lists, Prefetch, MRU, clipboard). UI: neon palette, glow card borders, gradient header banner. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Indiana Holmes
@@ -35,6 +35,13 @@ from .firewall import (
|
||||
is_engaged as fw_is_engaged,
|
||||
request_admin_relaunch,
|
||||
)
|
||||
from .artifact_wipe import wipe_artifacts
|
||||
from .browser_identity import (
|
||||
COOKIE_LABELS,
|
||||
COOKIE_MODES,
|
||||
PERSONA_LABELS,
|
||||
PERSONAS,
|
||||
)
|
||||
from .dns_leak import check_dns_leak_hint, flush_dns_cache
|
||||
from .leak_audit import AuditReport, run_audit_sync
|
||||
from .browser_launcher import (
|
||||
@@ -80,22 +87,26 @@ class UILogHandler(logging.Handler):
|
||||
|
||||
|
||||
# ── palette ──────────────────────────────────────────────────────────────────
|
||||
BG = "#0d0d1a"
|
||||
CARD = "#12122a"
|
||||
PANEL = "#1a1a3e"
|
||||
ACCENT = "#1e3a8a"
|
||||
ACCENT2 = "#2563eb"
|
||||
GREEN = "#00e676"
|
||||
RED = "#ff1744"
|
||||
YELLOW = "#ffab00"
|
||||
ORANGE = "#ff6d00"
|
||||
PURPLE = "#bb86fc"
|
||||
BLUE = "#448aff"
|
||||
CYAN = "#00bcd4"
|
||||
DIM = "#4a5568"
|
||||
TEXT = "#e2e8f0"
|
||||
TEXT2 = "#94a3b8"
|
||||
FONT = "Segoe UI"
|
||||
# Neon "blueprint terminal" theme — same overall vibe, deeper black, sharper
|
||||
# cyan/magenta accent, glow borders rendered via 1px border_color on cards.
|
||||
BG = "#070713"
|
||||
CARD = "#0f0f24"
|
||||
PANEL = "#171738"
|
||||
ACCENT = "#3052ff" # primary action / outlines
|
||||
ACCENT2 = "#00e5ff" # secondary highlight / glow
|
||||
GREEN = "#00ff9c"
|
||||
RED = "#ff2a6d"
|
||||
YELLOW = "#ffd000"
|
||||
ORANGE = "#ff8a00"
|
||||
PURPLE = "#c77dff"
|
||||
BLUE = "#5b8bff"
|
||||
CYAN = "#00f5ff"
|
||||
MAGENTA = "#ff00d6"
|
||||
DIM = "#2a2f4a"
|
||||
TEXT = "#eaf2ff"
|
||||
TEXT2 = "#7a8aab"
|
||||
GLOW = "#0a87ff"
|
||||
FONT = "Segoe UI"
|
||||
|
||||
|
||||
def _ts() -> str:
|
||||
@@ -108,10 +119,56 @@ def main() -> None:
|
||||
|
||||
root = ctk.CTk()
|
||||
root.title("Proxy God v2")
|
||||
root.geometry("1080x740")
|
||||
root.geometry("1080x760")
|
||||
root.minsize(900, 620)
|
||||
root.configure(fg_color=BG)
|
||||
|
||||
# ── 3D-ish neon header banner (drawn on a Canvas for a true gradient) ────
|
||||
def _draw_header_banner(parent: Any) -> Any:
|
||||
import tkinter as tk
|
||||
canvas = tk.Canvas(
|
||||
parent, height=44, highlightthickness=0, bd=0, bg=BG,
|
||||
)
|
||||
canvas.pack(fill="x", side="top")
|
||||
|
||||
def _hex(rgb: tuple[int, int, int]) -> str:
|
||||
return "#" + "".join(f"{v:02x}" for v in rgb)
|
||||
|
||||
def _lerp(a: tuple[int, int, int], b: tuple[int, int, int], t: float) -> tuple[int, int, int]:
|
||||
return tuple(int(a[i] + (b[i] - a[i]) * t) for i in range(3)) # type: ignore[return-value]
|
||||
|
||||
def _repaint(_event: Any = None) -> None:
|
||||
canvas.delete("grad")
|
||||
w = canvas.winfo_width() or 1080
|
||||
h = 44
|
||||
left = (8, 32, 64) # near-black blue
|
||||
mid = (48, 82, 255) # ACCENT
|
||||
right = (0, 229, 255) # ACCENT2 cyan
|
||||
for x in range(0, w, 2):
|
||||
t = x / max(1, w - 1)
|
||||
c = _lerp(left, mid, t * 2) if t < 0.5 else _lerp(mid, right, (t - 0.5) * 2)
|
||||
canvas.create_line(x, 0, x, h, fill=_hex(c), tags="grad")
|
||||
# 1px neon underline
|
||||
canvas.create_line(0, h - 1, w, h - 1, fill=CYAN, width=1, tags="grad")
|
||||
# Title text + small subtitle
|
||||
canvas.create_text(
|
||||
18, h // 2 - 2, anchor="w",
|
||||
text="P R O X Y G O D",
|
||||
fill="#ffffff", font=(FONT, 16, "bold"),
|
||||
tags="grad",
|
||||
)
|
||||
canvas.create_text(
|
||||
190, h // 2 - 1, anchor="w",
|
||||
text="// multi-hop · anti-fingerprint · paranoid mode",
|
||||
fill="#bfeaff", font=(FONT, 9, "italic"),
|
||||
tags="grad",
|
||||
)
|
||||
|
||||
canvas.bind("<Configure>", _repaint)
|
||||
return canvas
|
||||
|
||||
_draw_header_banner(root)
|
||||
|
||||
ui_q: queue.Queue[dict[str, Any]] = queue.Queue()
|
||||
svc = ChainService(notify=lambda m: ui_q.put(m))
|
||||
s = load_settings()
|
||||
@@ -847,7 +904,10 @@ def main() -> None:
|
||||
ctk.CTkLabel(
|
||||
br_scroll, text=subtitle, font=(FONT, 10), text_color=TEXT2, wraplength=900, justify="left"
|
||||
).pack(anchor="w", padx=8, pady=(0, 4))
|
||||
f = ctk.CTkFrame(br_scroll, fg_color=CARD, corner_radius=8)
|
||||
f = ctk.CTkFrame(
|
||||
br_scroll, fg_color=CARD, corner_radius=10,
|
||||
border_width=1, border_color=GLOW,
|
||||
)
|
||||
f.pack(fill="x", padx=8, pady=(4, 8))
|
||||
return f
|
||||
|
||||
@@ -895,6 +955,78 @@ def main() -> None:
|
||||
"Browser controls",
|
||||
"These settings are applied each time you launch from this tab.",
|
||||
)
|
||||
persona_card = _section_inline(
|
||||
br_scroll,
|
||||
"Identity persona (spoof when possible, block when not)",
|
||||
"Blend-in personas mimic the most common Windows + Chrome setup so the "
|
||||
"browser looks normal. Hardened mode is the classic 'block everything' "
|
||||
"profile but stands out as anti-fingerprint.",
|
||||
) if "_section_inline" in dir() else None
|
||||
if persona_card is None:
|
||||
persona_card = ctk.CTkFrame(
|
||||
br_scroll, fg_color=CARD, corner_radius=10,
|
||||
border_width=1, border_color=GLOW,
|
||||
)
|
||||
persona_card.pack(fill="x", padx=8, pady=(2, 8))
|
||||
ctk.CTkLabel(
|
||||
persona_card,
|
||||
text="Identity persona (spoof when possible, block when not)",
|
||||
font=(FONT, 12, "bold"),
|
||||
text_color=ORANGE,
|
||||
).pack(anchor="w", padx=12, pady=(10, 0))
|
||||
ctk.CTkLabel(
|
||||
persona_card,
|
||||
text=(
|
||||
"Blend-in personas mimic the most common Windows + Chrome setup "
|
||||
"so the browser looks normal. Hardened = full anti-fingerprint."
|
||||
),
|
||||
font=(FONT, 9), text_color=TEXT2, justify="left", wraplength=820,
|
||||
).pack(anchor="w", padx=12, pady=(0, 6))
|
||||
|
||||
persona_row = ctk.CTkFrame(persona_card, fg_color="transparent")
|
||||
persona_row.pack(fill="x", padx=12, pady=(0, 6))
|
||||
ctk.CTkLabel(persona_row, text="Persona:", font=(FONT, 11), text_color=TEXT,
|
||||
width=80, anchor="w").pack(side="left")
|
||||
persona_var = ctk.StringVar(
|
||||
value=PERSONA_LABELS.get(s.browser_persona, PERSONA_LABELS["blend_windows_chrome"])
|
||||
)
|
||||
persona_menu = ctk.CTkOptionMenu(
|
||||
persona_row,
|
||||
values=[PERSONA_LABELS[k] for k in PERSONAS],
|
||||
variable=persona_var,
|
||||
fg_color=BG, button_color=ACCENT2, button_hover_color=ACCENT,
|
||||
text_color=TEXT, font=(FONT, 11), width=500,
|
||||
)
|
||||
persona_menu.pack(side="left", padx=(4, 0))
|
||||
|
||||
cookie_row = ctk.CTkFrame(persona_card, fg_color="transparent")
|
||||
cookie_row.pack(fill="x", padx=12, pady=(0, 10))
|
||||
ctk.CTkLabel(cookie_row, text="Cookies:", font=(FONT, 11), text_color=TEXT,
|
||||
width=80, anchor="w").pack(side="left")
|
||||
cookie_var = ctk.StringVar(
|
||||
value=COOKIE_LABELS.get(s.browser_cookie_mode, COOKIE_LABELS["block_third_party"])
|
||||
)
|
||||
cookie_menu = ctk.CTkOptionMenu(
|
||||
cookie_row,
|
||||
values=[COOKIE_LABELS[k] for k in COOKIE_MODES],
|
||||
variable=cookie_var,
|
||||
fg_color=BG, button_color=ACCENT2, button_hover_color=ACCENT,
|
||||
text_color=TEXT, font=(FONT, 11), width=500,
|
||||
)
|
||||
cookie_menu.pack(side="left", padx=(4, 0))
|
||||
|
||||
def _persona_key() -> str:
|
||||
for k in PERSONAS:
|
||||
if PERSONA_LABELS[k] == persona_var.get():
|
||||
return k
|
||||
return "blend_windows_chrome"
|
||||
|
||||
def _cookie_key() -> str:
|
||||
for k in COOKIE_MODES:
|
||||
if COOKIE_LABELS[k] == cookie_var.get():
|
||||
return k
|
||||
return "block_third_party"
|
||||
|
||||
br_force_proxy_var = ctk.BooleanVar(value=s.browser_force_proxy)
|
||||
br_disable_webrtc_var = ctk.BooleanVar(value=s.browser_disable_webrtc)
|
||||
br_rfp_var = ctk.BooleanVar(value=s.browser_resist_fingerprinting)
|
||||
@@ -984,6 +1116,8 @@ def main() -> None:
|
||||
first_party_isolation=bool(br_fpi_var.get()),
|
||||
strict_tracking_protection=bool(br_strict_tp_var.get()),
|
||||
timezone_utc=bool(br_tz_utc_var.get()),
|
||||
persona=_persona_key(),
|
||||
cookie_mode=_cookie_key(),
|
||||
)
|
||||
|
||||
def _refresh_browser_status() -> None:
|
||||
@@ -1043,7 +1177,10 @@ def main() -> None:
|
||||
ctk.CTkLabel(priv_scroll, text=subtitle, font=(FONT, 10),
|
||||
text_color=TEXT2, wraplength=900, justify="left").pack(
|
||||
anchor="w", padx=8, pady=(0, 4))
|
||||
f = ctk.CTkFrame(priv_scroll, fg_color=CARD, corner_radius=8)
|
||||
f = ctk.CTkFrame(
|
||||
priv_scroll, fg_color=CARD, corner_radius=10,
|
||||
border_width=1, border_color=GLOW,
|
||||
)
|
||||
f.pack(fill="x", padx=8, pady=(4, 8))
|
||||
return f
|
||||
|
||||
@@ -1090,6 +1227,7 @@ def main() -> None:
|
||||
ipv6_var = ctk.BooleanVar(value=s.disable_ipv6_while_active)
|
||||
webrtc_var = ctk.BooleanVar(value=s.harden_webrtc_enabled)
|
||||
lan_var = ctk.BooleanVar(value=s.lan_lockdown_enabled)
|
||||
telemetry_var = ctk.BooleanVar(value=s.telemetry_kill_enabled)
|
||||
|
||||
def _priv_toggle(parent: Any, text: str, var: ctk.BooleanVar, tip: str = "") -> ctk.CTkCheckBox:
|
||||
cb = ctk.CTkCheckBox(
|
||||
@@ -1132,6 +1270,13 @@ def main() -> None:
|
||||
"Stops your machine from advertising its hostname on the local network "
|
||||
"(admin required, reversible on stop).",
|
||||
)
|
||||
_priv_toggle(
|
||||
toggles_card,
|
||||
"Telemetry kill: DiagTrack, Activity History, Cortana web, ad ID",
|
||||
telemetry_var,
|
||||
"Stops Windows telemetry pipelines while chain is up. Reversible on stop "
|
||||
"(snapshot of original state taken).",
|
||||
)
|
||||
|
||||
fp_card = _priv_section(
|
||||
"Device fingerprint",
|
||||
@@ -1177,6 +1322,35 @@ def main() -> None:
|
||||
_btn(dns_btn_row, "Flush DNS now", _manual_dns_flush, w=110, h=28,
|
||||
fg_color=DIM).pack(side="left")
|
||||
|
||||
wipe_card = _priv_section(
|
||||
"Forensic artifact wipe",
|
||||
"One-button purge of common Windows breadcrumb trails. Irreversible.",
|
||||
)
|
||||
wipe_result_lbl = ctk.CTkLabel(
|
||||
wipe_card, text="Click 'Wipe now' to purge %TEMP%, Recent, Jump Lists, "
|
||||
"Prefetch (Admin), MRU lists, and the clipboard.",
|
||||
font=(FONT, 10), text_color=TEXT2, wraplength=880, justify="left",
|
||||
)
|
||||
wipe_result_lbl.pack(anchor="w", padx=12, pady=8)
|
||||
|
||||
def _do_wipe() -> None:
|
||||
wipe_result_lbl.configure(text="Wiping…", text_color=YELLOW)
|
||||
|
||||
def work() -> None:
|
||||
try:
|
||||
rep = wipe_artifacts(include_prefetch=True)
|
||||
except Exception as e:
|
||||
root.after(0, lambda: wipe_result_lbl.configure(
|
||||
text=f"Wipe failed: {e}", text_color=RED))
|
||||
return
|
||||
root.after(0, lambda: wipe_result_lbl.configure(
|
||||
text=rep.summary(), text_color=GREEN if not rep.errors else YELLOW))
|
||||
|
||||
threading.Thread(target=work, daemon=True).start()
|
||||
|
||||
_btn(wipe_card, "Wipe now", _do_wipe, w=110, h=28,
|
||||
fg_color=RED, hover_color="#8a1d1d").pack(anchor="w", padx=12, pady=(0, 10))
|
||||
|
||||
audit_card = _priv_section(
|
||||
"Leak audit (mission-critical)",
|
||||
"Probes every leak surface: IP, DNS, IPv6, WPAD, Group Policy, "
|
||||
@@ -1310,6 +1484,9 @@ def main() -> None:
|
||||
ipv6_var.set(False)
|
||||
webrtc_var.set(True)
|
||||
lan_var.set(False)
|
||||
telemetry_var.set(False)
|
||||
persona_var.set(PERSONA_LABELS["blend_windows_chrome"])
|
||||
cookie_var.set(COOKIE_LABELS["block_third_party"])
|
||||
br_force_proxy_var.set(True)
|
||||
br_disable_webrtc_var.set(True)
|
||||
br_rfp_var.set(True)
|
||||
@@ -1381,6 +1558,7 @@ def main() -> None:
|
||||
disable_ipv6_while_active=bool(ipv6_var.get()),
|
||||
harden_webrtc_enabled=bool(webrtc_var.get()),
|
||||
lan_lockdown_enabled=bool(lan_var.get()),
|
||||
telemetry_kill_enabled=bool(telemetry_var.get()),
|
||||
firefox_path=firefox_path_var.get().strip(),
|
||||
firefox_profile_dir=firefox_profile_var.get().strip(),
|
||||
browser_clear_on_close=bool(br_clear_var.get()),
|
||||
@@ -1395,6 +1573,8 @@ def main() -> None:
|
||||
browser_first_party_isolation=bool(br_fpi_var.get()),
|
||||
browser_strict_tracking_protection=bool(br_strict_tp_var.get()),
|
||||
browser_timezone_utc=bool(br_tz_utc_var.get()),
|
||||
browser_persona=_persona_key(),
|
||||
browser_cookie_mode=_cookie_key(),
|
||||
)
|
||||
if not (1 <= ns.local_port <= 65535):
|
||||
raise ValueError("Port must be 1-65535")
|
||||
|
||||
218
proxy_chain_manager/artifact_wipe.py
Normal file
218
proxy_chain_manager/artifact_wipe.py
Normal file
@@ -0,0 +1,218 @@
|
||||
"""One-button forensic artifact wipe.
|
||||
|
||||
Surfaces purged:
|
||||
|
||||
• %TEMP% — per-user temp
|
||||
• %SystemRoot%\\Prefetch\\*.pf (Admin only) — file-launch history
|
||||
• %APPDATA%\\Microsoft\\Windows\\Recent\\* — recent files
|
||||
• %APPDATA%\\Microsoft\\Windows\\Recent\\AutomaticDestinations\\* — Jump Lists
|
||||
• %APPDATA%\\Microsoft\\Windows\\Recent\\CustomDestinations\\* — Jump Lists
|
||||
• HKCU\\…\\Explorer\\RunMRU — Win+R history
|
||||
• HKCU\\…\\Explorer\\TypedPaths — Explorer typed paths
|
||||
• HKCU\\…\\Explorer\\WordWheelQuery — Start / Explorer search history
|
||||
• HKCU\\…\\Explorer\\RecentDocs — recent docs MRU
|
||||
• Clipboard — current contents
|
||||
|
||||
Counts are reported but specific filenames are never logged (this is the
|
||||
opposite of what we want to leak). All deletes use ``ignore_errors=True``
|
||||
because files in use by other apps are expected.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import ctypes
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import winreg
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from .firewall import is_admin
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
_REG_MRU_KEYS = (
|
||||
r"Software\Microsoft\Windows\CurrentVersion\Explorer\RunMRU",
|
||||
r"Software\Microsoft\Windows\CurrentVersion\Explorer\TypedPaths",
|
||||
r"Software\Microsoft\Windows\CurrentVersion\Explorer\WordWheelQuery",
|
||||
r"Software\Microsoft\Windows\CurrentVersion\Explorer\RecentDocs",
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class WipeReport:
|
||||
files_deleted: int = 0
|
||||
bytes_freed: int = 0
|
||||
folders_skipped: int = 0
|
||||
registry_keys_cleared: int = 0
|
||||
clipboard_cleared: bool = False
|
||||
errors: list[str] = field(default_factory=list)
|
||||
|
||||
def summary(self) -> str:
|
||||
mb = self.bytes_freed / (1024 * 1024)
|
||||
return (
|
||||
f"{self.files_deleted} files / {mb:.1f} MB freed, "
|
||||
f"{self.registry_keys_cleared} MRU entries cleared, "
|
||||
f"clipboard={'yes' if self.clipboard_cleared else 'no'}, "
|
||||
f"errors={len(self.errors)}"
|
||||
)
|
||||
|
||||
|
||||
def _walk_size(path: Path) -> int:
|
||||
total = 0
|
||||
for p in path.rglob("*"):
|
||||
try:
|
||||
if p.is_file():
|
||||
total += p.stat().st_size
|
||||
except OSError:
|
||||
pass
|
||||
return total
|
||||
|
||||
|
||||
def _purge_dir(path: Path, rep: WipeReport, keep_root: bool = True) -> None:
|
||||
"""Delete contents of `path`. Preserves the directory itself when keep_root."""
|
||||
if not path.exists():
|
||||
return
|
||||
try:
|
||||
size_before = _walk_size(path)
|
||||
except Exception:
|
||||
size_before = 0
|
||||
|
||||
count = 0
|
||||
for child in path.iterdir():
|
||||
try:
|
||||
if child.is_dir():
|
||||
shutil.rmtree(child, ignore_errors=True)
|
||||
else:
|
||||
child.unlink(missing_ok=True)
|
||||
count += 1
|
||||
except OSError:
|
||||
rep.folders_skipped += 1
|
||||
|
||||
if not keep_root:
|
||||
try:
|
||||
path.rmdir()
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
rep.files_deleted += count
|
||||
try:
|
||||
size_after = _walk_size(path)
|
||||
except Exception:
|
||||
size_after = 0
|
||||
rep.bytes_freed += max(0, size_before - size_after)
|
||||
|
||||
|
||||
def _clear_mru_key(path: str, rep: WipeReport) -> None:
|
||||
"""Remove every value under a Run/Typed/Search MRU key."""
|
||||
try:
|
||||
with winreg.OpenKey(
|
||||
winreg.HKEY_CURRENT_USER, path, 0, winreg.KEY_ALL_ACCESS
|
||||
) as key:
|
||||
count = 0
|
||||
try:
|
||||
while True:
|
||||
name, _, _ = winreg.EnumValue(key, 0)
|
||||
try:
|
||||
winreg.DeleteValue(key, name)
|
||||
count += 1
|
||||
except OSError:
|
||||
break
|
||||
except OSError:
|
||||
pass
|
||||
# Recurse into subkeys (e.g. RecentDocs/.png)
|
||||
sub_count = 0
|
||||
try:
|
||||
while True:
|
||||
sub_count += 1
|
||||
sub_name = winreg.EnumKey(key, 0)
|
||||
try:
|
||||
winreg.DeleteKey(key, sub_name)
|
||||
except OSError:
|
||||
break
|
||||
if sub_count > 200: # safety cap
|
||||
break
|
||||
except OSError:
|
||||
pass
|
||||
if count or sub_count:
|
||||
rep.registry_keys_cleared += 1
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _clear_clipboard(rep: WipeReport) -> None:
|
||||
try:
|
||||
user32 = ctypes.windll.user32 # type: ignore[attr-defined]
|
||||
if user32.OpenClipboard(None):
|
||||
try:
|
||||
user32.EmptyClipboard()
|
||||
rep.clipboard_cleared = True
|
||||
finally:
|
||||
user32.CloseClipboard()
|
||||
except Exception as e:
|
||||
rep.errors.append(f"clipboard: {e}")
|
||||
|
||||
|
||||
def wipe_artifacts(include_prefetch: bool = True) -> WipeReport:
|
||||
rep = WipeReport()
|
||||
appdata = Path(os.environ.get("APPDATA", "")) if os.environ.get("APPDATA") else None
|
||||
temp = Path(os.environ.get("TEMP", "")) if os.environ.get("TEMP") else None
|
||||
sysroot = Path(os.environ.get("SystemRoot", r"C:\Windows"))
|
||||
|
||||
# %TEMP%
|
||||
if temp and temp.exists():
|
||||
_purge_dir(temp, rep, keep_root=True)
|
||||
|
||||
# Prefetch (Admin)
|
||||
if include_prefetch and is_admin():
|
||||
pf = sysroot / "Prefetch"
|
||||
if pf.exists():
|
||||
count = 0
|
||||
for f in pf.glob("*.pf"):
|
||||
try:
|
||||
sz = f.stat().st_size
|
||||
f.unlink(missing_ok=True)
|
||||
count += 1
|
||||
rep.bytes_freed += sz
|
||||
except OSError as e:
|
||||
rep.errors.append(f"prefetch: {e}")
|
||||
rep.files_deleted += count
|
||||
|
||||
# Recent / Jump Lists
|
||||
if appdata:
|
||||
recent = appdata / "Microsoft" / "Windows" / "Recent"
|
||||
if recent.exists():
|
||||
_purge_dir(recent / "AutomaticDestinations", rep, keep_root=True)
|
||||
_purge_dir(recent / "CustomDestinations", rep, keep_root=True)
|
||||
_purge_dir(recent, rep, keep_root=True)
|
||||
|
||||
# MRU registry keys
|
||||
for path in _REG_MRU_KEYS:
|
||||
_clear_mru_key(path, rep)
|
||||
|
||||
# Clipboard
|
||||
_clear_clipboard(rep)
|
||||
|
||||
# Trigger Explorer's "Clear recent items" via Shell API (covers Win10/11
|
||||
# Quick Access pinned ↔ recent lists not covered by raw file delete).
|
||||
try:
|
||||
shell32 = ctypes.windll.shell32 # type: ignore[attr-defined]
|
||||
# SHCNE_ASSOCCHANGED tells Explorer to refresh its caches.
|
||||
shell32.SHChangeNotify(0x08000000, 0x0000, None, None)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Best-effort Defender quick-scan history flush — non-fatal.
|
||||
try:
|
||||
subprocess.run(
|
||||
["powershell", "-NoProfile", "-NonInteractive", "-Command",
|
||||
"Clear-RecycleBin -Force -ErrorAction SilentlyContinue"],
|
||||
capture_output=True, text=True, timeout=20,
|
||||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return rep
|
||||
141
proxy_chain_manager/browser_identity.py
Normal file
141
proxy_chain_manager/browser_identity.py
Normal file
@@ -0,0 +1,141 @@
|
||||
"""Browser identity personas + cookie policies.
|
||||
|
||||
The "lock everything down" approach (Firefox RFP + WebRTC off + cookies blocked)
|
||||
keeps you safe from fingerprinting but makes the machine *look weird* — sites
|
||||
flag you as suspicious and ban / captcha you. Worse: very few users use that
|
||||
config, so RFP-on is itself a fingerprint.
|
||||
|
||||
Two strategies are exposed:
|
||||
|
||||
• **Blend-in personas** — mimic the most common Windows-Chrome-US-en setup.
|
||||
JS / cookies / WebGL all work; everything looks normal; only WebRTC is
|
||||
blocked (because that leaks the real IP behind the proxy). The proxy
|
||||
chain itself does the actual hiding.
|
||||
|
||||
• **Hardened** — full RFP / blocked WebRTC / strict tracking / FPI. Stand-out
|
||||
but maximum block. Same behavior as the original profile.
|
||||
|
||||
Cookie modes are independent of persona and let the user pick per-session
|
||||
behavior from a dropdown.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
# Order matters: dropdown order in the UI.
|
||||
PERSONAS: list[str] = [
|
||||
"blend_windows_chrome",
|
||||
"blend_windows_firefox",
|
||||
"blend_mac_safari",
|
||||
"hardened",
|
||||
"custom",
|
||||
]
|
||||
|
||||
PERSONA_LABELS: dict[str, str] = {
|
||||
"blend_windows_chrome": "Blend in — Windows 10 + Chrome (US-en, most common)",
|
||||
"blend_windows_firefox": "Blend in — Windows 10 + Firefox (US-en)",
|
||||
"blend_mac_safari": "Blend in — macOS + Safari (US-en)",
|
||||
"hardened": "Hardened — block + RFP (stands out; max anti-fingerprint)",
|
||||
"custom": "Custom — use individual hardening toggles",
|
||||
}
|
||||
|
||||
COOKIE_MODES: list[str] = [
|
||||
"accept_all",
|
||||
"block_third_party",
|
||||
"block_third_party_trackers",
|
||||
"session_only",
|
||||
"block_all",
|
||||
]
|
||||
|
||||
COOKIE_LABELS: dict[str, str] = {
|
||||
"accept_all": "Accept all cookies (most sites work)",
|
||||
"block_third_party": "Block 3rd-party cookies (recommended)",
|
||||
"block_third_party_trackers": "Block 3rd-party trackers only (Firefox default-strict)",
|
||||
"session_only": "Session only — clear on close",
|
||||
"block_all": "Block ALL cookies (breaks logins)",
|
||||
}
|
||||
|
||||
# Firefox network.cookie.cookieBehavior values
|
||||
_COOKIE_BEHAVIOR = {
|
||||
"accept_all": 0,
|
||||
"block_third_party": 1,
|
||||
"block_third_party_trackers": 4,
|
||||
"session_only": 0,
|
||||
"block_all": 2,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Persona:
|
||||
"""A user-agent / locale / timezone / screen tuple that mimics a real,
|
||||
common configuration. All fields are deliberately popular values so the
|
||||
persona blends in.
|
||||
"""
|
||||
key: str
|
||||
user_agent: str
|
||||
accept_language: str
|
||||
timezone: str # IANA tz; used only when persona drives env TZ
|
||||
screen_w: int
|
||||
screen_h: int
|
||||
platform: str # navigator.platform
|
||||
locale: str # general.useragent.locale equivalent
|
||||
|
||||
|
||||
PERSONA_DATA: dict[str, Persona] = {
|
||||
"blend_windows_chrome": Persona(
|
||||
key="blend_windows_chrome",
|
||||
# Stable Chrome on Win10 x64 — by far the most common UA on the web.
|
||||
user_agent=(
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
|
||||
),
|
||||
accept_language="en-US,en;q=0.9",
|
||||
timezone="America/New_York",
|
||||
screen_w=1920,
|
||||
screen_h=1080,
|
||||
platform="Win32",
|
||||
locale="en-US",
|
||||
),
|
||||
"blend_windows_firefox": Persona(
|
||||
key="blend_windows_firefox",
|
||||
user_agent=(
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:128.0) "
|
||||
"Gecko/20100101 Firefox/128.0"
|
||||
),
|
||||
accept_language="en-US,en;q=0.5",
|
||||
timezone="America/New_York",
|
||||
screen_w=1920,
|
||||
screen_h=1080,
|
||||
platform="Win32",
|
||||
locale="en-US",
|
||||
),
|
||||
"blend_mac_safari": Persona(
|
||||
key="blend_mac_safari",
|
||||
user_agent=(
|
||||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
|
||||
"AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4.1 Safari/605.1.15"
|
||||
),
|
||||
accept_language="en-US,en;q=0.9",
|
||||
timezone="America/Los_Angeles",
|
||||
screen_w=1680,
|
||||
screen_h=1050,
|
||||
platform="MacIntel",
|
||||
locale="en-US",
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def cookie_behavior_value(mode: str) -> int:
|
||||
return _COOKIE_BEHAVIOR.get(mode, 1)
|
||||
|
||||
|
||||
def cookie_session_only(mode: str) -> bool:
|
||||
return mode == "session_only"
|
||||
|
||||
|
||||
def get_persona(key: str) -> Persona | None:
|
||||
return PERSONA_DATA.get(key)
|
||||
|
||||
|
||||
def is_blend_persona(key: str) -> bool:
|
||||
return key.startswith("blend_")
|
||||
@@ -1,13 +1,14 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from .browser_profile import FirefoxHardening, ensure_firefox_profile
|
||||
from .browser_profile import FirefoxHardening, ensure_firefox_profile, persona_env
|
||||
from .paths import app_data_dir
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
@@ -29,6 +30,8 @@ class BrowserConfig:
|
||||
first_party_isolation: bool = True
|
||||
strict_tracking_protection: bool = True
|
||||
timezone_utc: bool = True
|
||||
persona: str = "blend_windows_chrome"
|
||||
cookie_mode: str = "block_third_party"
|
||||
|
||||
|
||||
def default_firefox_path() -> str:
|
||||
@@ -80,6 +83,8 @@ class BrowserSession:
|
||||
strict_tracking_protection=cfg.strict_tracking_protection,
|
||||
clear_on_shutdown=cfg.clear_on_close,
|
||||
timezone_utc=cfg.timezone_utc,
|
||||
persona_key=cfg.persona,
|
||||
cookie_mode=cfg.cookie_mode,
|
||||
)
|
||||
if cfg.lock_managed_profile:
|
||||
ensure_firefox_profile(profile_dir, proxy_host, int(proxy_port), hard)
|
||||
@@ -89,6 +94,8 @@ class BrowserSession:
|
||||
args = [str(exe), "-no-remote", "-profile", str(profile_dir)]
|
||||
else:
|
||||
args = [str(exe)]
|
||||
env = os.environ.copy()
|
||||
env.update(persona_env(hard))
|
||||
try:
|
||||
self._proc = subprocess.Popen(
|
||||
args,
|
||||
@@ -96,6 +103,7 @@ class BrowserSession:
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
text=True,
|
||||
env=env,
|
||||
)
|
||||
# Smoke-check: catch immediate startup crashes and report clearly.
|
||||
time.sleep(0.4)
|
||||
|
||||
@@ -1,13 +1,24 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from .browser_identity import (
|
||||
Persona,
|
||||
cookie_behavior_value,
|
||||
cookie_session_only,
|
||||
get_persona,
|
||||
is_blend_persona,
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class FirefoxHardening:
|
||||
# Proxy & network always-on
|
||||
force_proxy: bool = True
|
||||
disable_webrtc: bool = True
|
||||
disable_webrtc: bool = True # always strongly recommended; leaks real IP
|
||||
|
||||
# Hardened block mode — overrides persona spoofing
|
||||
resist_fingerprinting: bool = True
|
||||
disable_telemetry: bool = True
|
||||
first_party_isolation: bool = True
|
||||
@@ -15,6 +26,46 @@ class FirefoxHardening:
|
||||
clear_on_shutdown: bool = True
|
||||
timezone_utc: bool = True
|
||||
|
||||
# Persona / cookie selectors (added in "spoof when possible" mode)
|
||||
persona_key: str = "custom" # see browser_identity.PERSONAS
|
||||
cookie_mode: str = "block_third_party" # see browser_identity.COOKIE_MODES
|
||||
|
||||
# Optional Firefox-managed user-agent override even outside of a persona
|
||||
user_agent_override: str = ""
|
||||
|
||||
# Filled in by ``apply_persona`` when persona_key starts with "blend_".
|
||||
persona: Persona | None = field(default=None, repr=False, compare=False)
|
||||
|
||||
|
||||
def apply_persona(hard: FirefoxHardening) -> FirefoxHardening:
|
||||
"""Mutate hardening flags consistent with the persona choice.
|
||||
|
||||
Blend-in personas relax the "stand out" toggles (RFP, FPI, strict TP) but
|
||||
keep the IP-leakers off (WebRTC, geolocation, etc.). Cookie mode is
|
||||
honored. The result is a profile that looks like a normal browser.
|
||||
"""
|
||||
if hard.persona_key == "hardened":
|
||||
hard.resist_fingerprinting = True
|
||||
hard.first_party_isolation = True
|
||||
hard.strict_tracking_protection = True
|
||||
hard.timezone_utc = True
|
||||
hard.persona = None
|
||||
return hard
|
||||
|
||||
if is_blend_persona(hard.persona_key):
|
||||
hard.resist_fingerprinting = False # RFP itself is a fingerprint
|
||||
hard.first_party_isolation = False # breaks logins on many sites
|
||||
hard.strict_tracking_protection = False # Firefox default ETP is enough
|
||||
hard.timezone_utc = False
|
||||
hard.persona = get_persona(hard.persona_key)
|
||||
if hard.persona and not hard.user_agent_override:
|
||||
hard.user_agent_override = hard.persona.user_agent
|
||||
return hard
|
||||
|
||||
# custom — honor the individual flags as-is.
|
||||
hard.persona = None
|
||||
return hard
|
||||
|
||||
|
||||
def _bool(v: bool) -> str:
|
||||
return "true" if v else "false"
|
||||
@@ -25,8 +76,10 @@ def build_user_js(
|
||||
proxy_port: int,
|
||||
hard: FirefoxHardening,
|
||||
) -> str:
|
||||
hard = apply_persona(hard)
|
||||
lines: list[str] = [
|
||||
"// Managed by Proxy God. Changes are overwritten on next launch.",
|
||||
# Always-on baseline (no UX cost)
|
||||
'user_pref("app.normandy.enabled", false);',
|
||||
'user_pref("app.shield.optoutstudies.enabled", false);',
|
||||
'user_pref("browser.newtabpage.activity-stream.feeds.telemetry", false);',
|
||||
@@ -45,14 +98,14 @@ def build_user_js(
|
||||
'user_pref("network.trr.mode", 5);',
|
||||
'user_pref("network.captive-portal-service.enabled", false);',
|
||||
'user_pref("geo.enabled", false);',
|
||||
'user_pref("media.peerconnection.enabled", false);',
|
||||
'user_pref("media.navigator.enabled", false);',
|
||||
'user_pref("dom.battery.enabled", false);',
|
||||
'user_pref("dom.gamepad.enabled", false);',
|
||||
'user_pref("dom.netinfo.enabled", false);',
|
||||
'user_pref("webgl.disabled", false);',
|
||||
# WebGL allowed by default (blocking it makes you stand out massively)
|
||||
'user_pref("webgl.enable-debug-renderer-info", false);',
|
||||
]
|
||||
|
||||
if hard.force_proxy:
|
||||
lines.extend(
|
||||
[
|
||||
@@ -71,7 +124,9 @@ def build_user_js(
|
||||
'user_pref("network.proxy.no_proxies_on", "");',
|
||||
]
|
||||
)
|
||||
|
||||
if hard.disable_webrtc:
|
||||
# Off regardless of persona — WebRTC leaks real IP through STUN.
|
||||
lines.extend(
|
||||
[
|
||||
'user_pref("media.peerconnection.enabled", false);',
|
||||
@@ -79,6 +134,7 @@ def build_user_js(
|
||||
'user_pref("media.peerconnection.ice.no_host", true);',
|
||||
]
|
||||
)
|
||||
|
||||
if hard.resist_fingerprinting:
|
||||
lines.extend(
|
||||
[
|
||||
@@ -88,10 +144,13 @@ def build_user_js(
|
||||
'user_pref("privacy.window.maxInnerHeight", 900);',
|
||||
]
|
||||
)
|
||||
|
||||
if hard.first_party_isolation:
|
||||
lines.append('user_pref("privacy.firstparty.isolate", true);')
|
||||
|
||||
if hard.disable_telemetry:
|
||||
lines.append('user_pref("browser.send_pings", false);')
|
||||
|
||||
if hard.strict_tracking_protection:
|
||||
lines.extend(
|
||||
[
|
||||
@@ -99,6 +158,38 @@ def build_user_js(
|
||||
'user_pref("privacy.trackingprotection.pbmode.enabled", true);',
|
||||
]
|
||||
)
|
||||
|
||||
# Persona spoofing (UA, accept-language, screen, platform)
|
||||
if hard.persona is not None:
|
||||
p = hard.persona
|
||||
lines.extend(
|
||||
[
|
||||
f'user_pref("general.useragent.override", "{p.user_agent}");',
|
||||
f'user_pref("intl.accept_languages", "{p.accept_language}");',
|
||||
f'user_pref("general.useragent.locale", "{p.locale}");',
|
||||
'user_pref("javascript.use_us_english_locale", true);',
|
||||
f'user_pref("privacy.window.maxInnerWidth", {int(p.screen_w)});',
|
||||
f'user_pref("privacy.window.maxInnerHeight", {int(p.screen_h)});',
|
||||
# Hint timezone — Firefox honors TZ env var at launch (set by
|
||||
# the launcher) but this pref nudges some site detection too.
|
||||
f'user_pref("intl.locale.requested", "{p.locale}");',
|
||||
]
|
||||
)
|
||||
elif hard.user_agent_override:
|
||||
lines.append(
|
||||
f'user_pref("general.useragent.override", "{hard.user_agent_override}");'
|
||||
)
|
||||
|
||||
# Cookie policy (persona-independent)
|
||||
behavior = cookie_behavior_value(hard.cookie_mode)
|
||||
lines.extend(
|
||||
[
|
||||
f'user_pref("network.cookie.cookieBehavior", {behavior});',
|
||||
# 0 = expire normally, 2 = current session only
|
||||
f'user_pref("network.cookie.lifetimePolicy", {2 if cookie_session_only(hard.cookie_mode) else 0});',
|
||||
]
|
||||
)
|
||||
|
||||
if hard.clear_on_shutdown:
|
||||
lines.extend(
|
||||
[
|
||||
@@ -111,12 +202,26 @@ def build_user_js(
|
||||
'user_pref("privacy.clearOnShutdown.sessions", true);',
|
||||
]
|
||||
)
|
||||
|
||||
if hard.timezone_utc:
|
||||
lines.append('user_pref("privacy.resistFingerprinting.reduceTimerPrecision", true);')
|
||||
|
||||
lines.append("")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def persona_env(hard: FirefoxHardening) -> dict[str, str]:
|
||||
"""Environment overrides applied when launching Firefox under a persona.
|
||||
|
||||
Only the timezone needs the OS-level env var (TZ) — Firefox reads it
|
||||
even when running under Windows.
|
||||
"""
|
||||
hard = apply_persona(hard)
|
||||
if hard.persona is None:
|
||||
return {}
|
||||
return {"TZ": hard.persona.timezone}
|
||||
|
||||
|
||||
def ensure_firefox_profile(
|
||||
profile_dir: Path,
|
||||
proxy_host: str,
|
||||
|
||||
@@ -194,6 +194,7 @@ class Settings:
|
||||
disable_ipv6_while_active: bool = False # disable IPv6 bindings while chain runs (Admin)
|
||||
harden_webrtc_enabled: bool = False # Chrome/Edge WebRTC policy (Admin)
|
||||
lan_lockdown_enabled: bool = False # kill LLMNR/NetBIOS/mDNS while chain runs
|
||||
telemetry_kill_enabled: bool = False # disable DiagTrack + Activity History (Admin)
|
||||
|
||||
# ── hardened browser ────────────────────────────────────────────────────
|
||||
firefox_path: str = ""
|
||||
@@ -210,6 +211,8 @@ class Settings:
|
||||
browser_first_party_isolation: bool = True
|
||||
browser_strict_tracking_protection: bool = True
|
||||
browser_timezone_utc: bool = True
|
||||
browser_persona: str = "blend_windows_chrome"
|
||||
browser_cookie_mode: str = "block_third_party"
|
||||
|
||||
# ── sources ───────────────────────────────────────────────────────────
|
||||
sources: list[str] = field(
|
||||
|
||||
@@ -35,6 +35,11 @@ from .privacy_lan import (
|
||||
engage_lan_lockdown,
|
||||
restore_lan,
|
||||
)
|
||||
from .telemetry_kill import (
|
||||
TelemetrySnapshot,
|
||||
engage_telemetry_kill,
|
||||
restore_telemetry,
|
||||
)
|
||||
from .sysproxy import (
|
||||
clear_system_proxy,
|
||||
detect_policy_overrides,
|
||||
@@ -82,6 +87,7 @@ class ChainService:
|
||||
self._ipv6_adapters: list[str] = []
|
||||
self._webrtc_was_applied: bool = False
|
||||
self._lan_snap: LanSnapshot | None = None
|
||||
self._telemetry_snap: TelemetrySnapshot | None = None
|
||||
|
||||
def _manual_exit_url(self) -> str | None:
|
||||
u = normalize_proxy_url(self._settings.manual_exit_proxy)
|
||||
@@ -181,6 +187,16 @@ class ChainService:
|
||||
elif s.lan_lockdown_enabled:
|
||||
self._notify({"type": "log", "text": "LAN lockdown enabled but not Admin — skipped."})
|
||||
|
||||
if s.telemetry_kill_enabled and is_admin():
|
||||
tsnap, tlogs = engage_telemetry_kill()
|
||||
self._telemetry_snap = tsnap
|
||||
for ln in tlogs[:8]:
|
||||
self._notify({"type": "log", "text": f"Telemetry: {ln}"})
|
||||
if len(tlogs) > 8:
|
||||
self._notify({"type": "log", "text": f"Telemetry: … +{len(tlogs) - 8} more changes."})
|
||||
elif s.telemetry_kill_enabled:
|
||||
self._notify({"type": "log", "text": "Telemetry kill enabled but not Admin — skipped."})
|
||||
|
||||
def _restore_privacy(self) -> None:
|
||||
if self._mac_originals:
|
||||
for ln in restore_macs(self._mac_originals):
|
||||
@@ -202,6 +218,10 @@ class ChainService:
|
||||
for ln in restore_lan(self._lan_snap):
|
||||
self._notify({"type": "log", "text": f"LAN: {ln}"})
|
||||
self._lan_snap = None
|
||||
if self._telemetry_snap is not None and is_admin():
|
||||
for ln in restore_telemetry(self._telemetry_snap)[:8]:
|
||||
self._notify({"type": "log", "text": f"Telemetry: {ln}"})
|
||||
self._telemetry_snap = None
|
||||
|
||||
def _run_thread(self) -> None:
|
||||
try:
|
||||
|
||||
292
proxy_chain_manager/telemetry_kill.py
Normal file
292
proxy_chain_manager/telemetry_kill.py
Normal file
@@ -0,0 +1,292 @@
|
||||
"""Reversibly disable Windows telemetry / data collection.
|
||||
|
||||
Targets, in order of leak severity:
|
||||
|
||||
• DiagTrack service — primary telemetry pipeline ("Connected User
|
||||
Experiences and Telemetry"). Phones home every
|
||||
few minutes.
|
||||
• dmwappushservice — WAP Push routing (DSMS), historically tied to
|
||||
DiagTrack. Disabled when present.
|
||||
• AllowTelemetry (HKLM) — DataCollection policy. 0 = lowest tier.
|
||||
• Activity History — EnableActivityFeed / Upload / PublishUserActivities.
|
||||
• Advertising ID — HKCU\\…\\AdvertisingInfo\\Enabled.
|
||||
• Cortana Bing search — HKCU\\…\\Search\\BingSearchEnabled.
|
||||
• Scheduled tasks — Compatibility Appraiser + CEIP Consolidator
|
||||
+ UsbCeip.
|
||||
|
||||
Every change is snapshotted before mutation; ``restore_telemetry`` rolls back.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import subprocess
|
||||
import winreg
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from .firewall import is_admin
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
_SERVICES = ("DiagTrack", "dmwappushservice")
|
||||
|
||||
_SCHED_TASKS = (
|
||||
r"\Microsoft\Windows\Application Experience\Microsoft Compatibility Appraiser",
|
||||
r"\Microsoft\Windows\Customer Experience Improvement Program\Consolidator",
|
||||
r"\Microsoft\Windows\Customer Experience Improvement Program\UsbCeip",
|
||||
)
|
||||
|
||||
# (hive, path, value_name, dword_when_killed)
|
||||
_REG_KILLS = (
|
||||
(winreg.HKEY_LOCAL_MACHINE,
|
||||
r"SOFTWARE\Policies\Microsoft\Windows\DataCollection",
|
||||
"AllowTelemetry", 0),
|
||||
(winreg.HKEY_LOCAL_MACHINE,
|
||||
r"SOFTWARE\Policies\Microsoft\Windows\System",
|
||||
"EnableActivityFeed", 0),
|
||||
(winreg.HKEY_LOCAL_MACHINE,
|
||||
r"SOFTWARE\Policies\Microsoft\Windows\System",
|
||||
"PublishUserActivities", 0),
|
||||
(winreg.HKEY_LOCAL_MACHINE,
|
||||
r"SOFTWARE\Policies\Microsoft\Windows\System",
|
||||
"UploadUserActivities", 0),
|
||||
(winreg.HKEY_LOCAL_MACHINE,
|
||||
r"SOFTWARE\Policies\Microsoft\Windows\CloudContent",
|
||||
"DisableWindowsConsumerFeatures", 1),
|
||||
(winreg.HKEY_CURRENT_USER,
|
||||
r"Software\Microsoft\Windows\CurrentVersion\AdvertisingInfo",
|
||||
"Enabled", 0),
|
||||
(winreg.HKEY_CURRENT_USER,
|
||||
r"Software\Microsoft\Windows\CurrentVersion\Search",
|
||||
"BingSearchEnabled", 0),
|
||||
(winreg.HKEY_CURRENT_USER,
|
||||
r"Software\Microsoft\Windows\CurrentVersion\Search",
|
||||
"CortanaConsent", 0),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class TelemetrySnapshot:
|
||||
service_state: dict[str, str] = field(default_factory=dict) # name -> 'auto'|'manual'|'disabled'|'absent'
|
||||
sched_tasks_state: dict[str, bool] = field(default_factory=dict) # task -> was_enabled
|
||||
reg_values: list[dict] = field(default_factory=list)
|
||||
|
||||
|
||||
def _run(args: list[str], timeout: float = 12.0) -> tuple[int, str, str]:
|
||||
try:
|
||||
r = subprocess.run(
|
||||
args, capture_output=True, text=True, timeout=timeout,
|
||||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
||||
)
|
||||
return r.returncode, r.stdout or "", r.stderr or ""
|
||||
except Exception as e:
|
||||
return 1, "", str(e)
|
||||
|
||||
|
||||
# ── service helpers ──────────────────────────────────────────────────────────
|
||||
|
||||
def _service_start_type(name: str) -> str:
|
||||
code, out, _ = _run(["sc", "qc", name])
|
||||
if code != 0:
|
||||
return "absent"
|
||||
for ln in out.splitlines():
|
||||
s = ln.strip().lower()
|
||||
if "start_type" in s:
|
||||
if "2" in s or "auto" in s:
|
||||
return "auto"
|
||||
if "3" in s or "demand" in s or "manual" in s:
|
||||
return "manual"
|
||||
if "4" in s or "disabled" in s:
|
||||
return "disabled"
|
||||
return "absent"
|
||||
|
||||
|
||||
def _set_service_disabled(name: str) -> bool:
|
||||
_run(["sc", "stop", name], timeout=15)
|
||||
code, _, _ = _run(["sc", "config", name, "start=", "disabled"])
|
||||
return code == 0
|
||||
|
||||
|
||||
def _set_service_start(name: str, want: str) -> bool:
|
||||
start_value = {"auto": "auto", "manual": "demand", "disabled": "disabled"}.get(want)
|
||||
if not start_value:
|
||||
return False
|
||||
_run(["sc", "stop", name], timeout=10)
|
||||
code, _, _ = _run(["sc", "config", name, "start=", start_value])
|
||||
if want == "auto":
|
||||
_run(["sc", "start", name], timeout=15)
|
||||
return code == 0
|
||||
|
||||
|
||||
# ── scheduled-task helpers ──────────────────────────────────────────────────
|
||||
|
||||
def _task_is_enabled(name: str) -> bool | None:
|
||||
code, out, _ = _run(["schtasks", "/Query", "/TN", name])
|
||||
if code != 0:
|
||||
return None
|
||||
for ln in out.splitlines():
|
||||
s = ln.strip().lower()
|
||||
if s.startswith("status:") or "ready" in s or "disabled" in s:
|
||||
if "disabled" in s:
|
||||
return False
|
||||
if "ready" in s or "running" in s:
|
||||
return True
|
||||
return True # assume ready when query succeeded
|
||||
|
||||
|
||||
def _set_task_state(name: str, enable: bool) -> bool:
|
||||
code, _, _ = _run([
|
||||
"schtasks", "/Change", "/TN", name,
|
||||
"/Enable" if enable else "/Disable",
|
||||
])
|
||||
return code == 0
|
||||
|
||||
|
||||
# ── registry helpers ────────────────────────────────────────────────────────
|
||||
|
||||
def _read_reg_dword(hive: int, path: str, name: str) -> tuple[bool, int | None]:
|
||||
"""Returns (present, value)."""
|
||||
try:
|
||||
with winreg.OpenKey(hive, path, 0, winreg.KEY_QUERY_VALUE) as key:
|
||||
v, _ = winreg.QueryValueEx(key, name)
|
||||
return True, int(v)
|
||||
except OSError:
|
||||
return False, None
|
||||
|
||||
|
||||
def _write_reg_dword(hive: int, path: str, name: str, value: int) -> bool:
|
||||
try:
|
||||
with winreg.CreateKeyEx(hive, path, 0, winreg.KEY_SET_VALUE) as key:
|
||||
winreg.SetValueEx(key, name, 0, winreg.REG_DWORD, value)
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def _delete_reg_value(hive: int, path: str, name: str) -> None:
|
||||
try:
|
||||
with winreg.OpenKey(hive, path, 0, winreg.KEY_SET_VALUE) as key:
|
||||
winreg.DeleteValue(key, name)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
# ── public API ──────────────────────────────────────────────────────────────
|
||||
|
||||
def telemetry_status() -> dict[str, str]:
|
||||
"""Read-only snapshot for the audit panel."""
|
||||
out: dict[str, str] = {}
|
||||
for svc in _SERVICES:
|
||||
out[f"svc.{svc}"] = _service_start_type(svc)
|
||||
pres, val = _read_reg_dword(
|
||||
winreg.HKEY_LOCAL_MACHINE,
|
||||
r"SOFTWARE\Policies\Microsoft\Windows\DataCollection",
|
||||
"AllowTelemetry",
|
||||
)
|
||||
out["AllowTelemetry"] = "0 (killed)" if pres and val == 0 else (
|
||||
f"{val}" if pres else "unset"
|
||||
)
|
||||
pres, val = _read_reg_dword(
|
||||
winreg.HKEY_CURRENT_USER,
|
||||
r"Software\Microsoft\Windows\CurrentVersion\AdvertisingInfo",
|
||||
"Enabled",
|
||||
)
|
||||
out["AdvertisingID"] = "off" if pres and val == 0 else (
|
||||
"on" if pres and val == 1 else "unset"
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def engage_telemetry_kill() -> tuple[TelemetrySnapshot | None, list[str]]:
|
||||
logs: list[str] = []
|
||||
if not is_admin():
|
||||
return None, ["Telemetry kill skipped — needs Administrator."]
|
||||
snap = TelemetrySnapshot()
|
||||
|
||||
for svc in _SERVICES:
|
||||
st = _service_start_type(svc)
|
||||
snap.service_state[svc] = st
|
||||
if st == "absent":
|
||||
continue
|
||||
if _set_service_disabled(svc):
|
||||
logs.append(f"Service '{svc}' stopped + disabled (was {st}).")
|
||||
else:
|
||||
logs.append(f"Service '{svc}' could not be disabled.")
|
||||
|
||||
for task in _SCHED_TASKS:
|
||||
was = _task_is_enabled(task)
|
||||
if was is None:
|
||||
continue
|
||||
snap.sched_tasks_state[task] = was
|
||||
if was:
|
||||
if _set_task_state(task, enable=False):
|
||||
logs.append(f"Scheduled task disabled: {task}")
|
||||
|
||||
for hive, path, name, kill_val in _REG_KILLS:
|
||||
present, prev = _read_reg_dword(hive, path, name)
|
||||
snap.reg_values.append({
|
||||
"hive": "HKLM" if hive == winreg.HKEY_LOCAL_MACHINE else "HKCU",
|
||||
"path": path,
|
||||
"name": name,
|
||||
"present": present,
|
||||
"prev": prev,
|
||||
})
|
||||
if _write_reg_dword(hive, path, name, kill_val):
|
||||
logs.append(f"reg: {('HKLM' if hive == winreg.HKEY_LOCAL_MACHINE else 'HKCU')}\\{path}\\{name} → {kill_val}")
|
||||
else:
|
||||
logs.append(f"reg write failed: {path}\\{name}")
|
||||
|
||||
return snap, logs
|
||||
|
||||
|
||||
def restore_telemetry(snap: TelemetrySnapshot | None) -> list[str]:
|
||||
if snap is None or not is_admin():
|
||||
return []
|
||||
logs: list[str] = []
|
||||
|
||||
for svc, prev in snap.service_state.items():
|
||||
if prev == "absent":
|
||||
continue
|
||||
if _set_service_start(svc, prev):
|
||||
logs.append(f"Service '{svc}' restored → {prev}.")
|
||||
|
||||
for task, was in snap.sched_tasks_state.items():
|
||||
if was:
|
||||
_set_task_state(task, enable=True)
|
||||
logs.append(f"Scheduled task re-enabled: {task}")
|
||||
|
||||
for entry in snap.reg_values:
|
||||
hive = winreg.HKEY_LOCAL_MACHINE if entry["hive"] == "HKLM" else winreg.HKEY_CURRENT_USER
|
||||
path = entry["path"]
|
||||
name = entry["name"]
|
||||
if entry["present"]:
|
||||
_write_reg_dword(hive, path, name, int(entry["prev"]))
|
||||
logs.append(f"reg restore: {entry['hive']}\\{path}\\{name} ← {entry['prev']}")
|
||||
else:
|
||||
_delete_reg_value(hive, path, name)
|
||||
logs.append(f"reg removed (was absent): {entry['hive']}\\{path}\\{name}")
|
||||
return logs
|
||||
|
||||
|
||||
def snapshot_to_json(snap: TelemetrySnapshot | None) -> str:
|
||||
if snap is None:
|
||||
return ""
|
||||
return json.dumps({
|
||||
"services": snap.service_state,
|
||||
"tasks": snap.sched_tasks_state,
|
||||
"reg": snap.reg_values,
|
||||
})
|
||||
|
||||
|
||||
def snapshot_from_json(raw: str) -> TelemetrySnapshot | None:
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
d = json.loads(raw)
|
||||
return TelemetrySnapshot(
|
||||
service_state=dict(d.get("services") or {}),
|
||||
sched_tasks_state={k: bool(v) for k, v in (d.get("tasks") or {}).items()},
|
||||
reg_values=list(d.get("reg") or []),
|
||||
)
|
||||
except Exception:
|
||||
return None
|
||||
Reference in New Issue
Block a user