294 lines
11 KiB
Python
Executable file
294 lines
11 KiB
Python
Executable file
#!/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))
|