#!/usr/bin/env python3 """Measure how long elevated access on English Wikipedia stayed open after its holder's last edit. The data is English Wikipedia's public user-rights log (API `list=logevents`, `letype=rights`) for the closed period 2021-01-01 to 2025-12-31, and each removed person's last edit before the removal (API `list=usercontribs`). Wikipedia's text is CC BY-SA 4.0: https://creativecommons.org/licenses/by-sa/4.0/ python3 rights_removal_lag.py --out rights_removal_lag.csv python3 rights_removal_lag.py --out planted.csv --plant duplicate-events,unmapped-group,dead-mapping-key,missing-month Needs Python 3 and the duckdb module (tested on Python 3.12, DuckDB 1.5.x). It asks the API one request at a time with a pause and `maxlag`, prints the number of log events read and a SHA-256 fingerprint of their ids (so a re-run shows whether the source changed), prints the result table and the result of its own checks. It prints counts, shares and durations only: no user name is printed, stored in the CSV or sent anywhere except back to Wikipedia's own API to look up that person's last edit. It exits 1, and writes no CSV, if a check fails. Wikipedia is a public project. Its log stands in for a company's own permission log; it is not a client result. """ import argparse import csv import datetime import email.utils import hashlib import json import os import re import sys import time import urllib.error import urllib.parse import urllib.request import duckdb API = "https://en.wikipedia.org/w/api.php" # Change the contact address to your own before you run this: MediaWiki asks every client to identify itself. USER_AGENT = "rights-removal-lag-script/1.0 (https://www.datameaning.com/guides/; info@datameaning.com)" START = "2021-01-01T00:00:00Z" END = "2025-12-31T23:59:59Z" YEARS = [2021, 2022, 2023, 2024, 2025] PAUSE = 0.3 # seconds between requests MAXLAG = 5 # MediaWiki asks clients to back off when its replicas lag more than this many seconds # Every group name that appears in the log is mapped here. "elevated" groups are administrator level # and above (delete pages, block users, change others' rights, view suppressed material); the rest are # ordinary editing tools, including the group that views IP addresses of temporary accounts. A group seen in the data that is not # listed here is counted as unmapped and fails the run. These are our choices: change them to yours. GROUPS = { "sysop": "elevated", # administrator "bureaucrat": "elevated", "interface-admin": "elevated", "checkuser": "elevated", "oversight": "elevated", "suppress": "elevated", "steward": "elevated", "autoreviewer": "ordinary", "extendedconfirmed": "ordinary", "confirmed": "ordinary", "rollbacker": "ordinary", "reviewer": "ordinary", "patroller": "ordinary", "filemover": "ordinary", "extendedmover": "ordinary", "templateeditor": "ordinary", "eventcoordinator": "ordinary", "massmessage-sender": "ordinary", "accountcreator": "ordinary", "ipblock-exempt": "ordinary", "abusefilter": "ordinary", "abusefilter-helper": "ordinary", "bot": "ordinary", "import": "ordinary", "transwiki": "ordinary", "researcher": "ordinary", "founder": "ordinary", "temporary-account-viewer": "ordinary", "electionclerk": "ordinary", "copyviobot": "ordinary", "flow-bot": "ordinary", "oathauth-twofactorauth": "ordinary", } # The reason typed in the log comment, bucketed by the first rule that matches (lower-cased text). # Order matters: a comment that says "inactivity" is inactivity even if it also says "request". REASON_RULES = [ ("inactivity", ["inactiv"]), ("resignation", ["resign", "voluntar", "per request", "requested", "retire", "no longer need"]), ("committee or security", ["arbitration", "arbcom", "committee", "motion", "security"]), ] OTHER = "other or no reason given" REASONS = [r for r, _ in REASON_RULES] + [OTHER] PLANTS = {"duplicate-events", "unmapped-group", "dead-mapping-key", "missing-month"} def wait_seconds(retry_after): """Retry-After is either seconds or an HTTP date; fall back to 5 seconds.""" try: return max(0.0, float(retry_after)) except (TypeError, ValueError): pass try: when = email.utils.parsedate_to_datetime(retry_after) return max(0.0, (when - datetime.datetime.now(datetime.timezone.utc)).total_seconds()) except (TypeError, ValueError): return 5.0 def has_keyword(comment, key): """Keyword at the start of a word, so 'motion' does not match 'demotion' or 'promotion'.""" return re.search(r"\b" + re.escape(key), comment) is not None def api(params): """One API request, politely: User-Agent, maxlag, a pause, and a wait if the server says so.""" q = dict(params, format="json", formatversion="2", maxlag=str(MAXLAG)) url = API + "?" + urllib.parse.urlencode(q) for attempt in range(6): time.sleep(PAUSE) try: with urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": USER_AGENT}), timeout=60) as r: retry_after = r.headers.get("Retry-After") data = json.load(r) except urllib.error.HTTPError as e: if e.code in (429, 500, 502, 503, 504): time.sleep(wait_seconds(e.headers.get("Retry-After"))) continue raise except (urllib.error.URLError, TimeoutError, json.JSONDecodeError): time.sleep(5) # a dropped connection, a read timeout or a cut-off body: one more attempt continue if data.get("error", {}).get("code") == "maxlag": time.sleep(wait_seconds(retry_after)) continue if "error" in data: sys.exit(f"API error: {data['error'].get('code')}") return data sys.exit("API kept asking us to wait: stopping") def fetch_events(): """Every rights-log event in the period, oldest first. Returns a list of tuples (no actor is fetched).""" events, requests, cont = [], 0, {} while True: data = api(dict(action="query", list="logevents", letype="rights", lestart=START, leend=END, ledir="newer", lelimit="500", leprop="ids|title|type|timestamp|comment|details", **cont)) requests += 1 for e in data["query"]["logevents"]: p = e.get("params") or {} hidden = bool(e.get("actionhidden") or e.get("commenthidden") or "oldgroups" not in p) events.append((e["logid"], e["timestamp"].replace("T", " ").rstrip("Z"), e.get("action", ""), e.get("title", ""), p.get("oldgroups", []), p.get("newgroups", []), (e.get("comment") or "") if not e.get("commenthidden") else "", hidden)) if "continue" not in data: return events, requests cont = {"lecontinue": data["continue"]["lecontinue"]} def last_edit(title, removed_at): """Timestamp of the person's last edit at or before removed_at, or None if there is none.""" data = api(dict(action="query", list="usercontribs", ucuser=title.split(":", 1)[1], ucstart=removed_at, ucdir="older", uclimit="1", ucprop="timestamp")) rows = data["query"]["usercontribs"] return rows[0]["timestamp"].replace("T", " ").rstrip("Z") if rows else None def main(): ap = argparse.ArgumentParser() ap.add_argument("--out", required=True, help="CSV to write") ap.add_argument("--plant", default="", help=f"comma list of test faults: {', '.join(sorted(PLANTS))}") a = ap.parse_args() out_dir = os.path.dirname(os.path.abspath(a.out)) if not os.path.isdir(out_dir): ap.error(f"--out folder {out_dir} does not exist; run from the script's folder with --out rights_removal_lag.csv") plant = [x for x in a.plant.split(",") if x] if set(plant) - PLANTS: ap.error(f"unknown --plant value(s) {sorted(set(plant) - PLANTS)}; use {sorted(PLANTS)}") print(f"duckdb {duckdb.__version__}, python {sys.version.split()[0]}") print(f"source {API} list=logevents letype=rights {START} to {END}") events, n_requests = fetch_events() ids = sorted({e[0] for e in events}) print(f"read {len(events)} log events in {n_requests} requests; " f"sha256 of the sorted ids = {hashlib.sha256(','.join(map(str, ids)).encode()).hexdigest()}") source_events = len(events) # what the API returned, before any planted fault print(f"{sum(1 for e in events if e[7])} of them are hidden or carry no group details; they are left out of the analysis") con = duckdb.connect() con.sql("CREATE TABLE ev (logid BIGINT, ts TIMESTAMP, action VARCHAR, title VARCHAR, old_groups VARCHAR[], " "new_groups VARCHAR[], comment VARCHAR, hidden BOOLEAN)") con.executemany("INSERT INTO ev VALUES (?, ?, ?, ?, ?, ?, ?, ?)", events) con.sql("CREATE TABLE grp (name VARCHAR, kind VARCHAR)") con.executemany("INSERT INTO grp VALUES (?, ?)", list(GROUPS.items())) if "duplicate-events" in plant: con.sql("INSERT INTO ev SELECT * FROM ev ORDER BY logid LIMIT 300") print("PLANTED: the first 300 log events loaded twice") if "unmapped-group" in plant: con.sql("UPDATE ev SET old_groups = list_append(old_groups, 'new-test-group') WHERE logid = " "(SELECT MIN(logid) FROM ev WHERE list_contains(old_groups, 'sysop') " "AND NOT list_contains(new_groups, 'sysop'))") print("PLANTED: one administrator removal given a group name that is not in GROUPS") if "dead-mapping-key" in plant: GROUPS["ghost-group"] = "elevated" print("PLANTED: a group name in GROUPS that no log event uses") if "missing-month" in plant: n = con.sql("SELECT COUNT(*) FROM ev WHERE ts >= '2023-03-01' AND ts < '2023-04-01'").fetchone()[0] con.sql("DELETE FROM ev WHERE ts >= '2023-03-01' AND ts < '2023-04-01'") print(f"PLANTED: the {n} events of March 2023 left out") # ---- checks on the events, before any per-person lookup ---- failures = [] n_rows, n_distinct = con.sql("SELECT COUNT(*), COUNT(DISTINCT logid) FROM ev").fetchone() if n_rows != n_distinct: failures.append(f"{n_rows} rows but {n_distinct} distinct log ids: {n_rows - n_distinct} events loaded twice") if n_rows != source_events: failures.append(f"the API returned {source_events} events but {n_rows} are loaded ({n_rows - source_events:+d})") out_of_range = con.sql(f"SELECT COUNT(*) FROM ev WHERE ts < '{START[:-1]}' OR ts > '{END[:-1]}'").fetchone()[0] if out_of_range: failures.append(f"{out_of_range} events fall outside the period") # every group name that was held before or after an event must be in GROUPS seen = {g for (g,) in con.sql("SELECT DISTINCT UNNEST(old_groups || new_groups) FROM ev").fetchall()} unmapped = sorted(seen - set(GROUPS)) if unmapped: counts = {g: con.execute("SELECT COUNT(*) FROM ev WHERE list_contains(old_groups || new_groups, ?)", [g]).fetchone()[0] for g in unmapped} failures.append("groups not in GROUPS: " + ", ".join(f"'{g}' in {c} events" for g, c in counts.items())) # every key of GROUPS must match a group in the data, or it is a mapping that silently does nothing dead = sorted(set(GROUPS) - seen) if dead: failures.append("keys of GROUPS that match no group in the log: " + ", ".join(f"'{g}'" for g in dead)) # every reason keyword must match at least one comment of a removal removals = con.sql(""" SELECT logid, ts, title, comment, list_filter(old_groups, g -> NOT list_contains(new_groups, g)) AS lost FROM ev WHERE NOT hidden AND action = 'rights' AND title NOT LIKE '%@%' AND len(list_filter(old_groups, g -> NOT list_contains(new_groups, g))) > 0""").fetchall() low = [r[3].lower() for r in removals] for reason, keys in REASON_RULES: for k in keys: if not any(has_keyword(c, k) for c in low): failures.append(f"reason keyword '{k}' ({reason}) matches no removal comment") # month-by-month completeness: a month far below the usual means a page of results went missing months = {m: c for m, c in con.sql("SELECT strftime(ts, '%Y-%m'), COUNT(*) FROM ev GROUP BY 1").fetchall()} expected_months = [f"{y}-{m:02d}" for y in YEARS for m in range(1, 13)] median_month = sorted(months.values())[len(months) // 2] if months else 0 for m in expected_months: if months.get(m, 0) < 0.25 * median_month: failures.append(f"month {m} has {months.get(m, 0)} events, under a quarter of the median month ({median_month})") if failures: print() print("SELF-CHECK FAILED: no lookups made, no CSV written") for f in failures: print(" -", f) sys.exit(1) # ---- classify and look up each removal ---- kind_of = GROUPS rows = [] for logid, ts, title, comment, lost in removals: c = comment.lower() reason = next((r for r, keys in REASON_RULES if any(has_keyword(c, k) for k in keys)), OTHER) elevated = [g for g in lost if kind_of[g] == "elevated"] if not elevated: continue rows.append({"logid": logid, "ts": ts, "title": title, "reason": reason, "sysop": "sysop" in elevated}) print(f"{len(removals)} removals of any group by hand; {len(rows)} remove an elevated group; " f"looking up each person's last edit (one request each)") for r in rows: r["last"] = last_edit(r["title"], r["ts"].strftime("%Y-%m-%dT%H:%M:%SZ")) r["days"] = None if r["last"] is None else (r["ts"] - _ts(r["last"])).total_seconds() / 86400.0 con.sql("CREATE TABLE rem (logid BIGINT, yr INT, reason VARCHAR, sysop BOOLEAN, days DOUBLE)") con.executemany("INSERT INTO rem VALUES (?, ?, ?, ?, ?)", [(r["logid"], r["ts"].year, r["reason"], r["sysop"], r["days"]) for r in rows]) out = [] for scope, cond in (("any elevated group", "TRUE"), ("administrator (sysop)", "sysop")): for year in [None] + YEARS: for reason in [None] + REASONS: where = f"{cond} AND ({'yr = ' + str(year) if year else 'TRUE'}) AND " \ f"({'reason = ' + repr(reason) if reason else 'TRUE'})" n, with_edit, med, q1, q3, over = con.sql(f""" SELECT COUNT(*), COUNT(days), quantile_cont(days, 0.5), quantile_cont(days, 0.25), quantile_cont(days, 0.75), COUNT(*) FILTER (WHERE days > 365) FROM rem WHERE {where}""").fetchone() out.append({"scope": scope, "year": year or "all", "reason": reason or "all", "removals": n, "with_prior_edit": with_edit, "no_edit_found": n - with_edit, "median_days": med, "q1_days": q1, "q3_days": q3, "pct_over_365_days": (100.0 * over / with_edit) if with_edit else None}) fields = ["scope", "year", "reason", "removals", "with_prior_edit", "no_edit_found", "median_days", "q1_days", "q3_days", "pct_over_365_days"] print() print(f"{'scope':<24}{'year':<6}{'reason':<26}{'removals':>9}{'with edit':>10}{'median d':>9}{'q1 d':>8}{'q3 d':>8}{'>365d %':>9}") for o in out: if o["removals"] == 0: continue f1 = lambda v: "-" if v is None else f"{v:.1f}" print(f"{o['scope']:<24}{o['year']!s:<6}{o['reason']:<26}{o['removals']:>9}{o['with_prior_edit']:>10}" f"{f1(o['median_days']):>9}{f1(o['q1_days']):>8}{f1(o['q3_days']):>8}{f1(o['pct_over_365_days']):>9}") print() print(f"SELF-CHECK PASSED (made before the lookups): {source_events} events read, none twice, every group " f"mapped, every mapping key and reason keyword matched, every month present") with open(a.out, "w", newline="") as f: w = csv.DictWriter(f, fieldnames=fields, lineterminator="\n") w.writeheader() for o in out: w.writerow({k: (f"{v:.1f}" if isinstance(v, float) else ("" if v is None else v)) for k, v in o.items()}) print(f"wrote {a.out}") def _ts(s): import datetime return datetime.datetime.strptime(s, "%Y-%m-%d %H:%M:%S") if __name__ == "__main__": main()