taler-monitoring/check_qr_payloads.py

#!/usr/bin/env python3
"""Collect + validate QR-related payloads for taler-monitoring (www.qr-*).

Sources (CLI paths / URLs handled by caller writing local files):
  - landing HTML (href, data-qr-url, data-qr-taler, raw text)
  - demo-withdraw.json / auto-account.json (taler_withdraw_uri / qr_payload)
  - optional: already-decoded lines from zbarimg (kind=decoded)

Stdout TSV: source\\tkind\\turi\\tstatus\\tdetail
  kind: withdraw | pay | pay-template | refund | payto | https | other | empty
  status: ok | bad

Exit 0 always (caller tallies ok/bad lines). stderr for hard parse errors only.
"""
from __future__ import annotations

import json
import re
import sys
from html.parser import HTMLParser
from pathlib import Path
from urllib.parse import unquote, urlparse

# Schemes we care about for wallet / settlement QRs
TALER_RE = re.compile(r"taler://[^\s\"'<>]+", re.I)
PAYTO_RE = re.compile(r"payto://[^\s\"'<>]+", re.I)
HTTPS_RE = re.compile(r"https://[^\s\"'<>]+", re.I)

WITHDRAW_RE = re.compile(
    r"^taler://withdraw/([^/]+)/taler-integration/([0-9a-fA-F-]+)/?$"
)
# order pay / template — host may include path segments after
PAY_RE = re.compile(r"^taler://pay/([^/?#]+)/", re.I)
PAY_TEMPLATE_RE = re.compile(r"^taler://pay-template/([^/?#]+)/", re.I)
REFUND_RE = re.compile(r"^taler://refund/", re.I)
PAYTO_OK_RE = re.compile(r"^payto://[a-z0-9][a-z0-9+.-]*/", re.I)


def looks_like_real_uri(uri: str) -> bool:
    """Drop JS regex fragments, docs ellipses, incomplete placeholders."""
    if not uri or len(uri) < 12:
        return False
    if any(c in uri for c in "[]{}<>\n\r\t"):
        return False
    if "…" in uri or "..." in uri:
        return False
    # incomplete withdraw stubs in docs
    if re.match(r"^taler://withdraw/?$", uri, re.I):
        return False
    if re.match(r"^taler://pay/?$", uri, re.I):
        return False
    # trailing junk / JS regex fragments (allow ? & = in query strings)
    if uri.endswith((")", "(", "\\", "^", "*")):
        return False
    if re.search(r"[\\^$*]", uri):
        return False
    # lone incomplete `taler://…?` without path
    if re.match(r"^taler://[^/]+$", uri, re.I):
        return False
    return True


def strip_default_port_host(host: str) -> str | None:
    if not host or host.startswith(":"):
        return None
    if ":" in host:
        h, _, p = host.rpartition(":")
        if not h or not p.isdigit():
            return None
        if p in ("443", "80"):
            return None  # must be stripped for wallets
    return host


def classify_and_validate(uri: str) -> tuple[str, str, str]:
    """Return (kind, status, detail)."""
    uri = (uri or "").strip()
    if not uri:
        return "empty", "bad", "empty"
    # strip accidental trailing punctuation from HTML scrape
    uri = uri.rstrip(").,;]")
    low = uri.lower()

    if low.startswith("taler://withdraw/"):
        m = WITHDRAW_RE.match(uri)
        if not m:
            return "withdraw", "bad", "shape need taler://withdraw/HOST/taler-integration/UUID"
        host = m.group(1)
        if strip_default_port_host(host) is None:
            return "withdraw", "bad", "host/port invalid or default :443/:80 not stripped"
        return "withdraw", "ok", f"host={host} id={m.group(2)[:12]}…"

    if low.startswith("taler://pay-template/"):
        m = PAY_TEMPLATE_RE.match(uri)
        if not m:
            return "pay-template", "bad", "need taler://pay-template/HOST/…"
        host = m.group(1)
        if strip_default_port_host(host) is None:
            return "pay-template", "bad", "host/port invalid or default port not stripped"
        return "pay-template", "ok", f"host={host}"

    if low.startswith("taler://pay/"):
        m = PAY_RE.match(uri)
        if not m:
            return "pay", "bad", "need taler://pay/HOST/…"
        host = m.group(1)
        if strip_default_port_host(host) is None:
            return "pay", "bad", "host/port invalid or default port not stripped"
        if "instances" not in uri and "/orders/" not in uri:
            # still often valid private order URIs
            return "pay", "ok", f"host={host}"
        return "pay", "ok", f"host={host}"

    if low.startswith("taler://refund/"):
        if not REFUND_RE.match(uri):
            return "refund", "bad", "bad refund URI"
        return "refund", "ok", "refund"

    if low.startswith("taler://"):
        return "other", "ok", "other taler:// (accepted)"

    if low.startswith("payto://"):
        if not PAYTO_OK_RE.match(uri):
            return "payto", "bad", "need payto://method/…"
        # wallet pay QR must not be payto for GOA demo-withdraw (checked elsewhere)
        return "payto", "ok", uri.split("?", 1)[0][:72]

    if low.startswith("https://") or low.startswith("http://"):
        p = urlparse(uri)
        if p.scheme not in ("http", "https") or not p.netloc:
            return "https", "bad", "bad URL"
        return "https", "ok", p.netloc + (p.path or "/")[:40]

    return "other", "bad", "unsupported scheme"


