felhom.eu/documentation/audits/lockouts-2026-10-01/tools/lk.py

#!/usr/bin/env python3
"""Login probes for R-752 on 9202 — each one a fresh, cookie-less visitor through traefik (a STRANGER unless it holds
the right password). Every probe returns one word: ok / wrong / locked / other:<detail>. EVIDENCE, NOT PRODUCT."""
import json, os, re, subprocess, tempfile, time
from datetime import datetime, timezone
import walk as w


def now():
    return datetime.now(timezone.utc).strftime("%H:%M:%S")


def p(*a):
    print(now(), *a, flush=True)


def curl(sub, path, *extra, data=None, jar=None, timeout=20):
    args = ["curl", "-sk", "--max-time", str(timeout), "-D", "-", "-H", f"Host: {sub}.{w.DOMAIN}"]
    if jar:
        args += ["-b", jar, "-c", jar]
    args += list(extra)
    if data is not None:
        args += ["--data-raw", data]
    args.append(f"{w.BASE}{path}")
    r = subprocess.run(args, capture_output=True, text=True)
    return r.stdout or ""


def status(out):
    m = re.findall(r"(?m)^HTTP/\S+ (\d{3})", out)
    return m[-1] if m else "000"


def location(out):
    m = re.findall(r"(?im)^location:\s*(\S+)", out)
    return m[-1] if m else ""


def form_token(html, name):
    m = re.search(r'name="%s"\s+value="([^"]+)"' % name, html) or re.search(r'value="([^"]+)"\s+name="%s"' % name, html)
    return m.group(1) if m else ""


def enc(**kw):
    from urllib.parse import urlencode
    return urlencode(kw)


# ---- BookStack: the web form (Laravel CSRF _token), key email|ip, 5 tries / 60 s ----------------------------------
def bookstack(email, pw, sub="wiki"):
    jar = tempfile.mktemp()
    try:
        page = curl(sub, "/login", jar=jar)
        tok = form_token(page, "_token")
        out = curl(sub, "/login", "-H", "Content-Type: application/x-www-form-urlencoded", jar=jar,
                   data=enc(_token=tok, email=email, password=pw))
        loc = location(out)
        if status(out) == "302" and not loc.rstrip("/").endswith("/login"):
            return "ok"
        back = curl(sub, "/login", jar=jar)
        if re.search(r"(?i)too many|throttle|t[uú]l sok", back):
            return "locked"
        if status(out) in ("302", "422", "200"):
            return "wrong"
        return "other:" + status(out)
    finally:
        try:
            os.unlink(jar)
        except OSError:
            pass


# ---- Grafana: POST /login JSON ----------------------------------------------------------------------------------
def grafana(user, pw, sub="grafana"):
    out = curl(sub, "/login", "-H", "Content-Type: application/json", data=json.dumps({"user": user, "password": pw}))
    body = out.split("\r\n\r\n")[-1]
    if status(out) == "200":
        return "ok"
    if re.search(r"(?i)temporarily blocked|too many", body):
        return "locked"
    if status(out) in ("401", "400"):
        return "wrong"
    return "other:" + status(out) + " " + body[:80]


# ---- calibre-web-automated: the web form (Flask-WTF csrf_token), and OPDS basic auth -----------------------------
def calibre(user, pw, sub="books"):
    jar = tempfile.mktemp()
    try:
        page = curl(sub, "/login", jar=jar)
        tok = form_token(page, "csrf_token")
        out = curl(sub, "/login", "-H", "Content-Type: application/x-www-form-urlencoded", jar=jar,
                   data=enc(csrf_token=tok, username=user, password=pw, next="/"))
        if status(out) == "302" and "/login" not in location(out):
            return "ok"
        if re.search(r"(?i)wait one minute|v[aá]rj", out):
            return "locked"
        if re.search(r"(?i)wrong username|hib[aá]s", out) or status(out) == "200":
            return "wrong"
        return "other:" + status(out)
    finally:
        try:
            os.unlink(jar)
        except OSError:
            pass


def calibre_opds(user, pw, sub="books"):
    out = curl(sub, "/opds", "-u", f"{user}:{pw}")
    c = status(out)
    return {"200": "ok", "401": "wrong", "429": "locked"}.get(c, "other:" + c)


# ---- wger: the web login form (Django CSRF), as the catalog's own fixture logs in; django-axes on authenticate() ----
def wger(user, pw, sub="fitness"):
    jar = tempfile.mktemp()
    try:
        o = f"https://{sub}.{w.DOMAIN}"
        hdr = ["-H", f"Origin: {o}", "-H", f"Referer: {o}/en/user/login"]
        page = curl(sub, "/en/user/login", *hdr, jar=jar)
        m = re.search(r'name="csrfmiddlewaretoken" value="([^"]+)"', page)
        if not m:
            return "other:no-csrf " + status(page)
        out = curl(sub, "/en/user/login", *hdr, "-H", "Content-Type: application/x-www-form-urlencoded", jar=jar,
                   data=enc(csrfmiddlewaretoken=m.group(1), login=user, password=pw))
        c = status(out)
        jt = open(jar).read() if os.path.exists(jar) else ""
        if c == "302" and "sessionid" in jt:
            return "ok"
        if c in ("429", "403") or re.search(r"(?i)account locked|too many", out):
            return "locked"
        if c == "200":
            return "wrong"
        return "other:" + c
    finally:
        try:
            os.unlink(jar)
        except OSError:
            pass


def wait_until(fn, want, every, cap_s, label):
    """Poll fn() every `every` s until it answers `want`; returns minutes waited (None if never)."""
    t0 = time.time()
    while time.time() - t0 < cap_s:
        time.sleep(every)
        r = fn()
        p(f"  +{(time.time() - t0) / 60:.1f} min {label}: {r}")
        if r == want:
            return (time.time() - t0) / 60
    return None