selfhostedops-stack/offboarding/offboard.py

#!/usr/bin/env python3
"""Offboarding for the selfhostedops stack (episode 15).

    ./offboard.py <username> [--dry-run]

Run it on the box, as root, from this directory. It switches one person off in
every service that has an API for it, one step after the other, and prints
what it did. At the end it prints what only a human can do: the shared
passwords that person could read are not safe until someone changes them.

Nothing clever happens here. Each block is one service, one or two API calls.
Credentials come from offboard.env (see offboard.env.example), the
Vaultwarden admin token from the stack's own .env.

Order matters a little: the identity provider goes first, so no new login is
possible while the rest is still running.
"""
import http.cookiejar, json, os, sqlite3, subprocess, sys, time, urllib.parse, urllib.request

HERE = os.path.dirname(os.path.abspath(__file__))
STACK = os.path.dirname(HERE)


def load_env(path):
    env = {}
    for line in open(path):
        line = line.strip()
        if line and not line.startswith("#") and "=" in line:
            k, v = line.split("=", 1)
            env[k] = v.strip().strip('"')
    return env


CFG = {**load_env(os.path.join(STACK, ".env")), **load_env(os.path.join(HERE, "offboard.env"))}
DOMAIN = CFG["DOMAIN"]
todo, report, started = [], [], time.time()


def ip(service):
    # Admin endpoints that Caddy keeps closed (vaultwarden /admin, the n8n API) are
    # called on the container itself. The box is on the docker bridge, so this works.
    out = subprocess.run(["docker", "inspect", "-f", "{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}",
                          f"selfhostedops-stack-{service}-1"], capture_output=True, text=True).stdout.split()
    return out[0]


def volume(name):
    return subprocess.run(["docker", "volume", "inspect", "-f", "{{.Mountpoint}}", f"selfhostedops-stack_{name}"],
                          capture_output=True, text=True).stdout.strip()


def call(method, url, body=None, headers=None, form=False, opener=None):
    data = None
    headers = dict(headers or {})
    if body is not None and form:
        data = urllib.parse.urlencode(body).encode()
        headers["Content-Type"] = "application/x-www-form-urlencoded"
    elif body is not None:
        data = json.dumps(body).encode()
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with (opener or urllib.request.build_opener()).open(req, timeout=20) as r:
            raw = r.read()
            return r.status, json.loads(raw) if raw[:1] in (b"{", b"[") else raw.decode(errors="replace")
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode(errors="replace")[:200]


def step(service, what, ok, detail=""):
    report.append((service, what, ok))
    print(f"{time.time() - started:6.1f}s  [{'ok' if ok else '!!'}] {service:12} {what} {detail}".rstrip(), flush=True)


def pocket_id(user):
    h = {"X-API-KEY": CFG["POCKET_ID_API_KEY"]}
    base = f"https://id.{DOMAIN}/api"
    s, j = call("GET", f"{base}/users?search={urllib.parse.quote(user)}", headers=h)
    hits = [u for u in (j.get("data", []) if s == 200 else []) if u["username"] == user]
    if not hits:
        return step("pocket-id", "user not found", False)
    u = hits[0]
    dto = {k: u[k] for k in ("username", "email", "firstName", "lastName", "isAdmin", "locale")}
    if not DRY:
        s, _ = call("PUT", f"{base}/users/{u['id']}", {**dto, "disabled": True}, h)
    step("pocket-id", "disabled (no new login, refresh tokens refused)", DRY or s == 200)
    return u["email"]


def forgejo(user):
    h = {"Authorization": "token " + CFG["FORGEJO_ADMIN_TOKEN"]}
    base = f"https://git.{DOMAIN}/api/v1"
    if not DRY:
        s, _ = call("PATCH", f"{base}/admin/users/{user}", {"login_name": user, "source_id": 0, "prohibit_login": True}, h)
        step("forgejo", "prohibit_login (web, git over ssh and https, api)", s == 200)
    s, keys = call("GET", f"{base}/users/{user}/keys", headers=h)
    for k in keys if s == 200 else []:
        if not DRY:
            s2, _ = call("DELETE", f"{base}/admin/users/{user}/keys/{k['id']}", headers=h)
        step("forgejo", f"ssh key '{k['title']}' deleted", DRY or s2 == 204)
    # An admin may list a user's access tokens (Sudo header), but deleting one needs the
    # owner's password (basic auth). prohibit_login already blocks them, so: name them.
    s, toks = call("GET", f"{base}/users/{user}/tokens", headers={**h, "Sudo": user})
    names = [t["name"] for t in toks] if s == 200 else []
    if names:
        todo.append(f"Forgejo: access tokens {names} are blocked by prohibit_login but still exist; "
                    "delete them if the account is ever re-enabled (only the owner can)")


def outline(email):
    h = {"Authorization": "Bearer " + CFG["OUTLINE_API_KEY"]}
    base = f"https://wiki.{DOMAIN}/api"
    s, j = call("POST", f"{base}/users.list", {"query": email, "filter": "all"}, h)
    hits = [u for u in (j.get("data", []) if s == 200 else []) if u.get("email") == email]
    if not hits:
        return step("outline", "user not found", False)
    uid = hits[0]["id"]
    if not DRY:
        s, _ = call("POST", f"{base}/users.suspend", {"id": uid}, h)
    step("outline", "suspended (browser session and api keys refused)", DRY or s == 200)
    # Outline 1.10 lists an admin only their own API keys, so the user's keys cannot be
    # deleted from here. A suspended user's keys and browser sessions are refused anyway.


