#!/usr/bin/env python3 """Re-runs the check behind proton-drive-photos-privacy (checked 2026-10-26). Three parts, all read from dataset.json next to this file: 1. Pages. Every table cell and every claim names a page and the exact sentence it rests on. Each page is fetched fresh, reduced to text (entities, curly quotes and dashes normalised) and searched for the sentence. The Google One price is searched in the raw HTML too, because the visible price follows the visitor's currency while the schema.org block states USD. 2. Proton's web client. The files of github.com/ProtonMail/WebClients named in the dataset are fetched at the pinned commit and searched for the code lines the findings quote. A sparse checkout of the photos section at the same commit confirms that no file there reads a location field (git, about 5 s). 3. Proton's Drive SDK. The @protontech/drive-sdk 0.23.0 tarball is downloaded from the npm registry, its SHA-256 compared with the recorded one, and the quoted lines searched in the named files. No account was used and none is needed: Proton refused the automated sign-up on 25 October 2026 (see notes in dataset.json), so there is no upload, timing or download leg to re-run, and no share link is read. Also writes dataset.csv from dataset.json, so the two cannot drift. Exit 0 when every quote and code line still holds, 1 otherwise. --recheck does the same; no step takes longer than a minute. """ import csv import gzip import hashlib import html import io import json import os import re import subprocess import sys import tarfile import tempfile import urllib.error import urllib.parse import urllib.request from html.parser import HTMLParser HERE = os.path.dirname(os.path.abspath(__file__)) UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120 Safari/537.36") COLUMNS = [ ("free_storage", "Free storage"), ("price_entry", "Entry paid plan"), ("encryption", "Encryption"), ("ai_training", "AI training"), ("compression", "Compression"), ("share_without_account", "Share without an account"), ("location_grouping", "Location grouping"), ] class Text(HTMLParser): def __init__(self): super().__init__() self.skip = 0 self.out = [] def handle_starttag(self, tag, attrs): if tag in ("script", "style", "noscript", "svg"): self.skip += 1 if tag in ("p", "div", "li", "tr", "br", "h1", "h2", "h3", "h4", "td", "th", "section"): self.out.append(" ") def handle_endtag(self, tag): if tag in ("script", "style", "noscript", "svg") and self.skip: self.skip -= 1 def handle_data(self, data): if not self.skip: self.out.append(data) def normalise(s): s = html.unescape(s) for a, b in (("’", "'"), ("‘", "'"), ("“", '"'), ("”", '"'), ("–", "-"), ("—", "-"), (" ", " "), ("…", "...")): s = s.replace(a, b) s = s.replace("**", "") return re.sub(r"\s+", " ", s).strip() def get(url, depth=0): req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept-Encoding": "gzip"}) try: resp = urllib.request.urlopen(req, timeout=60) except urllib.error.HTTPError as err: # Python 3.9's urllib does not follow 308, which viallo.app answers with. if err.code in (301, 302, 303, 307, 308) and depth < 5: return get(urllib.parse.urljoin(url, err.headers["Location"]), depth + 1) raise raw = resp.read() if resp.headers.get("Content-Encoding") == "gzip": raw = gzip.decompress(raw) return raw _pages = {} def page(url): if url not in _pages and url.startswith("repo:"): # repo: is read from the Viallo checkout this file sits in # (apps/web/public/blog//check.py -> repository root) root = os.path.abspath(os.path.join(HERE, "..", "..", "..", "..", "..")) with open(os.path.join(root, url[5:]), encoding="utf-8") as fh: _pages[url] = normalise(fh.read()) if url not in _pages: body = get(url).decode("utf-8", "ignore") if url.startswith("https://raw.githubusercontent.com/"): _pages[url] = normalise(body) else: parser = Text() parser.feed(body) _pages[url] = normalise("".join(parser.out)) + " " + normalise(body) return _pages[url] def has(url, quote): try: return normalise(quote) in page(url), None except Exception as exc: # network error, 403, timeout return False, "fetch failed: %s" % exc def report(failures, label, url, quote): ok, err = has(url, quote) print(" %-24s %-5s %s" % (label, "ok" if ok else ("FETCH" if err else "GONE"), url)) if not ok: failures.append("%s | %s | %s" % (label, url, err or "quote gone: %r" % quote[:80])) def check_pages(data, failures): for row in data["results"]: print("\n%s" % row["option"]) for prop, cell in row["cells"].items(): for q in [{"quote": cell["quote"], "url": cell["url"]}] + cell.get("extra", []): report(failures, prop, q["url"], q["quote"]) print("\nClaims in the text") for i, claim in enumerate(data["claims"], 1): report(failures, "claim %d" % i, claim["url"], claim["quote"]) def check_code(data, failures): code = data["code"] base = "https://raw.githubusercontent.com/ProtonMail/WebClients/%s/" % code["commit"] print("\nWebClients at %s" % code["commit"][:12]) for item in code["files"]: for s in item["strings"]: report(failures, os.path.basename(item["file"]), base + item["file"], s) absent = code["absent"] with tempfile.TemporaryDirectory() as tmp: try: for cmd in (["git", "init", "-q"], ["git", "remote", "add", "origin", code["repository"] + ".git"], ["git", "fetch", "-q", "--depth", "1", "--filter=blob:none", "origin", code["commit"]], ["git", "sparse-checkout", "set", "--no-cone", "/%s/" % absent["dir"]], ["git", "checkout", "-q", "FETCH_HEAD"]): subprocess.run(cmd, cwd=tmp, check=True, capture_output=True, timeout=55) hits = [] pattern = re.compile(absent["pattern"]) for root, _, names in os.walk(os.path.join(tmp, absent["dir"])): for name in names: path = os.path.join(root, name) with open(path, encoding="utf-8", errors="ignore") as fh: for line in fh: if pattern.search(line) and "window.location" not in line: hits.append(os.path.relpath(path, tmp)) break files = sum(len(n) for _, _, n in os.walk(os.path.join(tmp, absent["dir"]))) ok = files > 0 and not hits print(" %-24s %-5s %d files in %s, %d read a location field" % ("no location reader", "ok" if ok else "FAIL", files, absent["dir"], len(hits))) if not ok: failures.append("absent | %s | %s" % (absent["dir"], hits or "checkout empty")) except Exception as exc: print(" %-24s FETCH %s" % ("no location reader", exc)) failures.append("absent | git checkout failed: %s" % exc) def check_sdk(data, failures): sdk = data["code"]["sdk"] print("\n@protontech/drive-sdk %s" % sdk["version"]) try: blob = get(sdk["url"]) except Exception as exc: failures.append("sdk | %s | fetch failed: %s" % (sdk["url"], exc)) print(" %-24s FETCH %s" % ("tarball", exc)) return digest = hashlib.sha256(blob).hexdigest() ok = digest == sdk["sha256"] print(" %-24s %-5s sha256 %s" % ("tarball", "ok" if ok else "DIFF", digest[:16])) if not ok: failures.append("sdk | sha256 %s, recorded %s" % (digest, sdk["sha256"])) with tarfile.open(fileobj=io.BytesIO(blob), mode="r:gz") as tar: for item in sdk["findings"]: text = normalise(tar.extractfile(item["file"]).read().decode("utf-8", "ignore")) for s in item["strings"]: ok = normalise(s) in text print(" %-24s %-5s %s" % (os.path.basename(item["file"]), "ok" if ok else "GONE", s[:60])) if not ok: failures.append("sdk | %s | line gone: %r" % (item["file"], s)) def write_csv(data): with open(os.path.join(HERE, "dataset.csv"), "w", newline="") as fh: writer = csv.writer(fh, lineterminator="\n") writer.writerow(["option", "checked_at"] + [label for _, label in COLUMNS]) for row in data["results"]: writer.writerow([row["option"], data["checkedAt"]] + [row["cells"][key]["value"] for key, _ in COLUMNS]) def main(): with open(os.path.join(HERE, "dataset.json")) as fh: data = json.load(fh) failures = [] check_pages(data, failures) check_code(data, failures) check_sdk(data, failures) write_csv(data) print("\n%d failures" % len(failures)) for f in failures: print("FAIL " + f) return 1 if failures else 0 if __name__ == "__main__": sys.exit(main())