koopa-admin-log/scripts/taler-bank/demo-withdraw-api.py

#!/usr/bin/env python3
"""
HTTP helper for bank landing:
  - GET /demo-withdraw.json  — mint shared-pool (explorer) demo withdraw
  - GET /auto-account.json   — create a personal bank account (balance 0)
    and return one-time credentials for the user to copy

Listens on 127.0.0.1:19096 (only inside bank container / localhost).
Nginx proxies /intro/*.json → this service.

Env:
  BANK_URL   default http://127.0.0.1:9012
  BANK_USER  default explorer
  BANK_PASS  or /root/bank-explorer-password.txt
  AMOUNT     default GOA:10
  LANDING_DIR default /var/www/bank-landing
"""
from __future__ import annotations

import json
import os
import re
import secrets
import ssl
import string
import time
import urllib.error
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

BANK = os.environ.get("BANK_URL", "http://127.0.0.1:9012").rstrip("/")
# Public HTTPS base (Caddy) — used for webui links and x-taler-bank payto host:port
BANK_PUBLIC = os.environ.get("BANK_PUBLIC", "https://bank.hacktivism.ch").rstrip("/")
USER = os.environ.get("BANK_USER", "explorer")
AMOUNT = os.environ.get("AMOUNT", "GOA:10")
LANDING = Path(os.environ.get("LANDING_DIR", "/var/www/bank-landing"))
LISTEN = ("127.0.0.1", int(os.environ.get("DEMO_WITHDRAW_PORT", "19096")))


def bank_payto_host() -> str:
    """Hostname for payto://x-taler-bank/HOST/account (libeufin style: no :443)."""
    override = os.environ.get("BANK_PAYTO_HOST", "").strip()
    if override:
        # allow full host or host:port if operator sets it
        return override
    u = urllib.parse.urlparse(
        BANK_PUBLIC if "://" in BANK_PUBLIC else f"https://{BANK_PUBLIC}"
    )
    return u.hostname or "bank.hacktivism.ch"


def make_account_payto(username: str, bank_payto: str = "") -> str:
    """Account payto for QR/link.

    Prefer libeufin internal_payto_uri when present (authoritative).
    Fallback matches bank registration: host without default HTTPS port.
    """
    rn = urllib.parse.quote(username, safe="")
    if isinstance(bank_payto, str) and bank_payto.startswith("payto://x-taler-bank/"):
        # Keep bank-issued host/account; ensure receiver-name is encoded
        try:
            # payto://x-taler-bank/HOST/ACCOUNT?…
            rest = bank_payto[len("payto://x-taler-bank/") :]
            host_acct, _, _qs = rest.partition("?")
            host, _, acct = host_acct.partition("/")
            if host and acct:
                return f"payto://x-taler-bank/{host}/{acct}?receiver-name={rn}"
        except Exception:
            pass
    host = bank_payto_host()
    return f"payto://x-taler-bank/{host}/{username}?receiver-name={rn}"


def public_webui_url() -> str:
    return f"{BANK_PUBLIC}/webui/"


def load_pass() -> str:
    p = os.environ.get("BANK_PASS", "").strip()
    if p:
        return p
    for f in (
        Path(f"/root/bank-{USER}-password.txt"),
        Path("/root/bank-explorer-password.txt"),
    ):
        if f.is_file():
            return f.read_text().strip()
    raise RuntimeError("no BANK_PASS / explorer password file")


def http_json(method: str, url: str, body=None, headers=None, auth=None):
    data = None if body is None else json.dumps(body).encode()
    h = dict(headers or {})
    if body is not None:
        h["Content-Type"] = "application/json"
    if auth:
        import base64

        token = base64.b64encode(f"{auth[0]}:{auth[1]}".encode()).decode()
        h["Authorization"] = f"Basic {token}"
    req = urllib.request.Request(url, data=data, method=method, headers=h)
    ctx = ssl.create_default_context()
    try:
        with urllib.request.urlopen(req, context=ctx, timeout=20) as r:
            raw = r.read().decode()
            return r.status, json.loads(raw) if raw.strip() else {}
    except urllib.error.HTTPError as e:
        raw = e.read().decode()
        try:
            return e.code, json.loads(raw)
        except Exception:
            return e.code, {"raw": raw[:500]}


