Dans cette fenêtre, seuls s’ouvrent les sites libres choisis par tes parents (Pronote, ENT…) et les pages vers lesquelles ils mènent.
Pour un autre site, utilise le navigateur habituel : il compte alors dans ton temps d’écran.
""" EXT_BLOCKED_JS = 'document.getElementById("retour").onclick = () => history.length > 1 ? history.back() : window.close();\n' def write_free_extension(path, free_domains): """Écrit l'extension du navigateur des sites libres (réécrite à chaque ouverture, avec la liste du moment).""" os.makedirs(path, exist_ok=True) files = { "manifest.json": json.dumps(EXT_MANIFEST, ensure_ascii=False, indent=1), "sw.js": EXT_SW, "bloque.html": EXT_BLOCKED_HTML, "bloque.js": EXT_BLOCKED_JS, "config.json": json.dumps({"free": sorted(set(free_domains) | set(BUILTIN_FREE)), "deny": list(ENTERTAINMENT)}), "rules.json": json.dumps(ext_rules(sorted(set(free_domains) | set(BUILTIN_FREE)), list(ENTERTAINMENT))), } for name, content in files.items(): with open(os.path.join(path, name), "w", encoding="utf-8") as fh: fh.write(content) def is_noise(host): return host_allowed(host, BROWSER_NOISE) or bool(re.match(r"^clients\d*\.google\.com$", host)) def host_allowed(host, domains): host = (host or "").lower().rstrip(".") return any(host == d or host.endswith("." + d) for d in domains) def version_tuple(v): try: return tuple(int(x) for x in str(v).split(".")) except ValueError: return (0,) # ─────────────────────────────── Historique pour les parents ─────────────────────────────── ACTIVITY_EVERY = 300 # Historiques des navigateurs du compte de l'enfant (lus par l'agent ; seul le NOM du site est gardé). CHROME_HISTORY = ( ".config/google-chrome/*/History", ".config/chromium/*/History", ".config/BraveSoftware/Brave-Browser/*/History", ".config/microsoft-edge/*/History", ".config/vivaldi/*/History", "snap/chromium/common/chromium/*/History", ".var/app/*/config/*/*/History", ".local/share/%s/*/History" % FREE_PROFILE_NAME, "snap/*/common/%s/*/History" % FREE_PROFILE_NAME, ".var/app/*/%s/*/History" % FREE_PROFILE_NAME, ) FIREFOX_HISTORY = (".mozilla/firefox/*/places.sqlite", "snap/firefox/common/.mozilla/firefox/*/places.sqlite", ".var/app/org.mozilla.firefox/.mozilla/firefox/*/places.sqlite") CHROME_EPOCH = 11644473600 # secondes entre 1601 et 1970 (horloge de Chrome) def site_name(url): """Nom du site d'une adresse (« https://www.YouTube.com/watch?v=… » → « youtube.com ») ; None hors pages web.""" m = re.match(r"^https?://([^/:?#]+)", url or "", re.I) if not m: return None host = m.group(1).lower().rstrip(".") host = host[4:] if host.startswith("www.") else host return host if _DOMAIN.match(host) else None def history_files(home): files = [] for pattern in CHROME_HISTORY: files += [("chrome", f) for f in glob.glob(os.path.join(home, pattern))] for pattern in FIREFOX_HISTORY: files += [("firefox", f) for f in glob.glob(os.path.join(home, pattern))] return files def read_visits(kind, path, since_ts): """Visites postérieures à `since_ts` (secondes) : [(horodatage, adresse)]. Copie le fichier (le navigateur le verrouille).""" import sqlite3 import tempfile tmp = tempfile.mktemp(prefix="limitecran-hist-") try: shutil.copyfile(path, tmp) for ext in ("-wal",): if os.path.exists(path + ext): shutil.copyfile(path + ext, tmp + ext) con = sqlite3.connect("file:%s?mode=ro" % tmp, uri=True) try: if kind == "safari": rows = con.execute( "select v.visit_time, i.url from history_visits v join history_items i on i.id = v.history_item where v.visit_time > ? order by v.visit_time", (since_ts - SAFARI_EPOCH,), ).fetchall() return [(t + SAFARI_EPOCH, u) for t, u in rows] if kind == "chrome": rows = con.execute( "select v.visit_time, u.url from visits v join urls u on u.id = v.url where v.visit_time > ? order by v.visit_time", (int((since_ts + CHROME_EPOCH) * 1_000_000),), ).fetchall() return [(t / 1_000_000 - CHROME_EPOCH, u) for t, u in rows] rows = con.execute( "select v.visit_date, p.url from moz_historyvisits v join moz_places p on p.id = v.place_id where v.visit_date > ? order by v.visit_date", (int(since_ts * 1_000_000),), ).fetchall() return [(t / 1_000_000, u) for t, u in rows] finally: con.close() except Exception: # noqa: BLE001 - historique illisible (navigateur ouvert, format inconnu) : réessayé plus tard return [] finally: for f in (tmp, tmp + "-wal"): try: os.remove(f) except OSError: pass def collect_sites(home, watermarks, now_ts): """Sites visités depuis le dernier relevé, par jour : {(date, site): visites}. Met à jour `watermarks` (par fichier).""" out = {} for kind, path in history_files(home): since = watermarks.get(path, now_ts - 3600) # premier relevé : la dernière heure seulement latest = since for ts, url in read_visits(kind, path, since): latest = max(latest, ts) name = site_name(url) if name: key = (paris_now(ts).date().isoformat(), name) out[key] = out.get(key, 0) + 1 watermarks[path] = latest return out # ─────────────────────────────── Système ─────────────────────────────── def run(cmd, timeout=15, **kw): try: return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, **kw) except (OSError, subprocess.TimeoutExpired) as e: return subprocess.CompletedProcess(cmd, 127, "", str(e)) def read_json(path, default=None): try: with open(path, encoding="utf-8") as fh: return json.load(fh) except (OSError, ValueError): return default def write_file(path, content, mode=0o644): os.makedirs(os.path.dirname(path), exist_ok=True) tmp = path + ".limitecran-tmp" with open(tmp, "w", encoding="utf-8") as fh: fh.write(content) os.chmod(tmp, mode) try: os.replace(tmp, path) except OSError as e: # fichier monté à part (conteneurs, certains systèmes pour /etc/hosts) : réécrit sur place if e.errno not in (errno.EBUSY, errno.EXDEV, errno.EPERM): raise with open(path, "w", encoding="utf-8") as fh: fh.write(content) os.remove(tmp) def write_json(path, data, mode=0o600): write_file(path, json.dumps(data, ensure_ascii=False, indent=2) + "\n", mode) def sha(text): return hashlib.sha256(text.encode("utf-8")).hexdigest() def read_text(path): try: with open(path, encoding="utf-8", errors="replace") as fh: return fh.read() except OSError: return None class Child: def __init__(self, user): pw = pwd.getpwnam(user) self.user, self.uid, self.home = user, pw.pw_uid, pw.pw_dir def is_admin(self): groups = {g.gr_name for g in grp.getgrall() if self.user in g.gr_mem} try: groups.add(grp.getgrgid(pwd.getpwnam(self.user).pw_gid).gr_name) except KeyError: pass return self.uid == 0 or bool(groups & set(ADMIN_GROUPS)) def sessions(self): """Sessions de l'enfant : [{id, active, locked, idle, type}] (loginctl).""" r = run(["loginctl", "show-user", self.user, "-p", "Sessions", "--value"]) out = [] for sid in r.stdout.split(): s = run(["loginctl", "show-session", sid, "-p", "Active", "-p", "LockedHint", "-p", "IdleHint", "-p", "Type", "-p", "Class", "-p", "State"]) props = dict(line.split("=", 1) for line in s.stdout.splitlines() if "=" in line) if props.get("Class", "user") != "user" or props.get("State") == "closing": continue out.append({ "id": sid, "active": props.get("Active") == "yes", "locked": props.get("LockedHint") == "yes", "idle": props.get("IdleHint") == "yes", "type": props.get("Type", ""), }) return out def in_use(self, sessions): return any(s["active"] and not s["locked"] and not s["idle"] for s in sessions) def notify(self, title, body, urgent=False): bus = "unix:path=/run/user/%d/bus" % self.uid if not os.path.exists("/run/user/%d/bus" % self.uid) or not shutil.which("notify-send"): return cmd = ["runuser", "-u", self.user, "--", "env", "DBUS_SESSION_BUS_ADDRESS=" + bus, "notify-send", "-a", "Limit'Écran", "-i", "appointment-soon"] if urgent: cmd += ["-u", "critical"] run(cmd + [title, body], timeout=10) def desktop_dir(self): """Dossier « Bureau » de l'enfant (xdg-user-dir), None s'il n'en a pas.""" r = run(["runuser", "-u", self.user, "--", "xdg-user-dir", "DESKTOP"], timeout=10) d = r.stdout.strip() if r.returncode == 0 else "" if d and os.path.isdir(d) and os.path.realpath(d) != os.path.realpath(self.home): return d for name in ("Bureau", "Desktop"): if os.path.isdir(os.path.join(self.home, name)): return os.path.join(self.home, name) return None def install_launcher(self): """Icône « Limit'Écran » dans le menu des applications et sur le bureau de l'enfant (remise si elle a disparu).""" for path, content in ((ICON_FILE, ICON_SVG), (LAUNCHER_FILE, LAUNCHER)): if read_text(path) != content: write_file(path, content) for old in OLD_LAUNCHER_FILES: # ancien emplacement (0.2 à 0.3.7) : retiré, sinon l'icône apparaît deux fois if read_text(old) in (LAUNCHER, ICON_SVG): try: os.remove(old) except OSError: pass desk = self.desktop_dir() if not desk: return target = os.path.join(desk, "limitecran.desktop") if read_text(target) != LAUNCHER: write_file(target, LAUNCHER, mode=0o755) os.chown(target, self.uid, pwd.getpwnam(self.user).pw_gid) # GNOME : un lanceur sur le bureau doit être marqué « de confiance » pour s'ouvrir d'un clic bus = "/run/user/%d/bus" % self.uid if os.path.exists(bus) and shutil.which("gio"): run(["runuser", "-u", self.user, "--", "env", "DBUS_SESSION_BUS_ADDRESS=unix:path=" + bus, "gio", "set", target, "metadata::trusted", "true"], timeout=10) def processes(self): """Processus de l'enfant : [{pid, ppid, exe, argv}].""" out = [] for pid in os.listdir("/proc"): if not pid.isdigit(): continue try: if os.stat("/proc/" + pid).st_uid != self.uid: continue with open("/proc/%s/cmdline" % pid, "rb") as fh: argv = split_cmdline([a.decode("utf-8", "replace") for a in fh.read().split(b"\0") if a]) with open("/proc/%s/stat" % pid) as fh: ppid = int(fh.read().rsplit(")", 1)[1].split()[1]) try: exe = os.readlink("/proc/%s/exe" % pid) except OSError: exe = "" out.append({"pid": int(pid), "ppid": ppid, "exe": exe, "argv": argv}) except (OSError, ValueError, IndexError): continue return out def matching(self, names, procs=None, exclude=()): """Processus de l'enfant appartenant aux applications `names` (hors navigateur des sites libres et Limit'Écran).""" if not names: return [] procs = self.processes() if procs is None else procs return [p for p in procs if p["pid"] not in exclude and not any("limitecran-agent" in a for a in p["argv"][:2]) and process_matches(names, p["exe"], p["argv"])] def kill_apps(self, names, procs=None, exclude=()): """Ferme les processus de l'enfant qui appartiennent à ces applications. Renvoie le nombre fermé.""" killed = 0 for p in self.matching(names, procs, exclude): try: os.kill(p["pid"], signal.SIGKILL) killed += 1 # journal (journalctl -u limitecran-agent) : ce qui a été fermé, pour comprendre un blocage inattendu log("fermé :", p["pid"], p.get("exe") or "?", " ".join(p["argv"][:3])[:200], remote=False) log("fermé :", os.path.basename(p.get("exe") or (p["argv"][0] if p["argv"] else "?")), "(%s)" % next((x.split("=", 1)[1] for x in p["argv"] if x.startswith("--type=")), "principal")) except OSError: pass return killed class ApiError(Exception): def __init__(self, status, code, message=""): super().__init__("%s %s %s" % (status, code, message)) self.status, self.code = status, code class Api: def __init__(self, server, token=None): self.server, self.token = server.rstrip("/"), token def call(self, method, path, body=None, timeout=15): data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request(self.server + path, data=data, method=method) req.add_header("content-type", "application/json") req.add_header("user-agent", "LimitEcran-Linux/" + VERSION) if self.token: req.add_header("authorization", "Bearer " + self.token) try: with urllib.request.urlopen(req, timeout=timeout) as r: raw = r.read() return json.loads(raw) if raw else None except urllib.error.HTTPError as e: try: err = json.loads(e.read()).get("error", {}) except (ValueError, AttributeError): err = {} raise ApiError(e.code, err.get("code", "http_%d" % e.code), err.get("message", "")) from None def resolve_ipv4(name): try: infos = socket.getaddrinfo(name, 443, socket.AF_INET, socket.SOCK_STREAM) return infos[0][4][0] if infos else None except OSError: return None class WebFilterApplier: """Écrit les fichiers du filtrage et vérifie qu'ils n'ont pas été modifiés (sinon : rétablis et signalés).""" def __init__(self): self.applied = read_json(APPLIED_FILE, {}) or {} self.safe_ips = {} def _save(self): write_json(APPLIED_FILE, self.applied) def desired(self, f): files = {} active = filter_active(f) if f["safeSearch"]: for target in SAFE_SEARCH: ip = resolve_ipv4(target) or self.safe_ips.get(target) if ip: self.safe_ips[target] = ip hosts = read_text(HOSTS_FILE) or "127.0.0.1 localhost\n" files[HOSTS_FILE] = merge_hosts(hosts, hosts_section(f, self.safe_ips)) chromium = json.dumps(chromium_policy(f), indent=2) + "\n" if active else None for d in CHROMIUM_POLICY_DIRS: files[d + "/limitecran.json"] = chromium ff_paths = list(FIREFOX_POLICY_FILES) + [d + "/distribution/policies.json" for d in FIREFOX_INSTALL_DIRS if os.path.isdir(d)] for p in ff_paths: files[p] = self._firefox_file(p, f) if active else self._firefox_restore(p) return files def _firefox_file(self, path, f): """Nos règles Firefox, ajoutées à celles qui existaient déjà (gardées de côté, rétablies à la désinstallation).""" backup = path + ".avant-limitecran" current = read_text(path) if current is not None and self.applied.get(path) != sha(current) and not os.path.exists(backup): write_file(backup, current) # fichier de l'administrateur, pas le nôtre base = read_json(backup, {}) if os.path.exists(backup) else {} policies = dict((base or {}).get("policies") or {}) policies.update(firefox_policies(f)) return json.dumps({"policies": policies}, indent=2, ensure_ascii=False) + "\n" def _firefox_restore(self, path): backup = path + ".avant-limitecran" return read_text(backup) if os.path.exists(backup) else None def apply(self, f, report_tamper): """Applique le filtrage. Renvoie True si tout est en place (sinon le parent est prévenu par le signe de vie).""" ok = True tampered = False for path, content in self.desired(f).items(): current = read_text(path) known = self.applied.get(path) if path == HOSTS_FILE: mine, theirs = hosts_part(content), hosts_part(current) if known and sha(theirs) != known: tampered = True if mine != theirs: try: write_file(path, content) except OSError as e: log("écriture impossible", path, e) ok = False continue if mine: self.applied[path] = sha(mine) else: self.applied.pop(path, None) continue if content is None: if known and current is not None and sha(current) == known: os.remove(path) if known: self.applied.pop(path, None) continue if current == content: self.applied[path] = sha(content) continue if known and current is not None and sha(current) != known: tampered = True try: write_file(path, content) self.applied[path] = sha(content) except OSError as e: log("écriture impossible", path, e) ok = False ok = self._apply_nft(f, report_tamper) and ok self._save() if tampered: log("filtrage modifié par quelqu'un : rétabli") report_tamper("web_filter") return ok def _apply_nft(self, f, report_tamper): rules = nft_ruleset(f) if not shutil.which("nft"): return not rules listed = run(["nft", "list", "table", "inet", NFT_TABLE]) present = listed.returncode == 0 if not rules: if present: run(["nft", "delete", "table", "inet", NFT_TABLE]) run(["resolvectl", "flush-caches"]) self.applied.pop("nft", None) return True boot = (read_text("/proc/sys/kernel/random/boot_id") or "").strip() same = self.applied.get("nft") == sha(rules) and self.applied.get("nftBoot") == boot if present and same and "dnat" in listed.stdout: return True if same and not present: report_tamper("web_filter") # table retirée pendant ce démarrage : quelqu'un l'a supprimée script = ("delete table inet %s\n" % NFT_TABLE if present else "") + rules r = run(["nft", "-f", "-"], input=script) if r.returncode != 0: log("nftables refuse les règles :", r.stderr.strip()[:300]) return False self.applied["nft"] = sha(rules) self.applied["nftBoot"] = boot run(["resolvectl", "flush-caches"]) return True def remove_all(self): self.apply({"adult": False, "safeSearch": False, "blocked": []}, lambda _k: None) # ─────────────────────────────── Agent ─────────────────────────────── class Agent: def __init__(self, conf): self.conf = conf self.api = Api(conf["server"], conf["token"]) self.child = Child(conf["user"]) self.filter = WebFilterApplier() cached = read_json(CACHE_FILE) or {} self.state = cached.get("state") self.offline = cached.get("offline", 0) # secondes d'écran comptées sans réseau, pas encore envoyées self.remaining = cached.get("remaining", 0) self.online = False self.session_open = False self.blocked_since = None self.warned = set() self.last = {"state": 0, "heartbeat": 0, "filter": 0, "inventory": cached.get("inventoryAt", 0)} self.apps = {} self.filter_ok = True self.running = True self.lock = threading.RLock() # la boucle et la fenêtre de l'enfant partagent l'état self.counting_apps = [] self.activity = {} # (date, kind, nom) -> {seconds, visits} pas encore envoyé self.history_marks = (read_json(CACHE_FILE) or {}).get("historyMarks") or {} self.refused = {} # sites refusés par le navigateur des sites libres : {hôte: date} self.refused_sent = 0 self.last["launcher"] = 0 # — échanges avec le serveur — def _call(self, method, path, body=None): try: r = self.api.call(method, path, body) self.online = True return r except ApiError as e: self.online = True if e.status == 401: self.revoked() raise except (urllib.error.URLError, OSError, ValueError) as e: self.online = False raise ConnectionError(str(e)) from None def revoked(self): log("appareil retiré par le parent : protection levée, l'agent s'arrête") self.filter.remove_all() try: os.remove(CONF_FILE) except OSError: pass service_disable() sys.exit(0) def report_tamper(self, kind): try: self._call("POST", "/api/device/tamper", {"kind": kind}) except Exception as e: # noqa: BLE001 - l'alerte est perdue, le signe de vie le dira log("alerte non envoyée :", e) def refresh_state(self): try: st = self._call("GET", "/api/device/state") except (ApiError, ConnectionError) as e: log("état non lu :", e) return self.state = st self.remaining = max(0, int(st.get("screenSeconds") or 0)) self.last["state"] = time.time() self.save_cache() def save_cache(self): write_json(CACHE_FILE, {"state": self.state, "offline": self.offline, "remaining": self.remaining, "inventoryAt": self.last["inventory"], "historyMarks": getattr(self, "history_marks", {})}) def screen(self, action): body = {"packageName": SESSION_PACKAGE} if action == "start" else {} if self.offline: body["offlineSeconds"] = min(MAX_OFFLINE, int(self.offline)) try: st = self._call("POST", "/api/device/screen/" + action, body) except ApiError as e: log("séance", action, "refusée :", e) return None except ConnectionError: return None self.offline = 0 if st: self.remaining = max(0, int(st.get("remainingSeconds") or 0)) self.session_open = bool(st.get("sessionOpen")) if self.state is not None: self.state["access"] = {"allowed": st.get("allowed"), "reason": st.get("reason")} self.save_cache() return st def heartbeat(self): try: self._call("POST", "/api/device/heartbeat", { "accessibilityEnabled": self.filter_ok, "deviceAdminEnabled": not self.child.is_admin(), "appVersion": "linux-" + VERSION, }) self.last["heartbeat"] = time.time() except (ApiError, ConnectionError) as e: log("signe de vie non envoyé :", e) def send_inventory(self, nightly): self.apps = scan_apps(self.child.home) payload = [{"id": k, "label": v["label"], "system": v["system"]} for k, v in sorted(self.apps.items())][:500] try: self._call("POST", "/api/device/apps", {"apps": payload, "nightly": nightly}) self.last["inventory"] = time.time() self.save_cache() except (ApiError, ConnectionError) as e: log("liste des applications non envoyée :", e) # — une boucle — def step(self, now_ts): st = self.state if st is None: self.refresh_state() return if now_ts - self.last["state"] >= STATE_EVERY: self.refresh_state() st = self.state removed = bool((st.get("protection") or {}).get("removed")) f = clean_filter(st.get("webFilter")) if not removed else clean_filter(None) if now_ts - self.last["filter"] >= FILTER_CHECK_EVERY or f != self.last.get("filterValue"): self.filter_ok = self.filter.apply(f, self.report_tamper) self.last["filter"], self.last["filterValue"] = now_ts, f if now_ts - self.last["heartbeat"] >= HEARTBEAT_EVERY: self.heartbeat() hour = paris_now(now_ts).hour if not self.apps or (now_ts - self.last["inventory"] > 20 * 3600 and 1 <= hour < 5): self.send_inventory(nightly=bool(self.apps)) sessions = self.child.sessions() using = self.child.in_use(sessions) policy = st.get("policy") or {} procs = self.child.processes() if using else [] free_pids = free_browser_pids(procs) # applications bloquées : fermées dès qu'elles s'ouvrent, même avec du temps if not removed and using: forbidden = policy.get("forbidden") or [] if self.child.kill_apps(app_names(self.apps, forbidden) | set(forbidden), procs, free_pids): self.child.notify("Application bloquée", "Un parent a bloqué cette application sur ton ordinateur.") # le temps ne s'écoule que si une application limitée est ouverte (les libres et les sites libres ne comptent pas) limited_list = limited_ids(self.apps, policy) limited = app_names(self.apps, limited_list) counting = using and bool(self.child.matching(limited, procs, free_pids)) # historique : applications ouvertes (libres comme limitées) pendant ce passage if using: day = paris_now(now_ts).date().isoformat() for i, app in self.apps.items(): if not app.get("system") and self.child.matching(app_names(self.apps, [i]), procs, free_pids): k = (day, "app", app["label"]) self.activity.setdefault(k, {"seconds": 0, "visits": 0})["seconds"] += TICK_SECONDS # pour l'enfant (et le parent) : quelles applications font compter le temps en ce moment self.counting_apps = sorted({self.apps[i]["label"] for i in limited_list if self.child.matching(app_names(self.apps, [i]), procs, free_pids)}) if counting else [] # décompte du temps if counting and not removed: if self.session_open: if self.screen("tick") is None and not self.online: self.count_offline() else: allowed, _ = effective_access(st, not self.online, paris_now(now_ts), self.remaining) if allowed and self.screen("start") is None and not self.online: self.count_offline() elif self.session_open: self.screen("stop") self.session_open = False allowed, reason = effective_access(st, not self.online, paris_now(now_ts), self.remaining) self.write_status(allowed, reason) if counting and not allowed: self.block(limited, procs, free_pids, reason, now_ts) elif allowed: self.blocked_since = None # le délai d'avertissement ne repart qu'une fois le temps revenu if counting: self.warn() def count_offline(self): self.offline = min(MAX_OFFLINE, self.offline + TICK_SECONDS) self.remaining = max(0, self.remaining - TICK_SECONDS) self.save_cache() def warn(self): for limit, text in ((300, "Plus que 5 minutes d'écran."), (60, "Plus qu'une minute d'écran.")): if self.remaining <= limit and limit not in self.warned and self.remaining > 0: self.warned.add(limit) self.child.notify("Limit'Écran", text, urgent=limit == 60) if self.remaining > 300: self.warned.clear() def block(self, limited, procs, free_pids, reason, now_ts): """Comme sur le téléphone : les applications limitées se ferment (après un avertissement), les libres restent.""" if self.blocked_since is None: self.blocked_since = now_ts log("blocage :", reason, "— applications limitées ouvertes :", ", ".join(getattr(self, "counting_apps", [])) or "?", "— navigateur des sites libres :", len(free_pids), "processus") apps = getattr(self, "counting_apps", []) or ["les applications limitées"] self.child.notify( "Pause écran", REASON_TEXT.get(reason or "", "L'écran est en pause.") + " %s va se fermer dans 30 secondes. Pronote et les autres sites libres restent ouverts si tu les ouvres avec le bouton « Ouvrir » de ta fenêtre Limit'Écran." % ", ".join(apps[:3]), urgent=True, ) return if now_ts - self.blocked_since >= BLOCK_GRACE_SECONDS and self.child.kill_apps(limited, procs, free_pids): if now_ts - self.last.get("killNotice", 0) > 60: self.last["killNotice"] = now_ts self.child.notify("Application fermée", REASON_TEXT.get(reason or "", "L'écran est en pause.")) def write_status(self, allowed, reason): st = self.state or {} status = { "child": (st.get("child") or {}).get("firstName"), "allowed": allowed, "reason": reason, "remainingSeconds": self.remaining, "balance": st.get("balance"), "online": self.online, "countingApps": getattr(self, "counting_apps", []), "filter": clean_filter(st.get("webFilter")), "updatedAt": int(time.time()), } try: write_json(STATUS_FILE, status, mode=0o644) except OSError: pass # — navigateur des sites libres — def serve_free_proxy(self): """ Filtre local (127.0.0.1) du navigateur des sites libres : seuls les sites que le parent a déclarés libres passent. HTTPS (CONNECT) relayé tel quel ; tout le reste est refusé avec une page claire. """ srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srv.bind(("127.0.0.1", FREE_PROXY_PORT)) srv.listen(64) def refuse(conn, host): body = ("%s ne fait pas partie des sites libres choisis par tes parents. " "Ouvre-le dans le navigateur habituel : il compte alors dans ton temps d’écran.
" % (host or "Ce site")).encode("utf-8") conn.sendall(b"HTTP/1.1 403 Forbidden\r\nContent-Type: text/html; charset=utf-8\r\nContent-Length: %d\r\nConnection: close\r\n\r\n" % len(body) + body) def relay(a, b): socks = [a, b] while True: ready, _, _ = select.select(socks, [], [], 300) if not ready: return for r in ready: data = r.recv(65536) if not data: return (b if r is a else a).sendall(data) def handle(conn): with conn: try: conn.settimeout(30) head = b"" while b"\r\n\r\n" not in head and len(head) < 16384: chunk = conn.recv(4096) if not chunk: return head += chunk parts = head.split(b"\r\n", 1)[0].decode("latin-1").split() if len(parts) < 2: return method, target = parts[0].upper(), parts[1] free = clean_filter((self.state or {}).get("webFilter"))["free"] + list(BUILTIN_FREE) target_host = target.rpartition(":")[0].lower() if method == "CONNECT" and target_host == EXT_HEARTBEAT_HOST: self.ext_seen = time.time() # l'extension des sites libres est active conn.sendall(b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") return # Extension active : elle contrôle les pages ouvertes ; tout ce dont elles ont besoin passe ici. # Sans elle (coupée, ou navigateur qui la refuse) : seuls les sites libres passent. if method == "CONNECT" and time.time() - getattr(self, "ext_seen", 0) < EXT_ALIVE_SECONDS: free = [target_host] if method != "CONNECT": host = re.sub(r"^[a-z]+://", "", target).split("/")[0].split(":")[0] if not is_noise(host.lower()): self.refused[host.lower()] = time.time() refuse(conn, host) return host, _, port = target.rpartition(":") if port != "443" or not host_allowed(host, free): if not is_noise(host.lower()): self.refused[host.lower()] = time.time() refuse(conn, host) return upstream = socket.create_connection((host, 443), timeout=30) with upstream: conn.sendall(b"HTTP/1.1 200 Connection established\r\n\r\n") conn.settimeout(None) upstream.settimeout(None) relay(conn, upstream) except (OSError, ValueError): return while self.running: try: conn, _ = srv.accept() except OSError: continue threading.Thread(target=handle, args=(conn,), daemon=True).start() # — historique pour les parents — def send_activity(self, now_ts): try: for (day, site), n in collect_sites(self.child.home, self.history_marks, now_ts).items(): self.activity.setdefault((day, "site", site), {"seconds": 0, "visits": 0})["visits"] += n except Exception as e: # noqa: BLE001 log("historique des sites illisible :", repr(e)) if not self.activity: return items = [{"date": d, "kind": k, "name": n, **v} for (d, k, n), v in list(self.activity.items())[:500]] try: self.api.call("POST", "/api/device/activity", {"items": items}) for it in items: self.activity.pop((it["date"], it["kind"], it["name"]), None) cache = read_json(CACHE_FILE) or {} cache["historyMarks"] = self.history_marks write_json(CACHE_FILE, cache) except Exception as e: # noqa: BLE001 - gardé pour le prochain envoi print("[limitecran] historique non envoyé :", e, flush=True) # — mise à jour automatique — def check_update(self): """ Une fois par jour : nouvelle version publiée sur le site ? Téléchargée, empreinte SHA-256 vérifiée, code relu par Python, puis l'agent s'arrête et systemd le relance sur la nouvelle version. """ base = self.conf["server"].rstrip("/") + "/linux/" try: with urllib.request.urlopen(base + "version.json", timeout=30) as r: info = json.loads(r.read()) if version_tuple(info.get("version")) <= version_tuple(VERSION): return False with urllib.request.urlopen(base + "limitecran_agent.py", timeout=60) as r: code = r.read() if hashlib.sha256(code).hexdigest() != info.get("sha256"): log("mise à jour refusée : empreinte différente") return False compile(code, AGENT_BIN, "exec") tmp = AGENT_BIN + ".nouveau" with open(tmp, "wb") as fh: fh.write(code) os.chmod(tmp, 0o755) os.replace(tmp, AGENT_BIN) log("mise à jour installée :", VERSION, "→", info.get("version"), "; redémarrage") return True except (OSError, ValueError, SyntaxError) as e: log("mise à jour impossible pour l'instant :", e) return False # — fenêtre de l'enfant — def child_request(self, req): """Une demande de la fenêtre de l'enfant. Renvoie la vue à jour, ou {"error": message lisible}.""" op = req.get("op") with self.lock: if op == "state": if time.time() - self.last["state"] > 5: self.refresh_state() return child_view(self.state, self.online, self.remaining, self.refused, getattr(self, "counting_apps", [])) st = self.state or {} try: if op == "complete": task = str(req.get("taskId", "")) if not any(t.get("id") == task for t in st.get("tasks") or []): return {"error": "Tâche inconnue."} self._call("POST", "/api/device/tasks/%s/complete" % task, {"clientOpId": str(uuid.uuid4()), "forDate": (st.get("day") or {}).get("date")}) elif op == "buy": reward = str(req.get("rewardId", "")) if not any(r.get("id") == reward for r in st.get("rewards") or []): return {"error": "Récompense inconnue."} self._call("POST", "/api/device/rewards/%s/redeem" % reward, {"clientOpId": str(uuid.uuid4())}) elif op == "request_time": minutes = req.get("minutes") if minutes not in TIME_REQUEST_MINUTES: return {"error": "Choisis 10, 20 ou 30 minutes."} self._call("POST", "/api/device/time-requests", {"minutes": minutes}) else: return {"error": "Demande inconnue."} except ApiError as e: return {"error": str(e.args[0]).split(" ", 2)[-1] or "Le serveur a refusé."} except ConnectionError: return {"error": "Il faut Internet pour faire ça. Réessaie quand l'ordinateur est connecté."} self.refresh_state() return child_view(self.state, self.online, self.remaining, self.refused, getattr(self, "counting_apps", [])) def serve_child(self): """Canal local de la fenêtre : seul le compte de l'enfant (ou root) peut s'y connecter (vérifié par le noyau).""" os.makedirs(RUN_DIR, exist_ok=True) os.chmod(RUN_DIR, 0o755) try: os.remove(CHILD_SOCKET) except OSError: pass srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) srv.bind(CHILD_SOCKET) os.chmod(CHILD_SOCKET, 0o666) # le contrôle se fait sur l'identité du client, ci-dessous srv.listen(8) def handle(conn): with conn: try: uid = peer_uid(conn) if uid not in (self.child.uid, 0): conn.sendall(json.dumps({"error": "Réservé au compte de l’enfant."}, ensure_ascii=False).encode("utf-8") + b"\n") return conn.settimeout(30) data = b"" while not data.endswith(b"\n") and len(data) < 8192: chunk = conn.recv(4096) if not chunk: break data += chunk reply = self.child_request(json.loads(data.decode("utf-8") or "{}")) except Exception as e: # noqa: BLE001 log("fenêtre :", repr(e)) reply = {"error": "Erreur inattendue, réessaie."} conn.sendall(json.dumps(reply, ensure_ascii=False).encode("utf-8") + b"\n") while self.running: try: conn, _ = srv.accept() except OSError: continue threading.Thread(target=handle, args=(conn,), daemon=True).start() def loop(self): signal.signal(signal.SIGTERM, lambda *_: setattr(self, "running", False)) log("démarrage, version", VERSION, "— compte surveillé :", self.child.user) threading.Thread(target=self.serve_child, daemon=True).start() threading.Thread(target=self.serve_free_proxy, daemon=True).start() self.last["update"] = time.time() - UPDATE_EVERY + 600 # première vérification 10 min après le démarrage if self.child.is_admin(): self.report_tamper("device_admin") while self.running: started = time.time() try: with self.lock: self.step(started) # historique des parents : sites visités et applications, envoyés toutes les 5 minutes if started - self.last.get("activity", 0) > ACTIVITY_EVERY: self.last["activity"] = started self.send_activity(started) # journal de diagnostic : envoyé toutes les 5 minutes (gardé s'il n'est pas passé) if REMOTE_LOG and started - self.last.get("logs", 0) > LOGS_EVERY: self.last["logs"] = started batch = REMOTE_LOG[:100] try: self.api.call("POST", "/api/device/logs", {"entries": batch}) del REMOTE_LOG[: len(batch)] except Exception as e: # noqa: BLE001 - réessayé au prochain passage print("[limitecran] journal non envoyé :", e, flush=True) # sites refusés : envoyés au parent (onglet Ordinateur), au plus une fois par minute fresh = [h for h, t in list(self.refused.items()) if t > self.refused_sent] if fresh and started - self.refused_sent > 60: self.refused_sent = started try: self._call("POST", "/api/device/free-refused", {"hosts": fresh[:50]}) except (ApiError, ConnectionError) as e: log("sites refusés non envoyés :", e) if started - self.last["launcher"] > 3600: self.last["launcher"] = started self.child.install_launcher() if started - self.last["update"] > UPDATE_EVERY and os.path.abspath(sys.argv[0]) == AGENT_BIN: self.last["update"] = started if self.check_update(): self.running = False except SystemExit: raise except Exception as e: # noqa: BLE001 - une erreur ne doit jamais arrêter la surveillance log("erreur :", repr(e)) time.sleep(max(1.0, TICK_SECONDS - (time.time() - started))) if self.session_open: self.screen("stop") # ─────────────────────────────── Fenêtre de l'enfant (Tkinter) ─────────────────────────────── def ask_agent(req, path=CHILD_SOCKET, timeout=30): """Envoie une demande à l'agent (service root) par le canal local ; {"error": …} si l'agent ne répond pas.""" try: with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as c: c.settimeout(timeout) c.connect(path) c.sendall(json.dumps(req).encode("utf-8") + b"\n") data = b"" while not data.endswith(b"\n"): chunk = c.recv(65536) if not chunk: break data += chunk return json.loads(data.decode("utf-8")) except (OSError, ValueError): return {"error": "Limit'Écran ne répond pas. Redémarre l'ordinateur ou préviens un parent."} def fmt_minutes(seconds): m = max(0, int(seconds)) // 60 return "%d h %02d" % (m // 60, m % 60) if m >= 60 else "%d min" % m def limis(n): return "%d Limi%s" % (n, "s" if abs(n) > 1 else "") # Google Chrome en dernier : il refuse l'extension des sites libres (le filtre reste alors strict). BROWSERS = ("chromium", "chromium-browser", "chromium-freeworld", "brave-browser", "brave", "microsoft-edge", "microsoft-edge-stable", "vivaldi", "vivaldi-stable", "google-chrome", "google-chrome-stable") # Versions Flatpak (Fedora, Linux Mint, Silverblue…) : lancées par leur raccourci exporté. FLATPAK_BROWSERS = ("org.chromium.Chromium", "com.google.Chrome", "com.brave.Browser", "com.microsoft.Edge", "com.vivaldi.Vivaldi") def find_free_browser(home, which=shutil.which, exists=os.path.exists): """(commande, dossier du profil) du premier navigateur de la famille Chrome trouvé : paquet, snap ou Flatpak.""" for name in BROWSERS: path = which(name) if not path: continue if path.startswith("/snap/") or os.path.realpath(path).endswith("/snap"): return path, os.path.join(home, "snap", name, "common", FREE_PROFILE_NAME) # un navigateur snap n'écrit que là return path, os.path.join(home, ".local", "share", FREE_PROFILE_NAME) for app in FLATPAK_BROWSERS: for exports in (os.path.join(home, ".local/share/flatpak/exports/bin"), "/var/lib/flatpak/exports/bin"): path = os.path.join(exports, app) if exists(path): return path, os.path.join(home, ".var", "app", app, FREE_PROFILE_NAME) # dossier accessible dans le bac à sable return None def free_browser_cmd(url, home=None, which=shutil.which, exists=os.path.exists, extension=None): """ Commande du navigateur des sites libres (famille Chrome) : profil à part + filtre local, et l'extension Limit'Écran (`extension` = son dossier) qui ne laisse ouvrir en page principale que les sites libres et ce vers quoi ils mènent. None sans navigateur. """ home = home or os.path.expanduser("~") found = find_free_browser(home, which, exists) if found: path, data = found ext = ["--load-extension=" + extension, "--disable-extensions-except=" + extension] if extension else [] return [ path, "--user-data-dir=" + data, *ext, "--proxy-server=http://127.0.0.1:%d" % FREE_PROXY_PORT, "--no-first-run", "--no-default-browser-check", # services internes de Chrome (mises à jour, synchronisation, remplissage, optimisations) : inutiles ici "--disable-background-networking", "--disable-sync", "--disable-component-update", "--disable-features=OptimizationHints,AutofillServerCommunication,MediaRouter", "--new-window", url, ] return None TASK_STATUS = {"en_attente": "En attente d'un parent", "validee": "Validée", "refusee": "Refusée : refais-la"} def cmd_fenetre(_args): try: import tkinter as tk from tkinter import messagebox, ttk except ImportError: sys.exit("La fenêtre a besoin de Tkinter (paquet python3-tk).") NAVY, AMBER, CREAM, MINT, SLATE, WHITE = "#1F3A5F", "#F5A524", "#FFFBF5", "#127A6B", "#5B6B7F", "#FFFFFF" root = tk.Tk() root.title("Limit'Écran") root.geometry("500x660") root.minsize(380, 480) root.configure(bg=CREAM) base = ("TkDefaultFont", 11) bold = ("TkDefaultFont", 11, "bold") style = ttk.Style(root) try: style.theme_use("clam") except tk.TclError: pass style.configure("TNotebook", background=CREAM, borderwidth=0) style.configure("TNotebook.Tab", padding=(8, 6), font=bold) style.configure("Go.TButton", font=bold, padding=(10, 6), foreground=WHITE, background=MINT) style.map("Go.TButton", background=[("disabled", "#B9C4CF"), ("active", "#0E6357")]) style.configure("Plain.TButton", font=bold, padding=(10, 6)) def coin(parent, bg, size=22): """Le logo Limi : pièce ambrée marquée « L », dessinée par Tk.""" c = tk.Canvas(parent, width=size, height=size, bg=bg, highlightthickness=0, bd=0) c.create_oval(1, 2, size - 1, size, fill="#C97F0A", outline="") c.create_oval(1, 1, size - 1, size - 2, fill=AMBER, outline="#C97F0A") c.create_text(size / 2, size / 2 - 1, text="L", fill="#5A3A0A", font=("TkDefaultFont", max(7, int(size * 0.45)), "bold")) return c head = tk.Frame(root, bg=NAVY, padx=16, pady=14) head.pack(fill="x") hello = tk.Label(head, text="", bg=NAVY, fg=WHITE, font=("TkDefaultFont", 16, "bold"), anchor="w") hello.pack(fill="x") numbers = tk.Frame(head, bg=NAVY) numbers.pack(fill="x", pady=(8, 0)) time_lbl = tk.Label(numbers, text="", bg=NAVY, fg=WHITE, font=("TkDefaultFont", 22, "bold")) time_lbl.pack(side="left") tk.Label(numbers, text=" d'écran", bg=NAVY, fg="#C9D4E2", font=base).pack(side="left", pady=(8, 0)) limi_box = tk.Frame(numbers, bg=AMBER, padx=8, pady=3) limi_box.pack(side="right") limi_lbl = tk.Label(limi_box, text="", bg=AMBER, fg="#5A3A0A", font=("TkDefaultFont", 13, "bold")) limi_lbl.pack(side="left") coin(limi_box, AMBER, 20).pack(side="left", padx=(4, 0)) status = tk.Label(head, text="", bg=NAVY, fg="#FFD98A", font=base, anchor="w", justify="left", wraplength=460) status.pack(fill="x", pady=(8, 0)) tabs = ttk.Notebook(root) tabs.pack(fill="both", expand=True, padx=8, pady=8) message = tk.Label(root, text="", bg=CREAM, fg=SLATE, font=base, wraplength=420, justify="left") message.pack(fill="x", padx=12, pady=(0, 2)) # information de l'enfant (RGPD), discrète tk.Label(root, text="Tes parents voient les applications et les sites que tu utilises.", bg=CREAM, fg="#9AA5B1", font=("TkDefaultFont", 8), anchor="w").pack(fill="x", padx=12, pady=(0, 6)) def scroll_tab(title): outer = tk.Frame(tabs, bg=CREAM) canvas = tk.Canvas(outer, bg=CREAM, highlightthickness=0) bar = ttk.Scrollbar(outer, orient="vertical", command=canvas.yview) inner = tk.Frame(canvas, bg=CREAM) inner.bind("