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_authority() -> str:
    """Host:port for payto://x-taler-bank/… (port required; public HTTPS → :443)."""
    override = os.environ.get("BANK_PAYTO_HOST", "").strip()
    if override:
        return override
    u = urllib.parse.urlparse(
        BANK_PUBLIC if "://" in BANK_PUBLIC else f"https://{BANK_PUBLIC}"
    )
    host = u.hostname or "bank.hacktivism.ch"
    if u.port:
        port = u.port
    elif (u.scheme or "https") == "https":
        port = 443
    else:
        port = 80
    return f"{host}:{port}"


def make_account_payto(username: str) -> str:
    """Valid x-taler-bank payto with explicit port + encoded receiver-name."""
    auth = bank_payto_authority()
    rn = urllib.parse.quote(username, safe="")
    # path account name is bank-safe [a-z0-9-]; keep unencoded
    return f"payto://x-taler-bank/{auth}/{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}")
    # Always emit a correct public payto (libeufin often omits :443 → invalid for HTTPS)
    payto = make_account_payto(username)
    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,
        "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."
        ),
        "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()