def mint_withdraw() -> dict:
    pw = load_pass()
    code, tok = http_json(
        "POST",
        f"{BANK}/accounts/{USER}/token",
        {"scope": "readwrite", "refreshable": True},
        auth=(USER, pw),
    )
    if code != 200 or not tok.get("access_token"):
        raise RuntimeError(f"token failed HTTP {code}: {tok}")
    access = tok["access_token"]
    code, wd = http_json(
        "POST",
        f"{BANK}/accounts/{USER}/withdrawals",
        {"suggested_amount": AMOUNT},
        headers={"Authorization": f"Bearer {access}"},
    )
    if code not in (200, 201):
        raise RuntimeError(f"withdrawal create HTTP {code}: {wd}")
    uri = wd.get("taler_withdraw_uri") or ""
    wid = wd.get("withdrawal_id") or ""
    if not uri:
        raise RuntimeError(f"no taler_withdraw_uri: {wd}")
    if not wid:
        wid = uri.rstrip("/").split("/")[-1]
    # Keep host:port from libeufin (e.g. bank.hacktivism.ch:443). Stripping :443
    # breaks taler-integration withdraw links / main landing QR on HTTPS banks.
    uri = str(uri).strip()
    LANDING.mkdir(parents=True, exist_ok=True)
    (LANDING / "withdraw.uri").write_text(uri + "\n")
    (LANDING / "withdraw.amount").write_text(AMOUNT + "\n")
    (LANDING / "withdraw.created").write_text(
        time.strftime("%Y-%m-%dT%H:%MZ", time.gmtime()) + "\n"
    )
    watch = LANDING / "withdraw-watch.ids"
    ids = set()
    if watch.is_file():
        ids = {ln.strip() for ln in watch.read_text().splitlines() if ln.strip()}
    ids.add(wid)
    watch.write_text("\n".join(sorted(ids)) + "\n")
    return {
        "ok": True,
        "taler_withdraw_uri": uri,
        "withdrawal_id": wid,
        "amount": AMOUNT,
        "pool_account": USER,
        "taler_integration_base": f"{BANK_PUBLIC}/taler-integration/",
        "hint": "Open in GNU Taler Wallet (iOS/Android/desktop). No bank registration.",
        "created": time.strftime("%Y-%m-%dT%H:%MZ", time.gmtime()),
    }


# Funny stems for usernames (bank-safe [a-z0-9-]).
# Source: hand-curated in this file only — not scraped from the web.
# Keep culture-neutral: light space / physics wordplay, no animals, foods,
# body parts, religion, politics, or slang that can offend.
_FUNNY_STEMS = (
    "nebula-nudge",
    "orbit-echo",
    "voidwave-vibe",
    "comet-crumb",
    "quark-pulse",
    "plasma-spark",
    "astro-glint",
    "lunar-loop",
    "warp-ripple",
    "photon-bloom",
    "galaxy-drift",
    "rocket-ribbon",
    "satellite-swirl",
    "meteor-mint",
    "stardust-swirl",
    "hyperdrive-hum",
    "cosmic-coral",
    "space-spark",
    "nova-nibble",
    "aurora-arc",
    "solar-swish",
    "pulsar-pop",
    "comet-cloud",
    "orbit-opal",
    "zenith-zip",
    "eclipse-echo",
    "horizon-hum",
    "starlight-step",
)