def vaultwarden(email):
    jar = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
    base = f"http://{ip('vaultwarden')}"
    s, _ = call("POST", f"{base}/admin", {"token": CFG["VAULTWARDEN_ADMIN_TOKEN"]}, form=True, opener=jar)
    if s != 200:  # 429: the admin login allows only a few tries per 5 minutes
        return step("vaultwarden", f"admin login refused (HTTP {s}), run again in 5 minutes", False)
    s, users = call("GET", f"{base}/admin/users", opener=jar)
    hits = [u for u in (users if s == 200 else []) if u["email"] == email]
    if not hits:
        return step("vaultwarden", "user not found", False)
    uid = hits[0]["id"]
    if not DRY:
        # The admin routes only match with a JSON content type, without a body they answer 404.
        s, _ = call("POST", f"{base}/admin/users/{uid}/disable", {}, opener=jar)
    step("vaultwarden", "disabled + all devices logged out", DRY or s == 200)
    # What could this person read? The item names are encrypted, the count is not.
    db = volume("vaultwarden_data") + "/db.sqlite3"
    q = ("SELECT o.name, COUNT(DISTINCT cc.cipher_uuid) FROM users_organizations uo "
         "JOIN organizations o ON o.uuid = uo.org_uuid "
         "JOIN collections c ON c.org_uuid = uo.org_uuid "
         "LEFT JOIN users_collections ucl ON ucl.collection_uuid = c.uuid AND ucl.user_uuid = uo.user_uuid "
         "JOIN ciphers_collections cc ON cc.collection_uuid = c.uuid "
         "WHERE uo.user_uuid = ? AND (uo.access_all = 1 OR ucl.user_uuid IS NOT NULL) GROUP BY o.name")
    # ponytail: direct collection grants only; add collections_groups if you use Vaultwarden groups
    rows = sqlite3.connect(f"file:{db}?mode=ro", uri=True).execute(q, (uid,)).fetchall()
    for org, n in rows:
        todo.append(f"Vaultwarden: remove the user from organization '{org}' (needs an org owner) and "
                    f"CHANGE THE {n} SHARED PASSWORDS they could read. A copy on their laptop still opens "
                    f"with their master password, offline, forever.")


def netbird(email):
    h = {"Authorization": "Token " + CFG["NETBIRD_PAT"]}
    base = f"https://vpn.{DOMAIN}/api"
    s, users = call("GET", f"{base}/users", headers=h)
    s2, peers = call("GET", f"{base}/peers", headers=h)
    for u in [u for u in (users if s == 200 else []) if u.get("email") == email]:
        mine = [p["name"] for p in (peers if s2 == 200 else []) if p.get("user_id") == u["id"]]
        if not DRY:
            s3, _ = call("DELETE", f"{base}/users/{u['id']}", headers=h)
        step("netbird", f"user deleted with its peers {mine}", DRY or s3 == 200)


def n8n(email):
    h = {"X-N8N-API-KEY": CFG["N8N_API_KEY"]}
    base = f"http://{ip('n8n')}:5678/api/v1"
    s, u = call("GET", f"{base}/users/{urllib.parse.quote(email)}", headers=h)
    if s == 404:
        return step("n8n", "no such user (already gone)", True)
    if s != 200:
        return step("n8n", f"lookup failed (HTTP {s})", False)
    if not DRY:
        s, _ = call("DELETE", f"{base}/users/{u['id']}", headers=h)
    step("n8n", "user deleted (workflows stay with the instance owner)", DRY or s in (200, 204), "" if DRY else f"HTTP {s}")


def ntfy(user):
    cmd = ["docker", "compose", "-f", os.path.join(STACK, "compose.yml"), "exec", "-T", "ntfy", "ntfy", "user", "del", user]
    r = subprocess.run(cmd, capture_output=True, text=True) if not DRY else None
    gone = DRY or r.returncode == 0 or "does not exist" in r.stderr + r.stdout  # a second run is fine
    step("ntfy", "user and its tokens deleted", gone, "" if DRY else (r.stdout + r.stderr).strip())


def tinyauth(email):
    # tinyauth keeps its own session for a day and does not ask Pocket ID again.
    n = 0
    if not DRY:
        with sqlite3.connect(volume("tinyauth_data") + "/tinyauth.db") as db:
            n = db.execute("DELETE FROM sessions WHERE email = ?", (email,)).rowcount
    step("tinyauth", f"{n} gate sessions deleted", True)


if __name__ == "__main__":
    if len(sys.argv) < 2 or not sys.argv[1].replace("-", "").isalnum():
        sys.exit(__doc__)
    USER, DRY = sys.argv[1], "--dry-run" in sys.argv
    print(f"offboarding {USER}" + ("  (DRY RUN: only looks, changes nothing)" if DRY else ""))
    email = pocket_id(USER) or f"{USER}@{DOMAIN}"
    for fn, arg in ((tinyauth, email), (forgejo, USER), (outline, email), (vaultwarden, email),
                    (netbird, email), (n8n, email), (ntfy, USER)):
        try:
            fn(arg)
        except Exception as e:  # one broken service must not stop the others
            step(fn.__name__, f"FAILED: {e}", False)
    todo.append("Uptime Kuma and anything else with a shared login: change that password now")
    todo.append("Get the laptop back. Until then assume every local copy is readable.")
    failed = [r for r in report if not r[2]]
    print(f"\n{len(report) - len(failed)} of {len(report)} steps done in {time.time() - started:.1f} s"
          + (f", {len(failed)} FAILED, check above" if failed else ""))
    print("\nStill for a human:")
    for t in todo:
        print("  - " + t)