class LinkHarvester(HTMLParser):
    def __init__(self) -> None:
        super().__init__()
        self.uris: list[tuple[str, str]] = []  # (source_hint, uri)

    def handle_starttag(self, tag, attrs):
        ad = {k.lower(): (v or "") for k, v in attrs}
        for key in ("href", "src", "data-qr-url", "data-qr-taler", "data-uri", "data-payto"):
            if key in ad and ad[key]:
                self.uris.append((f"attr:{key}", ad[key].strip()))
        for k, v in ad.items():
            if k.startswith("data-") and v and (
                "taler://" in v.lower() or "payto://" in v.lower() or k.endswith("qr-url")
            ):
                self.uris.append((f"attr:{k}", v.strip()))

    def handle_data(self, data):
        if not data or "://" not in data:
            return
        for m in TALER_RE.finditer(data):
            self.uris.append(("text", m.group(0)))
        for m in PAYTO_RE.finditer(data):
            self.uris.append(("text", m.group(0)))


def harvest_html(path: Path, source: str) -> list[tuple[str, str, str]]:
    raw = path.read_text(encoding="utf-8", errors="replace")
    h = LinkHarvester()
    try:
        h.feed(raw)
    except Exception:
        pass
    out: list[tuple[str, str, str]] = []
    for hint, u in h.uris:
        lu = u.lower()
        if lu.startswith(("taler://", "payto://", "https://", "http://")):
            if looks_like_real_uri(u):
                out.append((f"{source}:{hint}", u, "html"))
        # raster image paths that look like QR assets (not .js)
        elif re.search(r"\.(png|jpe?g|gif)(\?|$)", lu) and "qr" in lu:
            out.append((f"{source}:imgpath", u, "imgpath"))
    # Prefer attribute/text harvest; light regex only for complete-looking URIs in pre/code
    for m in TALER_RE.finditer(raw):
        u = m.group(0).rstrip(").,;]'\"")
        if looks_like_real_uri(u) and (
            WITHDRAW_RE.match(u)
            or PAY_RE.match(u)
            or PAY_TEMPLATE_RE.match(u)
            or u.lower().startswith("taler://refund/")
        ):
            out.append((f"{source}:re", u, "html"))
    for m in PAYTO_RE.finditer(raw):
        u = m.group(0).rstrip(").,;]'\"")
        if looks_like_real_uri(u) and PAYTO_OK_RE.match(u):
            out.append((f"{source}:re", u, "html"))
    return out


def harvest_json(path: Path, source: str) -> list[tuple[str, str, str]]:
    try:
        d = json.loads(path.read_text(encoding="utf-8", errors="replace"))
    except Exception as e:
        return [(source, "", f"json-error:{e}")]
    out: list[tuple[str, str, str]] = []
    if not isinstance(d, dict):
        return out
    for key in (
        "taler_withdraw_uri",
        "qr_payload",
        "taler_pay_uri",
        "payto_uri",
        "payto",
        "uri",
    ):
        v = d.get(key)
        if isinstance(v, str) and v.strip():
            out.append((f"{source}:{key}", v.strip(), "json"))
    # nested
    for k, v in d.items():
        if isinstance(v, str) and (
            v.startswith("taler://") or v.startswith("payto://")
        ):
            out.append((f"{source}:{k}", v.strip(), "json"))
    return out


def main(argv: list[str]) -> int:
    # args: pairs of  source_label  path
    if len(argv) < 3 or len(argv) % 2 == 0:
        print(
            "usage: check_qr_payloads.py LABEL path [LABEL path ...]",
            file=sys.stderr,
        )
        return 2
    seen: set[str] = set()
    rows: list[tuple[str, str, str, str, str]] = []
    i = 1
    while i < len(argv):
        label, path_s = argv[i], argv[i + 1]
        i += 2
        path = Path(path_s)
        if not path.is_file():
            rows.append((label, "empty", "", "bad", f"missing file {path_s}"))
            continue
        if path.suffix.lower() in (".html", ".htm") or "html" in path.name:
            items = harvest_html(path, label)
        elif path.suffix.lower() == ".json" or path.name.endswith(".json"):
            items = harvest_json(path, label)
        else:
            # plain text: one URI per line or raw
            text = path.read_text(encoding="utf-8", errors="replace").strip()
            items = []
            for line in text.splitlines():
                line = line.strip()
                if line:
                    items.append((label, line, "text"))
            if not items and text:
                items.append((label, text, "text"))

        for src, uri, _origin in items:
            if not uri:
                continue
            # skip pure fragment / relative without scheme for validate
            if uri.startswith("#") or uri == "/":
                continue
            lu = uri.lower()
            if lu.startswith("/") or "imgpath" in src:
                continue
            if not looks_like_real_uri(uri):
                continue
            # https: app-store / wallet / explicit data-qr-* only
            if lu.startswith("http"):
                if "data-qr" not in src and not any(
                    x in lu
                    for x in (
                        "play.google",
                        "apps.apple",
                        "f-droid",
                        "wallet.taler",
                        "taler.net/wallet",
                    )
                ):
                    continue
            if uri in seen:
                continue
            seen.add(uri)
            kind, status, detail = classify_and_validate(uri)
            rows.append((src, kind, uri, status, detail))

    for src, kind, uri, status, detail in rows:
        # TSV — uri may contain tabs rarely; replace
        safe = uri.replace("\t", " ").replace("\n", " ")
        print(f"{src}\t{kind}\t{safe}\t{status}\t{detail}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))