#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""who_banned_me.py - "is anyone on the fediverse blocking me, and did they tell me?"

Written 2026-10-17 after I found out that two servers had banned me three days
earlier and nothing anywhere had informed me. Standard library only, no install.

    python3 who_banned_me.py https://instance.example/u/yourname
    python3 who_banned_me.py https://instance.example/u/yourname --limit 40
    python3 who_banned_me.py https://instance.example/u/yourname --json

Why a whole tool for this: Lemmy has two different kinds of ban and they live in
two different tables.

  * site-level    - `person.banned` on the instance's copy of your actor
  * community-level - a row in that instance's modlog, `ModBanFromCommunity`

A community ban is the one that will bite you, because it does not change the
flag that every dashboard shows. You can be banned from the one community you
post in and your account page will still say everything is fine. On 2026-10-17 a
single moderator click propagated to 68 servers in 31 seconds; 64 of those 68
showed `banned: false` at the site level.

Output is grouped by server, not by ban, because "68 bans" sounds like 68
decisions and it usually is one decision replicated.
"""

import argparse
import concurrent.futures as cf
import json
import sys
import urllib.error
import urllib.parse
import urllib.request

UA = {"User-Agent": "who-banned-me/1.0 (one-off self-audit; no scraping)"}


def get(url, timeout=12):
    req = urllib.request.Request(url, headers=UA)
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.load(r)


def servers(actor_id, extra=()):
    """Ask an instance which servers it federates with.

    Returns {domain: [sources]}. One instance's list is a scope, not the
    fediverse: on 2026-10-17 my own list had 343 domains and another instance's
    had 427, with only 231 in common. Anything reachable from just one of them
    is invisible to the other, and a ban there would never show up.

    `extra` is a list of hostnames to take lists from as well. The union is
    what gets asked, and every row remembers which list(s) named it.
    """
    own = urllib.parse.urlparse(actor_id).netloc
    out = {}
    for src in [own] + [h for h in extra if h != own]:
        d = get("https://%s/api/v3/federated_instances" % src, timeout=25)
        linked = d["federated_instances"]["linked"] or []
        for i in linked:
            if i.get("software") != "lemmy":
                continue
            out.setdefault(i["domain"], []).append(src)
    return out


def collapse_aliases(hosts, workers=24, timeout=10):
    """One machine can answer on two domains. Counting domains counts DNS, not decisions.

    Found the hard way on 2026-10-17: the audit reported five site-level bans,
    and two of them - lemmy.world and hackingne.ws - returned the *same* person
    id for the same actor, with the same moderator in the modlog. Their
    `/api/v3/site` both say `actor_id: https://lemmy.world/` and the same
    version string. One ban, two names.

    That matters because "N servers banned me" was the claim, and every ban was
    being counted twice. The identity to compare is the site actor_id, not the
    hostname: the hostname is just where the request went.
    """
    def ident(h):
        try:
            d = get("https://%s/api/v3/site" % h, timeout=timeout)
            return h, ((d.get("site_view") or {}).get("site") or {}).get("actor_id")
        except Exception:
            return h, None

    with cf.ThreadPoolExecutor(workers) as ex:
        pairs = list(ex.map(ident, hosts))

    groups = {}
    for h, a in pairs:
        if a:
            groups.setdefault(a.rstrip("/"), []).append(h)
    keep = {}
    for key, names in groups.items():
        want = urllib.parse.urlparse(key).netloc
        keep[key] = want if want in names else names[0]

    out, aliases = [], []
    for h, a in pairs:
        if not a:
            out.append(h)          # unasked stays in the report; do not drop it here
            continue
        key = a.rstrip("/")
        if h != keep[key]:
            aliases.append((keep[key], h))
            continue
        out.append(h)
    if aliases:
        print("collapsed aliases (same site actor_id): %s" %
              "; ".join("%s <- %s" % (k, v) for k, v in aliases), file=sys.stderr)
    return out, aliases


def check(host, actor_id, timeout):
    """Ask one server. The hard part is not asking, it is not merging two answers.

    `resolve_object` answers `HTTP 400 couldnt_find_object` when the server has
    never seen this actor. That is a *reply*, not a failure: this server cannot
    ban someone it does not have. The first version of this file caught every
    exception as "unreachable", so 107 servers that plainly said "no, I do not
    know this person" were reported as holes in the audit - and 137 HTTP errors
    got described in the README as "mostly 403s from Cloudflare", a number I had
    never actually broken down. Both were wrong in the same direction: a value
    that means "they answered no" was being spent on "nobody answered".

    Codes:
      400/404 -> does-not-know-me   (an answer)
      anything else / timeout -> unreachable (no answer)
    """
    q = urllib.parse.quote(actor_id, safe="")
    try:
        d = get("https://%s/api/v3/resolve_object?q=%s" % (host, q), timeout=timeout)
    except urllib.error.HTTPError as e:
        if e.code in (400, 404):
            return {"server": host, "state": "does-not-know-me", "http": e.code}
        return {"server": host, "state": "unreachable", "error": "HTTP%d" % e.code}
    except Exception as e:
        return {"server": host, "state": "unreachable", "error": type(e).__name__}

    person = (d.get("person") or {}).get("person") or \
             (d.get("person_view") or {}).get("person")
    if not person:
        return {"server": host, "state": "does-not-know-me"}

    out = {
        "server": host,
        "state": "knows-me",
        "id_on_that_server": person["id"],
        "site_banned": bool(person.get("banned")),
        "bot_flag": person.get("bot_account"),
        "community_bans": [],
    }
    try:
        m = get("https://%s/api/v3/modlog?type_=ModBanFromCommunity"
                "&other_person_id=%d&limit=50" % (host, person["id"]), timeout=timeout)
    except Exception:
        return out
    for e in m.get("banned_from_community", []):
        b, c = e["mod_ban_from_community"], e["community"]
        if b.get("banned"):
            out["community_bans"].append({
                "community": "%s@%s" % (c["name"], urllib.parse.urlparse(c["actor_id"]).netloc),
                "when": b["when_"][:19],
                "reason": b.get("reason"),
            })
    return out


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("actor", help="e.g. https://lemmy.ml/u/someone")
    ap.add_argument("--limit", type=int, default=0, help="only ask the first N servers")
    ap.add_argument("--workers", type=int, default=24)
    ap.add_argument("--timeout", type=float, default=10)
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--also", action="append", default=[], metavar="INSTANCE",
                    help="also take the server list from this instance; "
                         "repeatable. One list is a scope, not the fediverse.")
    a = ap.parse_args()

    own = urllib.parse.urlparse(a.actor).netloc
    listed = servers(a.actor, a.also)
    if a.limit:
        listed = dict(list(listed.items())[:a.limit])

    known_hosts, aliases = collapse_aliases(list(listed), a.workers, a.timeout)
    # collapse_aliases drops the losing hostname, but that hostname may be the
    # only one that a second server list mentioned. Carry its provenance to the
    # name that survived, or the union quietly shrinks back to one list's view.
    provenance = {h: set(listed.get(h, [])) for h in known_hosts}
    for kept, dropped in aliases:
        provenance.setdefault(kept, set()).update(listed.get(dropped, []))
    print("asking %d servers whether they know %s ..." % (len(known_hosts), a.actor),
          file=sys.stderr)

    with cf.ThreadPoolExecutor(a.workers) as ex:
        rows = list(ex.map(lambda h: check(h, a.actor, a.timeout), known_hosts))

    scope = {
        "list_from": own,
        "also_from": a.also,
        "listed": len(listed),
        "asked": len(known_hosts),
        "limit": a.limit,
        "only_in_other_lists": sum(1 for h, src in listed.items() if own not in src),
    }
    for r in rows:
        r["listed_by"] = sorted(provenance.get(r["server"], []))

    if a.json:
        print(json.dumps({"scope": scope, "servers": rows}, indent=1))
        return 0

    known = [r for r in rows if r["state"] == "knows-me"]
    unknown = [r for r in rows if r["state"] == "unreachable"]
    absent = [r for r in rows if r["state"] == "does-not-know-me"]
    site = [r for r in known if r["site_banned"]]
    comm = [(r["server"], b) for r in known for b in r["community_bans"]]

    print()
    if scope["also_from"]:
        print("scope                  : union of server lists from %s + %s, not the fediverse"
              % (scope["list_from"], ", ".join(scope["also_from"])))
    else:
        print("scope                  : this is the list %s knows about, not the fediverse"
              % scope["list_from"])
    print("domains listed         : %d (%d only in the other list%s)"
          % (scope["listed"], scope["only_in_other_lists"],
             "s" if len(scope["also_from"]) != 1 else ""))
    print("servers asked          : %d" % len(rows))
    print("servers that know you  : %d" % len(known))
    print("never heard of you     : %d  (they answered; they cannot ban what they lack)"
          % len(absent))
    print("site-level bans        : %d" % len(site))
    print("community-level bans   : %d across %d servers"
          % (len(comm), len({h for h, _ in comm})))
    if unknown:
        codes = {}
        for r in unknown:
            codes[r.get("error")] = codes.get(r.get("error"), 0) + 1
        print("could not be asked     : %d (%s)" % (
            len(unknown), ", ".join("%s x%d" % (k, v) for k, v in sorted(codes.items()))))
        print("  -> an unreachable server is NOT a clean one. It is a hole in this report.")
    print()

    if site:
        print("SITE-LEVEL (whole instance):")
        for r in site:
            print("   %-36s bot_flag=%s" % (r["server"], r["bot_flag"]))
        print()

    if comm:
        print("COMMUNITY-LEVEL (one community per row):")
        for h, b in sorted(comm, key=lambda x: x[1]["when"]):
            print("   %-28s !%-22s %s  %s" % (
                h, b["community"], b["when"], b["reason"] or ""))
        print()
        silent = {h for h, _ in comm} - {r["server"] for r in site}
        print("%d of those servers say your account is not banned at the site level."
              % len(silent))
        print("That is why this needs its own tool: the flag you would normally")
        print("check stays false while the community ban does its work.")
    if not site and not comm:
        print("Nobody is blocking you that I can see.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())