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.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) — absolute webui / login links
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 public_webui_url() -> str:
    return f"{BANK_PUBLIC}/webui/"


def normalize_taler_withdraw_uri(uri: str) -> str:
    """Strip default :443/:80 from taler://withdraw host (wallet base-URL fix).

    libeufin builds authority from https BASE_URL and includes port 443; Android /
    iOS wallets then fail host parsing or TLS. Path and id stay unchanged.
    """
    if not uri:
        return uri
    # taler://withdraw/bank.example:443/taler-integration/UUID
    uri = re.sub(
        r"(taler://withdraw/)([^/?#]+):443(?=/|$)",
        r"\1\2",
        uri,
    )
    uri = re.sub(
        r"(taler://withdraw/)([^/?#]+):80(?=/|$)",
        r"\1\2",
        uri,
    )
    return uri

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]
    # libeufin emits taler://withdraw/host:443/... from https BASE_URL; mobile
    # wallets and many desktop builds reject/mis-parse default port :443.
    uri = normalize_taler_withdraw_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_withdrawal(wid)
    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()),
    }


def watch_withdrawal(wid: str) -> None:
    """Queue withdrawal_id for auto-confirm loop (shared pool + personal)."""
    if not wid:
        return
    LANDING.mkdir(parents=True, exist_ok=True)
    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(str(wid).strip())
    watch.write_text("\n".join(sorted(ids)) + "\n")



# 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}")
    webui = public_webui_url()
    # Same shared-pool withdraw as step 2 (explorer + auto-confirm)
    wd = mint_withdraw()
    withdraw_uri = wd["taler_withdraw_uri"]
    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.",
        "taler_withdraw_uri": withdraw_uri,
        "withdrawal_id": wd["withdrawal_id"],
        "withdraw_amount": wd.get("amount") or AMOUNT,
        "pool_account": wd.get("pool_account") or USER,
        "qr_payload": withdraw_uri,
        "webui": webui,
        "account_url": webui,
        "login_url": webui,
        "hint": (
            f"Login at {webui} with username {username} and the password shown "
            "(not stored for recovery). Wallet QR is taler://withdraw/… from the "
            f"shared pool ({USER}), same as step 2."
        ),
        "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()