def _rand_username_and_name() -> tuple[str, str]:
    # Shown as goa-account-<funnypiece>-<tag> (e.g. goa-account-space-potato-k3m9x)
    stem = secrets.choice(_FUNNY_STEMS)
    alphabet = string.ascii_lowercase + string.digits
    tag = "".join(secrets.choice(alphabet) for _ in range(5))
    username = f"goa-account-{stem}-{tag}"
    name = username
    return username, name


def _rand_password() -> str:
    # Embed "pleasechangeme" with random material before and after.
    alphabet = "ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz23456789"
    prefix = "".join(secrets.choice(alphabet) for _ in range(4))
    suffix = "".join(secrets.choice(alphabet) for _ in range(6))
    return prefix + "pleasechangeme" + suffix


def create_personal_account() -> dict:
    """Public registration: auto username/password, balance starts at 0."""
    username, name = _rand_username_and_name()
    password = _rand_password()
    code, body = http_json(
        "POST",
        f"{BANK}/accounts",
        {
            "username": username,
            "password": password,
            "name": name,
        },
    )
    if code not in (200, 201, 204):
        # retry once on conflict
        if code in (409, 400):
            username, name = _rand_username_and_name()
            code, body = http_json(
                "POST",
                f"{BANK}/accounts",
                {
                    "username": username,
                    "password": password,
                    "name": name,
                },
            )
        if code not in (200, 201, 204):
            raise RuntimeError(f"register failed HTTP {code}: {body}")
    bank_payto = ""
    if isinstance(body, dict):
        bank_payto = (
            body.get("internal_payto_uri") or body.get("payto_uri") or ""
        )
    # Prefer bank-issued payto (no fake :443). QR encodes this account payto.
    payto = make_account_payto(username, bank_payto)
    webui = public_webui_url()
    return {
        "ok": True,
        "created_for_you": True,
        "username": username,
        "password": password,
        "name": name,
        "display_name": name,
        "balance": "GOA:0",
        "balance_note": "Starts at zero — not the shared community pool.",
        "payto_uri": payto,
        "qr_payload": payto,
        "webui": webui,
        "account_url": webui,
        "hint": (
            "This bank account was created for you automatically "
            f"({username}). "
            "Copy username and password now — we do not store them for later recovery. "
            "QR / blue link is the account payto:// (not the bank UI homepage)."
        ),
        "created": time.strftime("%Y-%m-%dT%H:%MZ", time.gmtime()),
        "created_human": time.strftime("%Y-%m-%d %H:%M %Z", time.localtime()),
    }


class Handler(BaseHTTPRequestHandler):
    def log_message(self, fmt, *args):
        sys_stderr = __import__("sys").stderr
        sys_stderr.write("%s - %s\n" % (self.address_string(), fmt % args))

    def _cors(self):
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS")
        self.send_header("Cache-Control", "no-store")

    def do_OPTIONS(self):
        self.send_response(204)
        self._cors()
        self.end_headers()

    def do_GET(self):
        path = self.path.split("?", 1)[0]
        try:
            if path in (
                "/",
                "/demo-withdraw.json",
                "/intro/demo-withdraw.json",
            ):
                body = mint_withdraw()
            elif path in (
                "/auto-account.json",
                "/intro/auto-account.json",
            ):
                body = create_personal_account()
            else:
                self.send_response(404)
                self._cors()
                self.send_header("Content-Type", "application/json")
                self.end_headers()
                self.wfile.write(b'{"ok":false,"error":"not found"}')
                return
            raw = json.dumps(body).encode()
            self.send_response(200)
            self._cors()
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(raw)))
            self.end_headers()
            self.wfile.write(raw)
        except Exception as e:
            raw = json.dumps({"ok": False, "error": str(e)}).encode()
            self.send_response(500)
            self._cors()
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(raw)))
            self.end_headers()
            self.wfile.write(raw)


def main():
    httpd = ThreadingHTTPServer(LISTEN, Handler)
    print(f"demo-withdraw-api on http://{LISTEN[0]}:{LISTEN[1]}/", flush=True)
    httpd.serve_forever()


if __name__ == "__main__":
    main()