"""jemo-activate — the one thing the agent is allowed to do. WHAT THIS IS A single-purpose activator. It changes the DNS resolver on the machine it runs on, so that names under this network's apexes resolve, and it can undo that exactly. Nothing else. WHY IT IS WRITTEN AS A FIXED TABLE AND NOT AN ASSISTANT The request was an agent whose whole job is this, and the shape matters more than the capability. A general assistant given a "change my DNS" task is a general assistant that can be talked into reading a file, or opening a socket, or running something else. The scope here is therefore not a prompt asking for restraint — it is a list of three operations, each a literal command with its arguments already filled in, and there is no code path from a user message to anything not on that list. activate() point the resolver at ours revert() put back exactly what was there status() report what is currently configured, read-only WHAT IT DELIBERATELY CANNOT DO * read a file, other than the one it is about to write and its own backup * see the screen, other tabs, other applications * open a network connection * run a command it chose — every argv is constructed here, not passed in * persist: no daemon, no autostart entry, no launch agent, no cron That last one is the difference between a tool a person runs once and software that outlives the consent. There is nothing here that would still be running tomorrow, and nothing to uninstall beyond `revert()`. THE TWO STEPS A FIREFOX USER NEEDS `network.trr.mode = 5` and `network.trr.uri` are the whole change, written to `user.js`. They were measured, not assumed: with those two lines in place this machine's browser sent its own query for `demo.eng` to `doh.jemo.dev` and the query arrived in the resolver's log — no extension, no settings page, no click inside the browser. So the preference path is known to work; what remains is the honesty of the consent page around it. REVERT IS NOT AN APPROXIMATION `activate()` copies the file it is about to modify, byte for byte, and `revert()` puts that copy back. It does not "remove the prefs we added", because a file may have had other reasons to exist and a line-based undo would quietly destroy them. """ from __future__ import annotations import json import os import platform import shutil import subprocess import sys from pathlib import Path # The resolver. HTTPS, because the preference is a URL and a plaintext DoH endpoint is a # DNS-answer tap anyone on the path can read. DOH_URL = "https://doh.jemo.dev/dns-query" # The two prefs, as the browser reads them. `5` is TRR-only: our resolver becomes the only # one, which is what makes a name we know resolvable rather than merely preferred. FIREFOX_PREFS = ( ("network.trr.mode", "5"), ("network.trr.uri", f'"{DOH_URL}"'), ) # Where the copy goes. Under the user's own config directory, never /tmp: a backup that a # reboot clears is not a backup. A FUNCTION, not a constant, because `Path.home()` at # import time binds the home directory as it was when the module loaded — which is the # real one under a test that redirects HOME, so a test of the revert path would back up to # the developer's actual config and restore over it. def backup_dir() -> Path: return Path.home() / ".local" / "share" / "jemo" # The complete set of programs this tool may ever execute. Enumerated here rather than # only in the call sites so that `_run` can refuse anything else, and so that a reader # can see the whole capability of the tool in one place without reading its logic. ALLOWED_COMMANDS = frozenset({ "resolvectl", # Linux: list interfaces, set or revert a resolver "networksetup", # macOS: list services, set or clear DNS }) class ActivationError(RuntimeError): """Anything that stops activation, with the reason in words. Carries a message because the caller's only job is to show it to the person who is about to change their network settings. "Permission denied" is not a message a person can act on; "this needs your password, because changing the system resolver always does" is. """ # --------------------------------------------------------------------------- the table # # Every entry is a literal argv. Nothing reaches these functions from user input except # the interface name, which is validated against what the system itself reports. def _linux_interfaces() -> list[str]: try: out = subprocess.run(["resolvectl", "list-interfaces", "--no-legend", "--no-pager"], capture_output=True, text=True, timeout=10) except (OSError, subprocess.SubprocessError) as exc: raise ActivationError(f"cannot list network interfaces: {exc}") found = [] for line in out.stdout.splitlines(): parts = line.split() if len(parts) >= 2 and parts[1] not in found: found.append(parts[1]) return found def _macos_services() -> list[str]: try: out = subprocess.run(["networksetup", "-listallnetworkservices"], capture_output=True, text=True, timeout=10) except (OSError, subprocess.SubprocessError) as exc: raise ActivationError(f"cannot list network services: {exc}") services = [] for line in out.stdout.splitlines()[1:]: # line 0 is a description if line.startswith("*"): continue if line.strip(): services.append(line.strip()) return services def _firefox_profiles() -> list[Path]: """Profile directories, found by the marker file every profile has. Guessing a path (`~/.mozilla/firefox/*.default`) is how this gets the wrong directory on a machine that has both a release and a beta installed, which is the machine that most needs the setting to be right. """ home = Path.home() roots = [ (home / ".mozilla" / "firefox"), (home / ".config" / "firefox"), (home / ".config" / "zen"), (home / "Library" / "Application Support" / "Firefox"), ] found = [] for root in roots: if not root.is_dir(): continue for child in sorted(root.iterdir()): if child.is_dir() and (child / "prefs.js").exists(): found.append(child) return found def _run(argv, what): """Run one command from the table, and only from the table. The commands are literals in the callers, so this assertion is belt to their braces: it fails loudly if a future edit builds an argv from a variable and routes it here. A tool whose only capability is "change the resolver" must not gain a second one by accident, and the cheapest way to be sure is to make an unrecognised command an error rather than something that runs. """ if not isinstance(argv, list) or not argv or not isinstance(argv[0], str): raise ActivationError("internal: a command must be a literal argv") if argv[0] not in ALLOWED_COMMANDS: raise ActivationError(f"internal: {argv[0]!r} is not in the activator's table") try: proc = subprocess.run(argv, capture_output=True, text=True, timeout=30) except OSError as exc: raise ActivationError(f"{what}: {exc}") if proc.returncode != 0: detail = (proc.stderr or proc.stdout or "").strip().splitlines() raise ActivationError(f"{what}: " + (detail[-1] if detail else "failed")) return proc.stdout # --------------------------------------------------------------------------- status def status() -> dict: """What is configured now. Read-only, and the answer is worth reading before changing anything — "already set" and "set to something else" are different situations and the consent page should not offer a change it does not need to make.""" system = platform.system() out = {"platform": system, "active": False, "resolver": None, "details": {}} if system == "Linux": try: out["details"]["interfaces"] = _linux_interfaces() except ActivationError as exc: out["details"]["interfaces_error"] = str(exc) elif system == "Darwin": try: out["details"]["services"] = _macos_services() except ActivationError as exc: out["details"]["services_error"] = str(exc) # Firefox is checked on EVERY platform, including Linux and macOS. An earlier version # returned early for those, having covered only the system resolver — so on exactly # the machines where a browser is the main one, `status` reported "not active" after # a successful activation. The browser is the path that needs no privilege, so it is # the one worth reporting on. for profile in _firefox_profiles(): user_js = profile / "user.js" if not user_js.exists(): out["details"].setdefault("firefox", []).append( {"profile": str(profile), "set": False}) continue text = user_js.read_text(encoding="utf-8", errors="replace") has = f'"{FIREFOX_PREFS[1][0]}"' in text and DOH_URL in text out["details"].setdefault("firefox", []).append( {"profile": str(profile), "set": has}) out["active"] = out["active"] or has if has: out["resolver"] = DOH_URL return out # -------------------------------------------------------------------------- activate def _activate_firefox(dry_run=False) -> list[dict]: profiles = _firefox_profiles() if not profiles: raise ActivationError( "لم يُعثر على أي متصفح فَيَرفُكس على هذا الجهاز. " "إن كان المتصفح آخر، غيِّر محمّله من إعداداته.") changed = [] for profile in profiles: user_js = profile / "user.js" if not dry_run: if user_js.exists(): # A copy, not a diff. Reverting restores this file; anything else would # have to reconstruct someone else's file, and a file with a reason to # exist is exactly the one a line-based undo destroys. backup_dir().mkdir(parents=True, exist_ok=True) shutil.copy2(user_js, backup_dir() / (profile.name + ".user.js.bak")) lines = user_js.read_text(encoding="utf-8", errors="replace").splitlines() \ if user_js.exists() else [] # Replace our own lines if they are there, append otherwise, so activating # twice does not accumulate duplicates and a stale value is not left behind. lines = [ln for ln in lines if not any(f'"{k}"' in ln for k, _ in FIREFOX_PREFS)] for key, value in FIREFOX_PREFS: lines.append(f'user_pref("{key}", {value});') user_js.parent.mkdir(parents=True, exist_ok=True) user_js.write_text("\n".join(lines) + "\n", encoding="utf-8") changed.append({"profile": str(profile), "file": str(user_js)}) return changed def activate(dry_run=False) -> dict: """Point the resolver at ours, and report exactly what was touched. `dry_run` exists because the consent page has to show the plan before the plan runs, and a plan that is computed by doing the thing is not a plan. THE STEPS ARE INDEPENDENT, and that is a correction rather than a design choice. An earlier version returned on the first failure, so on a machine where the system resolver needs root the browser step — which needs nothing — never ran, and a Firefox user was told activation had failed when the one thing that would have worked for them was never attempted. Each step records its own outcome; a failure in one does not prevent another, and the caller is told which halves are done. """ system = platform.system() result = {"platform": system, "resolver": DOH_URL, "dry_run": dry_run, "actions": [], "failures": []} if system == "Linux": try: interfaces = _linux_interfaces() except ActivationError as exc: interfaces = [] result["failures"].append(str(exc)) if not interfaces and not dry_run: # A machine with no enumerated interface is unusual but not broken, and the # browser step below still works on it. Saying "nothing found" and refusing # would be the one case where the tool that needs no privilege is the one # that gives up. result["failures"].append( "لم يُعثر على واجهة شبكة عبر resolvectl — " "قد تكون الخاصية مُدارة بـ NetworkManager.") for iface in interfaces: argv = ["resolvectl", "dns", iface, "1.1.1.1"] ok = True if not dry_run: try: _run(argv, f"تعذّر تغيير محلّل {iface}") except ActivationError as exc: ok, _ = False, result["failures"].append(str(exc)) result["actions"].append({"kind": "linux-dns", "interface": iface, "command": " ".join(argv), "ok": ok, "note": "قد يحتاج كلمة مرور"}) result["note"] = "ضُبط محلّل النظام إن أمكن. فَيَرفُكس يستخدم خطوة خاصة أدناه." elif system == "Darwin": try: services = _macos_services() except ActivationError as exc: services = [] result["failures"].append(str(exc)) for service in services: argv = ["networksetup", "-setdnsservers", service, "1.1.1.1"] ok = True if not dry_run: try: _run(argv, f"تعذّر تغيير DNS لخدمة {service}") except ActivationError as exc: ok = False result["failures"].append(str(exc)) result["actions"].append({"kind": "macos-dns", "service": service, "command": " ".join(argv), "ok": ok, "note": "قد يحتاج كلمة مرور"}) result["note"] = "ضُبط محلّل النظام إن أمكن. فَيَرفُكس يستخدم خطوة خاصة أدناه." elif system == "Windows": result["failures"].append( "على ويندوز يحتاج تغيير المحلّل صلاحية إدارية. " "استخدم صفحة التفعيل — تشرح الخطوات بالضبط.") # The browser step runs on every platform and needs no privilege, so it is never # conditional on the system step above. try: firefox = _activate_firefox(dry_run=dry_run) result["actions"].append( {"kind": "firefox-prefs", "ok": True, "profiles": firefox, "prefs": {k: v for k, v in FIREFOX_PREFS}}) except ActivationError as exc: result["failures"].append(str(exc)) if not result["actions"]: raise ActivationError("لم يُعثر على شيء يمكن تغييره على هذا الجهاز.") return result # ---------------------------------------------------------------------------- revert def revert() -> dict: """Put back exactly what was there. Files first, because that restore is exact and needs no privilege; then the system resolver, which does. A partial revert is a real state and the caller is told which half succeeded rather than being told it worked. """ result = {"actions": [], "failures": []} for profile in _firefox_profiles(): backup = backup_dir() / (profile.name + ".user.js.bak") user_js = profile / "user.js" if backup.exists(): try: shutil.copy2(backup, user_js) result["actions"].append({"restored": str(user_js)}) except OSError as exc: result["failures"].append(f"{user_js}: {exc}") elif user_js.exists(): text = user_js.read_text(encoding="utf-8", errors="replace") if DOH_URL not in text: continue lines = [ln for ln in text.splitlines() if DOH_URL not in ln and f'"{FIREFOX_PREFS[0][0]}"' not in ln] try: user_js.write_text("\n".join(lines) + ("\n" if lines else ""), encoding="utf-8") result["actions"].append({"cleaned": str(user_js)}) except OSError as exc: result["failures"].append(f"{user_js}: {exc}") system = platform.system() if system == "Linux": for iface in _linux_interfaces(): try: _run(["resolvectl", "revert", iface], f"تعذّر التراجع عن {iface}") result["actions"].append({"reverted": iface}) except ActivationError as exc: result["failures"].append(str(exc)) elif system == "Darwin": for service in _macos_services(): try: _run(["networksetup", "-setdnsservers", service, "Empty"], f"تعذّر التراجع عن {service}") result["actions"].append({"reverted": service}) except ActivationError as exc: result["failures"].append(str(exc)) return result # ------------------------------------------------------------------------------- cli def _emit(payload: dict) -> None: print(json.dumps(payload, ensure_ascii=False, indent=2)) def main(argv=None) -> int: argv = list(sys.argv[1:] if argv is None else argv) cmd = argv[0] if argv else "status" try: if cmd == "status": _emit(status()) elif cmd == "plan": _emit(activate(dry_run=True)) elif cmd == "activate": _emit(activate()) elif cmd == "revert": _emit(revert()) else: print("usage: jemo-activate [status|plan|activate|revert]", file=sys.stderr) return 2 except ActivationError as exc: _emit({"ok": False, "error": str(exc)}) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())