telemetry.py
Rama main · réplica de lectura del repositorio original.
"""
Anonymous usage stats: which `squidbrake` commands run, and counts of what the gateway checked, held and blocked.
They are always on. The first time you run `squidbrake` in a terminal it says so once (and asks, optionally, where
you heard about Squidbrake). Scripts, CI, `--yes` and the agent hooks send nothing until that first run happened.
A gateway that starts without one (Docker, a background service) writes the same notice to its log and then shares
its counts (gateway_start); never in CI or with the off switches.
squidbrake telemetry status exactly what is sent
squidbrake register EMAIL tell the Squidbrake team who you are (optional; asks first; also asked once on
the first run, Enter skips it)
What is sent, once each time you run a command:
- a random id made on this computer (not tied to you unless you give your email, below)
- the command name only (e.g. "connect", "doctor", "undo"), never its arguments
- the Squidbrake version, operating system, CPU type and Python version, and whether it came from pip
- the country, worked out by the stats service from the address the request comes from
- where you heard about Squidbrake, if you answered (one word from a list, e.g. "linkedin"), or the name of the
link you installed from (SQUIDBRAKE_REF, e.g. "whatsapp")
Never sent: commands an agent ran, file or folder names, file contents, prompts, rules, keys, the audit trail,
names of people, your name, work email and the coding agents installed here (only if you give your email, on the
first run or with `squidbrake register`).
The agent hooks (`hook`, `agent-hook`, `shell-guard`, `proxy`) send nothing, so nothing slows a tool call down.
The gateway's usage counts go to the Squidbrake team the same way, exactly as a pilot's do (see pilot.py for every
field): how many actions were allowed, held and blocked per day, and for each one held or blocked, the program
only (e.g. "rm"), the rule, what happened and its size as numbers. Never the command, files or prompts.
"""
from __future__ import annotations
import atexit
import json
import os
import platform
import re
import sys
import threading
import uuid
from datetime import datetime, timezone
from pathlib import Path
# A PostHog project key only lets someone send events, not read them, so it is safe to ship in the package.
SHIPPED_KEY = "phc_muPqCNPamFR8iPaewdgVWqdcrGburJSXeySGHqXatZnU" # PostHog > Project settings (US); send-only
POSTHOG_KEY = os.getenv("SQUIDBRAKE_POSTHOG_KEY", SHIPPED_KEY)
POSTHOG_HOST = os.getenv("SQUIDBRAKE_POSTHOG_HOST", "https://us.i.posthog.com")
QUIET = {"hook", "agent-hook", "shell-guard", "proxy", "telemetry", "register"} # never ask, never send
COMMANDS = {"start", "setup", "connect", "doctor", "lockdown", "evidence", "undo", "pilot", "add-key", "keys", "verify"}
WHAT_IS_SENT = __doc__.split("What is sent, once each time you run a command:")[1].strip()
# the gateway's counts go where `squidbrake connect all` sends them (connect.py), as the community "pilot"
COMMUNITY_SERVER, COMMUNITY_CODE = "https://pilots.squidbrake.com", "community-opt-in-ins-a42929"
# Where people heard about Squidbrake: one word, from this list or from the link they installed from (SQUIDBRAKE_REF).
SOURCES = [("linkedin", "LinkedIn"), ("github", "GitHub"), ("hn-reddit", "Hacker News or Reddit"),
("community", "WhatsApp, Discord or another community"), ("friend", "A friend or colleague"),
("search", "A search or an AI assistant"), ("other", "Somewhere else")]
REF_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,29}$")
def ref_from_env() -> str | None:
ref = os.getenv("SQUIDBRAKE_REF", "").strip().lower()
return ref if REF_RE.match(ref) else None
EMAIL_RE = re.compile(r"^[^@\s]{1,64}@[^@\s]+\.[^@\s]{2,}$")
def installed_agents() -> list[str]:
"""Names only of the coding agents installed here (the ones connect.py can hook), e.g. ["claude-code", "cursor"]."""
try:
import shutil
import connect
found = [n for n, t in connect.hook_agents().items() if t["present"]]
if shutil.which("claude") or (Path.home() / ".claude").exists():
found.insert(0, "claude-code")
return found
except Exception: # never break a command over this
return []
def _identify(cfg: dict, email: str, version: str, name: str | None = None, company: str | None = None) -> None:
cfg.setdefault("id", str(uuid.uuid4()))
cfg["email"] = email
_save(cfg)
props = {"email": email, "agents": installed_agents(), "version": version,
**({"name": name} if name else {}), **({"company": company} if company else {})}
_send("$identify", cfg["id"], {"$set": props}, wait=True)
def _ask_contact() -> tuple[str, str | None] | None:
"""Optional, once: work email and name, so the team can reach the person. Enter skips; nothing is sent then."""
print("Optional: your work email, so the Squidbrake team can help you set it up (Enter to skip).")
email = (_ask("Work email: ") or "")[:254]
if not EMAIL_RE.match(email):
if email:
print("That doesn't look like an email; skipped. Later: squidbrake register EMAIL")
print()
return None
name = (_ask("Your name: ") or "")[:120] or None
print(f"Sent with it: the coding agents installed here ({', '.join(installed_agents()) or 'none found'}).\n")
return email, name
def _ask_source() -> str | None:
print("One optional question: where did you hear about Squidbrake?")
for i, (_, label) in enumerate(SOURCES, 1):
print(f" {i}. {label}")
answer = _ask("Number (Enter to skip): ")
print()
if answer and answer.isdigit() and 1 <= int(answer) <= len(SOURCES):
return SOURCES[int(answer) - 1][0]
return None
def _path() -> Path:
return Path(os.getenv("SQUIDBRAKE_HOME") or Path.home() / ".squidbrake") / "telemetry.json"
def load() -> dict:
try:
return json.loads(_path().read_text(encoding="utf-8"))
except (OSError, ValueError):
return {}
def _save(cfg: dict) -> None:
try:
_path().parent.mkdir(parents=True, exist_ok=True)
_path().write_text(json.dumps(cfg, indent=2), encoding="utf-8")
except OSError:
pass
def _switched_off() -> bool:
if os.getenv("SQUIDBRAKE_TELEMETRY", "").strip().lower() in ("0", "false", "off", "no"):
return True
if os.getenv("DO_NOT_TRACK", "").strip().lower() in ("1", "true", "yes"):
return True
return bool(os.getenv("CI"))
def _pilot_dir() -> Path:
"""Where the gateway keeps pilot.json: the same place server.py works out (next to keys.json)."""
if os.getenv("KEYS_PATH"):
return Path(os.environ["KEYS_PATH"]).parent
here = Path(__file__).resolve().parent
home = os.getenv("SQUIDBRAKE_HOME") or (Path.home() / ".squidbrake" if (here / "__init__.py").exists() else here)
return Path(home) / "data"
def _share_counts(on: bool, version: str, ref: str | None = None, contact: tuple[str, str | None] | None = None) -> bool:
"""Join (or leave) the community pilot, so the gateway sends its counts. A real pilot's membership is kept.
Returns whether this install is now sharing counts (False when joining failed, e.g. no network)."""
import contextlib
import io
try:
import pilot
home = _pilot_dir()
mine = pilot.load(home)
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
if on and (not mine or (contact and mine.get("code") == COMMUNITY_CODE)):
# (already sharing, e.g. a gateway joined first: the same install moves onto the email's own pilot)
pilot.join(home, COMMUNITY_CODE, COMMUNITY_SERVER, True, version, ref=ref, contact=contact)
elif not on and mine and mine.get("code") == COMMUNITY_CODE:
pilot.leave(home)
home.mkdir(parents=True, exist_ok=True) # older versions' `connect all` asked about counts: it's settled
(home / "community-asked").write_text("said in the first-run notice (telemetry.py)\n", encoding="utf-8")
return bool(pilot.load(home))
except Exception: # sharing counts must never break a command
return False
def _today() -> str:
return datetime.now(timezone.utc).date().isoformat()
def _interactive() -> bool:
try:
return sys.stdin.isatty() and sys.stdout.isatty()
except (AttributeError, ValueError):
return False
def _ask(question: str) -> str | None:
try:
return input(question).strip()
except (EOFError, KeyboardInterrupt):
print()
return None
NOTICE = """\
Squidbrake sends anonymous usage counts to its team, so one small team can see what to fix next:
- which squidbrake commands run, the version, OS and country
- how many actions it allowed, held and blocked; for each one held or blocked, the program only
(e.g. "rm"), the rule, and sizes as numbers
Never your commands, files, prompts, rules or keys. Exactly what is sent: squidbrake telemetry status"""
def _first_run(cfg: dict, version: str) -> dict:
"""The first `squidbrake` command in a terminal: say once what is sent (it isn't a question), and ask,
optionally, where they heard about Squidbrake."""
print("\n" + NOTICE + "\n")
cfg = {"id": cfg.get("id") or str(uuid.uuid4()), "enabled": True,
"asked": datetime.now(timezone.utc).date().isoformat()}
if ref := ref_from_env() or _ask_source():
cfg["ref"] = ref
_save(cfg)
if contact := _ask_contact():
_identify(cfg, contact[0], version, name=contact[1])
cfg["counts_joined"] = _share_counts(True, version, cfg.get("ref"), contact)
if not cfg["counts_joined"]:
cfg["counts_retry"] = _today() # tried today: the next try is tomorrow (see maybe)
_save(cfg)
return cfg
def gateway_start(version: str, say=print) -> bool:
"""A gateway that never had a terminal first run (Docker, a background service): say in its log what is sent,
then share its counts the way a first run does. Nothing when switched off, already sharing, or settled before
(someone who left with `squidbrake pilot leave` isn't joined again). Returns whether it joined now."""
if _switched_off():
return False
try:
import pilot
home = _pilot_dir()
if pilot.load(home) or (home / "community-asked").exists():
return False
except Exception: # never break a gateway start over this
return False
cfg = load()
if cfg.get("enabled") is False:
return False
# telemetry.json is left alone, so a later first run in a terminal still asks its optional questions (and an
# email given then moves these counts onto that person's pilot: _share_counts). Offline: the next start tries again.
say(NOTICE + "\nTurn it off: SQUIDBRAKE_TELEMETRY=0 or DO_NOT_TRACK=1, then restart.")
joined = _share_counts(True, version, cfg.get("ref") or ref_from_env())
if not joined: # offline: unsettled again, so the next start tries again
(home / "community-asked").unlink(missing_ok=True)
return joined
def _props(command: str, version: str, ref: str | None = None) -> dict:
here = Path(__file__).resolve()
extra = {"ref": ref, "$set_once": {"ref": ref}} if ref else {}
return {**extra, "command": command if command in COMMANDS else "other", "version": version,
"os": platform.system(), "os_release": platform.release(), "arch": platform.machine(),
"python": "%d.%d" % sys.version_info[:2], "install": "pip" if "site-packages" in here.parts else "source",
"$lib": "squidbrake-cli"}
def _post(body: dict) -> None:
try:
import httpx
httpx.post(POSTHOG_HOST.rstrip("/") + "/i/v0/e/", json=body, timeout=3)
except Exception: # stats must never break or slow down a command
pass
def _send(event: str, distinct_id: str, props: dict, wait: bool = False) -> None:
body = {"api_key": POSTHOG_KEY, "event": event, "distinct_id": distinct_id, "properties": props,
"timestamp": datetime.now(timezone.utc).isoformat()}
t = threading.Thread(target=_post, args=(body,), daemon=True)
t.start()
if wait:
t.join(3)
else:
atexit.register(t.join, 2) # short commands: give it up to 2 seconds to finish on the way out
def maybe(argv: list[str], version: str) -> None:
"""Called once per `squidbrake` command. Says what is sent the first time in a terminal, then sends."""
command = argv[0] if argv and not argv[0].startswith("-") else "start"
if command in QUIET or _switched_off():
return
cfg = load()
if "enabled" not in cfg:
# not during `pilot join` either: the installer runs it, and a founder with their own code must land in
# their pilot, not in the community one
if command == "pilot" or not _interactive() or "--yes" in argv or any(a in ("-h", "--help") for a in argv):
return
cfg = _first_run(cfg, version)
if cfg.get("enabled") and cfg.get("counts_joined") is False and cfg.get("counts_retry") != _today():
# joining the counts failed (offline, server down): try again, once a day, until it works.
# Someone who later left (squidbrake pilot leave) has counts_joined True, so they're never joined again.
cfg["counts_retry"] = _today()
cfg["counts_joined"] = _share_counts(True, version, cfg.get("ref"),
(cfg["email"], None) if cfg.get("email") else None)
_save(cfg)
if cfg.get("enabled") and not cfg.get("ref") and (ref := ref_from_env()):
cfg["ref"] = ref # said yes before, then installed again from a named link
_save(cfg)
if cfg.get("enabled") and POSTHOG_KEY:
_send("cli_command", cfg["id"], _props(command, version, cfg.get("ref")))
def register(argv: list[str], version: str) -> int:
args = [a for a in argv if a != "--yes"]
opts = {}
for flag in ("--company", "--name"):
if flag in args:
i = args.index(flag)
opts[flag[2:]] = (" ".join(args[i + 1:i + 2]) or "")[:160] or None
del args[i:i + 2]
company, name = opts.get("company"), opts.get("name")
if len(args) != 1 or not EMAIL_RE.match(args[0]):
print("usage: squidbrake register EMAIL [--name NAME] [--company NAME] [--yes]")
return 2
if not POSTHOG_KEY:
print("This build of Squidbrake has no stats service set up; nothing was sent.")
return 1
email = args[0]
print(f"This sends to the Squidbrake team: {email}" + (f", name {name}" if name else "") +
(f", company {company}" if company else "") +
f", the coding agents installed here ({', '.join(installed_agents()) or 'none found'})"
", and this computer's random usage id, so they can reach you.")
if "--yes" not in argv and (_ask("Send it? [Y/n] ") or "").lower() not in ("", "y", "yes"):
print("Nothing was sent.")
return 0
_identify(load(), email, version, name=name, company=company)
try:
import pilot
if (pilot.load(_pilot_dir()) or {}).get("code") == COMMUNITY_CODE: # anonymous counts: under this email now
import contextlib, io
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
# same install id: the server moves this install onto the email's own pilot
pilot.join(_pilot_dir(), COMMUNITY_CODE, COMMUNITY_SERVER, True, version, contact=(email, name))
except Exception: # never break a command over this
pass
print("Sent. Thanks, the team will be in touch.")
return 0
def main(argv: list[str], version: str = "") -> int:
cfg = load()
action = argv[0] if argv else "status"
if action != "status":
print("usage: squidbrake telemetry status\n\n" + NOTICE)
return 2
if _switched_off():
state = "off (CI, DO_NOT_TRACK or SQUIDBRAKE_TELEMETRY is set)"
else:
state = {True: "on", False: "off (you said no when an earlier version asked; that is kept)"}.get(
cfg.get("enabled"), "on (from the first squidbrake command you run in a terminal)")
print(f"Anonymous usage stats: {state}")
if cfg.get("email"):
print(f"Registered email: {cfg['email']}")
print(f"Settings: {_path()}\n\nWhat is sent, once each time you run a command:\n {WHAT_IS_SENT}")
return 0