felhom.eu/documentation/audits/catalog-sweep-2026-10-24/tools/ui.py

#!/opt/pw/bin/python
"""ui.py — drive the household's screens in real Chromium, step by step (rehearsal 2026-10-20).

usage: ui.py STEPS.json        (STEPS may also be '-' to read JSON from the env var UI_STEPS)
STEPS = {"size": "phone"|"desktop", "lang": "hu"|"en", "state": "/root/work/state-dash.json" (load+save cookies),
         "out": "/root/shots/<prefix>", "steps": [ ... ]}
step kinds:
  {"goto": URL}                         {"click": SELECTOR} / {"click_text": TEXT}
  {"fill": SELECTOR, "env": NAME}       — the value comes from env NAME (never printed)
  {"fill": SELECTOR, "value": TEXT}     — a non-secret value
  {"fill": SELECTOR, "env": NAME, "typo": true} — one letter changed (a household's typo)
  {"shot": NAME}                        — blanks every text/password/textarea input first, then screenshots full page
  {"wait": MS}                          {"text": true}
  {"wait_for": SELECTOR, "timeout": MS} {"fill_all": SELECTOR, "env": NAME} — every enabled match gets the secret
  {"select": SELECTOR, "option": VALUE} — prints the visible page text (trimmed)
  {"expect_nav": true} before a click   — waits for the navigation the click causes
Every step prints one line: what it did, the URL (any 32+ hex run masked), the HTTP status of navigations.
"""
import json, os, re, sys
from playwright.sync_api import sync_playwright

spec = json.loads(os.environ["UI_STEPS"]) if sys.argv[1] == "-" else json.load(open(sys.argv[1]))
SIZES = {"phone": dict(viewport={"width": 390, "height": 844}, device_scale_factor=2, is_mobile=True, has_touch=True,
                       user_agent="Mozilla/5.0 (Linux; Android 14; Pixel 8) AppleWebKit/537.36 (KHTML, like Gecko) "
                                  "Chrome/155.0.0.0 Mobile Safari/537.36"),
         "desktop": dict(viewport={"width": 1366, "height": 768})}
LOC = {"hu": "hu-HU", "en": "en-US"}
mask = lambda u: re.sub(r"[0-9a-f]{32,}", "<token>", u)
size, lang = spec.get("size", "phone"), spec.get("lang", "hu")
out = spec.get("out", "/root/shots/x")


def typo(v):
    i = len(v) // 2
    while i < len(v) and not v[i].isalpha():
        i += 1
    return v[:i] + ("q" if v[i] != "q" else "w") + v[i + 1:] if i < len(v) else v + "x"


with sync_playwright() as p:
    b = p.chromium.launch(executable_path="/usr/bin/chromium", args=["--no-sandbox"])
    kw = dict(locale=LOC[lang], extra_http_headers={"Accept-Language": LOC[lang]}, timezone_id="Europe/Budapest",
              ignore_https_errors=True, **SIZES[size])
    st = spec.get("state")
    if st and os.path.exists(st):
        kw["storage_state"] = st
    ctx = b.new_context(**kw)
    pg = ctx.new_page()
    resp = {"s": None}
    pg.on("response", lambda r: resp.update(s=r.status) if r.request.is_navigation_request() else None)
    nav_next = False
    for s in spec["steps"]:
        try:
            if "goto" in s:
                r = pg.goto(s["goto"], wait_until=s.get("until", "networkidle"), timeout=60000)
                title = ""
                for _ in range(5):  # an app that redirects right after load destroys the context mid-title (sweep 2026-10-24)
                    try:
                        pg.wait_for_load_state("load", timeout=15000)
                        title = pg.title()
                        break
                    except Exception:
                        pg.wait_for_timeout(1500)
                print(f"goto {mask(pg.url)} -> {r.status if r else '-'} {title!r}")
            elif "fill" in s:
                v = os.environ[s["env"]] if "env" in s else s["value"]
                if s.get("typo"):
                    v = typo(v)
                pg.fill(s["fill"], v)
                print(f"fill {s['fill']} ({'secret ' + s['env'] if 'env' in s else repr(v)}{', with a typo' if s.get('typo') else ''})")
            elif "click" in s or "click_text" in s:
                target = pg.locator(s["click"]) if "click" in s else pg.get_by_text(s["click_text"], exact=False)
                if nav_next:
                    with pg.expect_navigation(wait_until="networkidle", timeout=60000):
                        target.first.click()
                    nav_next = False
                else:
                    target.first.click()
                    try:
                        pg.wait_for_load_state("networkidle", timeout=15000)
                    except Exception:  # an app that keeps a live connection never goes idle — not a failure
                        pass
                print(f"click {s.get('click') or s.get('click_text')!r} -> {mask(pg.url)} status={resp['s']}")
            elif "expect_nav" in s:
                nav_next = True
            elif "wait_for" in s:
                pg.wait_for_selector(s["wait_for"], state="visible", timeout=s.get("timeout", 600000))
                print(f"visible {s['wait_for']}")
            elif "fill_all" in s:
                n = 0
                for el in pg.locator(s["fill_all"]).all():
                    if el.is_enabled():
                        el.fill(os.environ[s["env"]]); n += 1
                print(f"fill_all {s['fill_all']} ({n} fields, secret {s['env']})")
            elif "select" in s:
                pg.select_option(s["select"], s["option"])
                print(f"select {s['select']} = {s['option']!r}")
            elif "wait" in s:
                pg.wait_for_timeout(s["wait"])
            elif "shot" in s:
                # blank every text/password field FOR THE PHOTO ONLY, then put the values back (a blanked form that is
                # submitted afterwards refuses — the 2026-10-20 Nextcloud slip)
                pg.evaluate("window.__keep = []; document.querySelectorAll('input[type=text],input[type=password],input:not([type]),textarea')"
                            ".forEach(e => { if (e.value) { window.__keep.push([e, e.value]); e.value = ''; } })")
                f = f"{out}-{s['shot']}-{size}-{lang}.png"
                pg.screenshot(path=f, full_page=True)
                pg.evaluate("(window.__keep || []).forEach(([e, v]) => { e.value = v; }); window.__keep = [];")
                hs = pg.evaluate("document.documentElement.scrollWidth > window.innerWidth")
                print(f"shot {f} hscroll={hs}")
            elif "text" in s:
                print("text:", re.sub(r"\s+", " ", pg.inner_text("body"))[:1500])
        except Exception as e:  # noqa: BLE001 — a failing step is a result; record it and stop
            print(f"STEP FAILED {s}: {type(e).__name__}: {str(e)[:300]}")
            break
    if st:
        ctx.storage_state(path=st)
        os.chmod(st, 0o600)
    b.close()