diff --git a/.gitignore b/.gitignore index 77b5a7d..461e583 100644 --- a/.gitignore +++ b/.gitignore @@ -34,6 +34,8 @@ infra/dev/backups/ *.sqlite.gz *.sqlite-wal *.sqlite-shm +# K0 spike (docs/spikes/k0): Enable Banking key, session and fetched transactions +/docs/spikes/k0/.local/ # Web client (web/): dependencies and build output web/node_modules/ web/dist/ diff --git a/docs/spikes/k0/README.md b/docs/spikes/k0/README.md new file mode 100644 index 0000000..036b6f2 --- /dev/null +++ b/docs/spikes/k0/README.md @@ -0,0 +1,41 @@ +# K0 spike: local Enable Banking test app + +A one-file local web app that connects a card account (SJ Prio Mastercard, SEB Kort Bank AB) through Enable Banking with BankID and shows what the API returns. Python 3 standard library only; the RS256 JWT is signed with `openssl`. + +## Run + +```bash +python3 docs/spikes/k0/app.py # serves http://localhost:8765/ and opens the browser +``` + +Options: `--port`, `--no-browser`, `--local-dir` (config and state folder), `--api-base` (a fake API for tests). + +## Where the key goes + +`docs/spikes/k0/.local/` (git-ignored, created on first run): + +* `.pem`: the private key the Enable Banking control panel downloaded at registration (the file name is the application id). +* `config.json` (optional): `{"application_id": "…", "key_file": "….pem"}` when the file name is not the id or there are several keys. +* `state.json`: written by the app. Holds the settings, the bank list, the session id, the account, the last two fetches and an aggregate fetch log. When you run it again it reuses the session, so you do not need BankID again until `valid_until`. + +## What each button tests + +| Section | Button | Call | What it shows | +|---|---|---|---| +| 1 Status | none | none | Application id, key, session with `valid_until` and days left, chosen account | +| 2 Banker | Lista banker / Välj bank | `GET /aspsps?country=SE` | All Swedish ASPSPs. SJ, SEB Kort and Eurocard are highlighted and sorted first. Per row: `beta`, `maximum_consent_validity`, `psu_types`, `auth_methods`, `required_psu_headers`. The chosen ASPSP's full JSON is shown below. The default is the one named "SJ Prio" | +| 3 Anslut | Anslut med BankID | `POST /auth` | Sets `valid_until` to now plus the ASPSP's maximum, capped at 180 days, less 5 min, with `psu_type: personal` and a random `state`. Opens the bank in a new tab, where you scan the QR with BankID | +| 4 Klistra in | Använd adressen | `POST /sessions` | Takes `code` and `state` from any pasted address and checks `state`. This is the main path: the redirect goes to `https://kvitt.suorra.duckdns.org/cards/callback`, which does not load, but the address bar still carries the code | +| 5 Konton | Kontrollera session / Välj konto | `GET /sessions/{id}` | Accounts with uid, name, product, currency and type. Identifiers are masked to the last 4 characters. If there is only one account (or one CARD account), it is chosen by default | +| 6 Hämta | Uppdatera | `GET /accounts/{uid}/transactions` | Sends PSU headers (user present): always `Psu-Ip-Address` and `Psu-User-Agent`, plus every header the ASPSP lists in `required_psu_headers`. Uses `date_from` = today − history days, optionally `strategy=longest`, and follows `continuation_key` | +| 6 Hämta | Bakgrundshämtning | same, without PSU headers | Tests the background path. A refusal is shown with its HTTP status and error JSON | +| 6 Hämta | Saldo | `GET /accounts/{uid}/balances` | Balances, fetched with PSU headers | +| 7 Köp | none | none | Purchases with both dates, text, signed amount, currency, direction, status, `entry_reference` and a "rå" toggle for the raw JSON. Compared with the previous fetch: new rows, changed rows (PDNG→BOOK) and rows that disappeared | +| 8 Analys | Kopiera analys | none | Aggregates only: counts, dates reached vs requested, field presence, non-SEK rows, `entry_reference` presence and stability, PDNG→BOOK matched by `entry_reference` vs by amount+date (±7 days), pages per fetch, and the fetch log | +| 9 Koppla från | Koppla från | `DELETE /sessions/{id}` | Clears the local session, accounts and transactions. Keeps the settings, the bank list and the aggregate log | + +The PSU IP address field is filled from `api.ipify.org` on the first Uppdatera if you leave it empty; you can also type an address. `http://localhost:8765/callback` is handled as well, but Enable Banking's control panel refuses `http` redirect URLs, so it is not the default. + +## Privacy + +stdout shows only method, path template and HTTP status. It never shows bodies, merchant texts, amounts or ids. Personal data stays in `.local/`. Copy only the Analys block into reports. diff --git a/docs/spikes/k0/app.py b/docs/spikes/k0/app.py new file mode 100644 index 0000000..963b273 --- /dev/null +++ b/docs/spikes/k0/app.py @@ -0,0 +1,1764 @@ +#!/usr/bin/env python3 +"""K0 spike: a local test app for Enable Banking (card account via BankID). + +Python 3 standard library only. The RS256 JWT is signed by shelling out to openssl. +Run: python3 docs/spikes/k0/app.py (serves http://localhost:8765/ and opens the browser) + +Config and state live in docs/spikes/k0/.local/ (git-ignored): + .pem the private key from the Enable Banking control panel + config.json optional: {"application_id": "...", "key_file": "...pem"} + state.json session, account, settings and the last fetched transactions + +stdout carries only method, path and status. Never response bodies, texts or amounts. +""" + +from __future__ import annotations + +import argparse +import base64 +import datetime as dt +import html +import json +import os +import re +import secrets +import shutil +import socket +import subprocess +import sys +import threading +import time +import urllib.error +import urllib.parse +import urllib.request +import webbrowser +from decimal import Decimal, InvalidOperation +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +HERE = os.path.dirname(os.path.abspath(__file__)) +DEFAULT_LOCAL_DIR = os.path.join(HERE, ".local") +DEFAULT_API_BASE = "https://api.enablebanking.com" +DEFAULT_PORT = 8765 +# The control panel rejects http://localhost ("uses unsupported scheme"), so the +# registered https URL is the default and the paste box is the main path. The +# localhost callback handler stays for a future sandbox or https tunnel. +REDIRECT_CHOICES = [ + "https://kvitt.suorra.duckdns.org/cards/callback", + "http://localhost:8765/callback", +] +PASTE_HINT = ( + "När BankID är klart hamnar du på en sida som inte laddar. " + "Kopiera hela adressen från adressfältet och klistra in den här." +) +HISTORY_CHOICES = [7, 30, 90, 180, 365, 730] +CONSENT_CAP_SECONDS = 180 * 24 * 3600 +CONSENT_FALLBACK_SECONDS = 90 * 24 * 3600 # used only when the ASPSP list is not loaded +MATCH_WINDOW_DAYS = 7 # spike heuristic for pending->booked matching by amount and date +MAX_PAGES = 200 +HIGHLIGHT = re.compile(r"\bSJ\b|SEB\s*Kort|Eurocard", re.IGNORECASE) +DEFAULT_ASPSP = re.compile(r"SJ\s*Prio", re.IGNORECASE) +UUID_PEM = re.compile( + r"^([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})\.pem$" +) +PSU_HEADER_CANON = { + "psu-ip-address": "Psu-Ip-Address", + "psu-user-agent": "Psu-User-Agent", + "psu-referer": "Psu-Referer", + "psu-accept": "Psu-Accept", + "psu-accept-charset": "Psu-Accept-Charset", + "psu-accept-encoding": "Psu-Accept-Encoding", + "psu-accept-language": "Psu-Accept-Language", + "psu-geo-location": "Psu-Geo-Location", +} + +LOCK = threading.RLock() +CFG = { + "local_dir": DEFAULT_LOCAL_DIR, + "api_base": DEFAULT_API_BASE, + "port": DEFAULT_PORT, + "open_browser": True, +} +LAST = {} # last API response or message shown on the page (in memory only) +JWT_CACHE = {"token": None, "exp": 0, "kid": None} + + +def log(msg): + print(time.strftime("%H:%M:%S"), msg, flush=True) + + +# ----------------------------------------------------------------------------- config and state + + +def local_path(name): + return os.path.join(CFG["local_dir"], name) + + +def read_json(path, default): + try: + with open(path, "r", encoding="utf-8") as f: + return json.load(f) + except FileNotFoundError: + return default + except (OSError, ValueError) as e: + log("could not read %s: %s" % (os.path.basename(path), type(e).__name__)) + return default + + +def load_config(): + """Return (application_id, key_path, problems).""" + problems = [] + cfg = read_json(local_path("config.json"), {}) + app_id = cfg.get("application_id") if isinstance(cfg, dict) else None + key_file = cfg.get("key_file") if isinstance(cfg, dict) else None + pems = [] + try: + pems = sorted(n for n in os.listdir(CFG["local_dir"]) if UUID_PEM.match(n)) + except FileNotFoundError: + problems.append("Mappen %s finns inte." % CFG["local_dir"]) + if not app_id: + if len(pems) == 1: + m = UUID_PEM.match(pems[0]) + app_id = m.group(1) if m else None + elif len(pems) > 1: + problems.append( + "Flera .pem i .local; ange application_id i config.json." + ) + if key_file: + key_path = key_file if os.path.isabs(key_file) else local_path(key_file) + elif app_id: + key_path = local_path(app_id + ".pem") + else: + key_path = None + if not app_id: + problems.append("Inget application id (lägg .pem i .local eller config.json).") + if key_path and not os.path.isfile(key_path): + problems.append("Nyckelfilen %s saknas." % os.path.basename(key_path)) + return app_id, key_path, problems + + +def default_state(): + return { + "settings": { + "aspsp_name": "", + "history_days": 90, + "redirect_url": REDIRECT_CHOICES[0], + "auth_method": "", + "strategy": "default", + "psu_ip": "", + }, + "aspsps": [], + "aspsps_fetched_at": None, + "pending_auth": None, + "session": None, + "account_uid": None, + "transactions": [], + "transactions_meta": None, + "previous_transactions": [], + "last_diff": None, + "balances": None, + "fetch_log": [], + } + + +def load_state(): + st = default_state() + saved = read_json(local_path("state.json"), {}) + if isinstance(saved, dict): + for k, v in saved.items(): + if k == "settings" and isinstance(v, dict): + st["settings"].update(v) + else: + st[k] = v + return st + + +def save_state(st): + os.makedirs(CFG["local_dir"], exist_ok=True) + try: + os.chmod(CFG["local_dir"], 0o700) + except OSError as e: + log("chmod .local failed: %s" % type(e).__name__) + path = local_path("state.json") + tmp = path + ".tmp" + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(st, f, ensure_ascii=False, indent=1) + os.replace(tmp, path) + + +# ----------------------------------------------------------------------------- JWT + + +def b64url(raw): + return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") + + +def openssl_bin(): + return shutil.which("openssl") or "/usr/bin/openssl" + + +def make_jwt(app_id, key_path, now=None): + iat = int(now if now is not None else time.time()) + header = {"typ": "JWT", "alg": "RS256", "kid": app_id} + payload = { + "iss": "enablebanking.com", + "aud": "api.enablebanking.com", + "iat": iat, + "exp": iat + 3600, + } + signing_input = ( + b64url(json.dumps(header, separators=(",", ":")).encode()) + + "." + + b64url(json.dumps(payload, separators=(",", ":")).encode()) + ) + proc = subprocess.run( + [openssl_bin(), "dgst", "-sha256", "-sign", key_path], + input=signing_input.encode("ascii"), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + ) + if proc.returncode != 0 or not proc.stdout: + raise RuntimeError( + "openssl kunde inte signera (exit %d): %s" + % (proc.returncode, proc.stderr.decode("utf-8", "replace").strip()[:300]) + ) + return signing_input + "." + b64url(proc.stdout), payload["exp"] + + +def get_jwt(): + app_id, key_path, problems = load_config() + if not app_id or not key_path or not os.path.isfile(key_path): + raise RuntimeError("; ".join(problems) or "Nyckel saknas.") + now = time.time() + if JWT_CACHE["token"] and JWT_CACHE["kid"] == app_id and JWT_CACHE["exp"] - now > 300: + return JWT_CACHE["token"] + token, exp = make_jwt(app_id, key_path, now) + JWT_CACHE.update(token=token, exp=exp, kid=app_id) + return token + + +# ----------------------------------------------------------------------------- HTTP client + + +class ApiResult(object): + def __init__(self, status, data, raw_text, error=None): + self.status = status + self.data = data + self.raw_text = raw_text + self.error = error + + @property + def ok(self): + return 200 <= self.status < 300 and self.error is None + + +def api(method, path, route, query=None, body=None, headers=None): + """Call Enable Banking. `route` is the path template that is logged (no ids).""" + try: + token = get_jwt() + except (RuntimeError, OSError) as e: + return ApiResult(0, None, "", error=str(e)) + url = CFG["api_base"].rstrip("/") + path + if query: + url += "?" + urllib.parse.urlencode(query) + data = json.dumps(body).encode("utf-8") if body is not None else None + req = urllib.request.Request(url, data=data, method=method) + req.add_header("Authorization", "Bearer " + token) + req.add_header("Accept", "application/json") + if data is not None: + req.add_header("Content-Type", "application/json") + for k, v in (headers or {}).items(): + req.add_header(k, v) + status = 0 + raw = b"" + try: + with urllib.request.urlopen(req, timeout=90) as resp: + status = resp.status + raw = resp.read() + except urllib.error.HTTPError as e: + status = e.code + try: + raw = e.read() + except OSError: + raw = b"" + except (urllib.error.URLError, OSError) as e: + log("EB %s %s network-error" % (method, route)) + reason = getattr(e, "reason", e) + return ApiResult(0, None, "", error="Nätverksfel: %s" % reason) + log("EB %s %s %d" % (method, route, status)) + text = raw.decode("utf-8", "replace") + try: + parsed = json.loads(text) if text.strip() else None + except ValueError: + parsed = None + return ApiResult(status, parsed, text) + + +def set_last(kind, title, result=None, message=None, section="top"): + entry = {"kind": kind, "title": title, "section": section, "at": now_iso()} + if result is not None: + entry["status"] = result.status + if result.error: + entry["body"] = result.error + elif result.data is not None: + entry["body"] = json.dumps(result.data, ensure_ascii=False, indent=2) + else: + entry["body"] = result.raw_text[:4000] + if message: + entry["message"] = message + LAST.clear() + LAST.update(entry) + + +def fail(title, result, section): + set_last("error", title, result=result, section=section) + + +def now_iso(): + return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat() + + +def public_ip(): + try: + with urllib.request.urlopen("https://api.ipify.org?format=text", timeout=10) as r: + ip = r.read().decode("ascii", "replace").strip() + log("ipify GET / %d" % r.status) + if re.match(r"^[0-9a-fA-F:.]+$", ip): + return ip + except (urllib.error.URLError, OSError): + log("ipify GET / network-error") + return None + + +# ----------------------------------------------------------------------------- helpers on data + + +def chosen_aspsp(st): + name = st["settings"].get("aspsp_name") + for a in st.get("aspsps") or []: + if a.get("name") == name: + return a + return None + + +def default_aspsp_name(aspsps): + for a in aspsps: + if DEFAULT_ASPSP.search(a.get("name") or ""): + return a.get("name") + return "" + + +def mask(value): + s = str(value) + if len(s) <= 4: + return "••••" + return "••••" + s[-4:] + + +def mask_ids(obj): + """Mask every string leaf of an identifier structure to its last 4 characters, + keeping scheme names readable.""" + if isinstance(obj, dict): + out = {} + for k, v in obj.items(): + if k in ("scheme_name", "issuer") and isinstance(v, str): + out[k] = v + else: + out[k] = mask_ids(v) + return out + if isinstance(obj, list): + return [mask_ids(v) for v in obj] + if obj is None: + return None + return mask(obj) + + +def amount_of(t): + ta = t.get("transaction_amount") or {} + try: + val = abs(Decimal(str(ta.get("amount")))) + except (InvalidOperation, ValueError, TypeError): + return None + if t.get("credit_debit_indicator") == "DBIT": + val = -val + return val + + +def currency_of(t): + return (t.get("transaction_amount") or {}).get("currency") or "" + + +def fmt_kr(val): + if val is None: + return "?" + sign = "−" if val < 0 else "+" + s = "{:,.2f}".format(abs(val)).replace(",", " ").replace(".", ",") + return sign + s + + +def text_of(t): + parts = [] + party = t.get("creditor") if t.get("credit_debit_indicator") == "DBIT" else t.get("debtor") + if isinstance(party, dict) and party.get("name"): + parts.append(str(party["name"])) + ri = t.get("remittance_information") + if isinstance(ri, list): + parts.extend(str(x) for x in ri if x) + elif ri: + parts.append(str(ri)) + if t.get("note"): + parts.append(str(t["note"])) + return " / ".join(parts) + + +def date_of(t): + return t.get("transaction_date") or t.get("booking_date") or t.get("value_date") + + +def parse_date(s): + try: + return dt.date.fromisoformat(str(s)[:10]) + except (ValueError, TypeError): + return None + + +def present(v): + return v is not None and v != "" and v != [] and v != {} + + +def flatten_presence(obj, prefix, out): + for k, v in obj.items(): + p = prefix + k + if isinstance(v, dict) and v: + out[p] = True + flatten_presence(v, p + ".", out) + else: + out[p] = present(v) + + +def field_presence(txs): + seen = {} + for t in txs: + flat = {} + flatten_presence(t, "", flat) + for k, v in flat.items(): + seen.setdefault(k, 0) + if v: + seen[k] += 1 + return seen + + +# ----------------------------------------------------------------------------- diff + + +def _er(t): + v = t.get("entry_reference") + return str(v) if present(v) else None + + +def diff_transactions(prev, curr): + """Match the previous fetch to the current one. + + 1. equal entry_reference; + 2a. same amount, direction, currency, date and text (entry_reference ignored); + 2b. same amount, direction and currency, date within MATCH_WINDOW_DAYS, only + when at least one side has no entry_reference (closest date wins). + """ + matched_prev = {} + matched_curr = {} + by = {} + prev_er = {} + for i, t in enumerate(prev): + er = _er(t) + if er is not None: + prev_er.setdefault(er, []).append(i) + for j, t in enumerate(curr): + er = _er(t) + if er is None: + continue + for i in prev_er.get(er, []): + if i not in matched_prev: + matched_prev[i] = j + matched_curr[j] = i + by[j] = "entry_reference" + break + + def fp(t, with_text): + return ( + str(amount_of(t)), + t.get("credit_debit_indicator"), + currency_of(t), + date_of(t) if with_text else None, + text_of(t) if with_text else None, + ) + + strict = {} + for i, t in enumerate(prev): + if i not in matched_prev: + strict.setdefault(fp(t, True), []).append(i) + for j, t in enumerate(curr): + if j in matched_curr: + continue + for i in strict.get(fp(t, True), []): + if i not in matched_prev: + matched_prev[i] = j + matched_curr[j] = i + by[j] = "amount_date" + break + + for j, t in enumerate(curr): + if j in matched_curr: + continue + dj = parse_date(date_of(t)) + best = None + for i, p in enumerate(prev): + if i in matched_prev: + continue + if fp(p, False) != fp(t, False): + continue + if _er(p) is not None and _er(t) is not None: + continue + di = parse_date(date_of(p)) + if dj is None or di is None: + continue + gap = abs((dj - di).days) + if gap > MATCH_WINDOW_DAYS: + continue + if best is None or gap < best[0]: + best = (gap, i) + if best is not None: + i = best[1] + matched_prev[i] = j + matched_curr[j] = i + by[j] = "amount_window" + + compare = [ + "status", + "booking_date", + "value_date", + "transaction_date", + "entry_reference", + "transaction_id", + ] + changed = [] + stats = { + "pdng_to_book_by_entry_reference": 0, + "pdng_to_book_by_amount_date": 0, + "er_same": 0, + "er_changed": 0, + "er_appeared": 0, + "er_prev_unmatched": 0, + } + for j, i in sorted(matched_curr.items()): + p, c = prev[i], curr[j] + fields = [f for f in compare if p.get(f) != c.get(f)] + if text_of(p) != text_of(c): + fields.append("text") + if str(amount_of(p)) != str(amount_of(c)): + fields.append("amount") + if _er(p) is not None and _er(c) is not None: + if _er(p) == _er(c): + stats["er_same"] += 1 + else: + stats["er_changed"] += 1 + elif _er(p) is None and _er(c) is not None: + stats["er_appeared"] += 1 + if p.get("status") == "PDNG" and c.get("status") == "BOOK": + if by[j] == "entry_reference": + stats["pdng_to_book_by_entry_reference"] += 1 + else: + stats["pdng_to_book_by_amount_date"] += 1 + if fields: + changed.append({"idx": j, "prev_idx": i, "fields": fields, "by": by[j]}) + new = [j for j in range(len(curr)) if j not in matched_curr] + gone = [i for i in range(len(prev)) if i not in matched_prev] + stats["er_prev_unmatched"] = sum(1 for i in gone if _er(prev[i]) is not None) + return { + "new": new, + "changed": changed, + "disappeared": gone, + "matched_by": {str(j): by[j] for j in by}, + "stats": stats, + "had_previous": bool(prev), + } + + +# ----------------------------------------------------------------------------- actions + + +def action_list_aspsps(st): + res = api("GET", "/aspsps", "/aspsps", query={"country": "SE"}) + if not res.ok or not isinstance(res.data, dict): + fail("Lista banker misslyckades", res, "banker") + return + aspsps = res.data.get("aspsps") or [] + for a in aspsps: + a.pop("logo", None) + st["aspsps"] = aspsps + st["aspsps_fetched_at"] = now_iso() + if not chosen_aspsp(st): + st["settings"]["aspsp_name"] = default_aspsp_name(aspsps) + set_last("ok", "Lista banker", message="%d banker i SE." % len(aspsps), section="banker") + + +def action_choose_aspsp(st, form): + name = form.get("aspsp", "") + if not any(a.get("name") == name for a in st.get("aspsps") or []): + set_last("error", "Välj bank", message="Okänd bank.", section="banker") + return + st["settings"]["aspsp_name"] = name + st["settings"]["auth_method"] = "" + set_last("ok", "Välj bank", message="Vald: %s" % name, section="banker") + + +def save_settings(st, form): + s = st["settings"] + try: + days = int(form.get("history_days", s["history_days"])) + if days in HISTORY_CHOICES: + s["history_days"] = days + except ValueError: + pass + choice = form.get("redirect_choice", "") + custom = form.get("redirect_custom", "").strip() + if choice == "custom" and custom: + s["redirect_url"] = custom + elif choice in REDIRECT_CHOICES: + s["redirect_url"] = choice + s["auth_method"] = form.get("auth_method", "") + if form.get("strategy") in ("default", "longest"): + s["strategy"] = form["strategy"] + + +def consent_seconds(aspsp): + if aspsp and isinstance(aspsp.get("maximum_consent_validity"), (int, float)): + return min(int(aspsp["maximum_consent_validity"]), CONSENT_CAP_SECONDS) + return CONSENT_FALLBACK_SECONDS + + +def build_auth_body(st, state_token, now=None): + s = st["settings"] + aspsp = chosen_aspsp(st) + now = now or dt.datetime.now(dt.timezone.utc) + # a few minutes' margin so the request never exceeds the ASPSP's maximum + valid_until = now + dt.timedelta(seconds=consent_seconds(aspsp) - 300) + body = { + "access": {"valid_until": valid_until.replace(microsecond=0).isoformat()}, + "aspsp": {"name": s["aspsp_name"], "country": "SE"}, + "state": state_token, + "redirect_url": s["redirect_url"], + "psu_type": "personal", + } + if s.get("auth_method"): + body["auth_method"] = s["auth_method"] + return body + + +def action_auth(st): + if not st["settings"].get("aspsp_name"): + set_last("error", "Anslut", message="Välj en bank först (Lista banker).", section="anslut") + return None + token = secrets.token_urlsafe(24) + body = build_auth_body(st, token) + res = api("POST", "/auth", "/auth", body=body) + if not res.ok or not isinstance(res.data, dict) or not res.data.get("url"): + fail("POST /auth misslyckades", res, "anslut") + return None + st["pending_auth"] = { + "state": token, + "aspsp": body["aspsp"], + "valid_until": body["access"]["valid_until"], + "redirect_url": body["redirect_url"], + "started_at": now_iso(), + "authorization_id": res.data.get("authorization_id"), + "url": res.data["url"], + } + set_last( + "ok", + "POST /auth (HTTP %d)" % res.status, + message="Bankens sida öppnas i en ny flik. Scanna BankID-QR-koden med telefonen. " + PASTE_HINT, + section="anslut", + ) + return res.data["url"] + + +def extract_code_state(text): + """Pull code, state, error and error_description out of any pasted URL or query.""" + text = (text or "").strip() + found = {} + candidates = [] + parsed = urllib.parse.urlparse(text) + candidates.append(parsed.query) + candidates.append(parsed.fragment) + if "?" in text: + candidates.append(text.split("?", 1)[1].split("#", 1)[0]) + candidates.append(text.lstrip("?#")) + for c in candidates: + if not c: + continue + qs = urllib.parse.parse_qs(c, keep_blank_values=False) + for key in ("code", "state", "error", "error_description"): + if key not in found and qs.get(key): + found[key] = qs[key][0] + return found + + +def complete_auth(st, params, source): + if params.get("error"): + set_last( + "error", + "Banken svarade med fel (%s)" % source, + message="%s: %s" % (params.get("error"), params.get("error_description", "")), + section="anslut", + ) + return + code = params.get("code") + if not code: + set_last("error", "Ingen code", message="Hittade ingen code i adressen.", section="anslut") + return + pending = st.get("pending_auth") or {} + if pending.get("state") and params.get("state") != pending.get("state"): + set_last( + "error", + "state stämmer inte", + message="state i adressen matchar inte den senaste anslutningen. Starta om Anslut med BankID.", + section="anslut", + ) + return + res = api("POST", "/sessions", "/sessions", body={"code": code}) + if not res.ok or not isinstance(res.data, dict): + fail("POST /sessions misslyckades", res, "anslut") + return + d = res.data + accounts = d.get("accounts") or [] + st["session"] = { + "session_id": d.get("session_id"), + "valid_until": (d.get("access") or {}).get("valid_until"), + "aspsp": d.get("aspsp"), + "psu_type": d.get("psu_type"), + "accounts": accounts, + "created_at": now_iso(), + } + st["pending_auth"] = None + st["account_uid"] = default_account_uid(accounts) + st["balances"] = None + set_last( + "ok", + "Session skapad (%s)" % source, + message="%d konto(n). Giltig till %s." % (len(accounts), st["session"]["valid_until"]), + section="konton", + ) + + +def default_account_uid(accounts): + if len(accounts) == 1: + return accounts[0].get("uid") + for a in accounts: + if (a.get("cash_account_type") or "").upper() == "CARD": + return a.get("uid") + return None + + +def action_choose_account(st, form): + uid = form.get("account_uid", "") + accounts = (st.get("session") or {}).get("accounts") or [] + if not any(a.get("uid") == uid for a in accounts): + set_last("error", "Välj konto", message="Okänt konto.", section="konton") + return + st["account_uid"] = uid + set_last("ok", "Välj konto", message="Konto valt.", section="konton") + + +def action_check_session(st): + sess = st.get("session") or {} + if not sess.get("session_id"): + set_last("error", "Session", message="Ingen session.", section="konton") + return + res = api( + "GET", + "/sessions/" + urllib.parse.quote(sess["session_id"], safe=""), + "/sessions/{id}", + ) + if not res.ok: + fail("GET /sessions/{id} misslyckades", res, "konton") + return + data = res.data if isinstance(res.data, dict) else {} + summary = {k: data.get(k) for k in ("status", "authorized", "created", "access", "psu_type")} + set_last( + "ok", + "GET /sessions/{id}", + message=json.dumps(summary, ensure_ascii=False), + section="konton", + ) + + +def psu_headers(st, browser): + """Headers that tell Enable Banking the user is present. + + Always Psu-Ip-Address and Psu-User-Agent, plus every header the ASPSP lists in + required_psu_headers that the browser request lets us fill.""" + s = st["settings"] + ip = (s.get("psu_ip") or "").strip() + note = [] + if not ip: + ip = public_ip() or "" + if ip: + s["psu_ip"] = ip + note.append("PSU-IP hämtad från ipify") + if not ip: + ip = browser.get("client_ip") or "127.0.0.1" + note.append("PSU-IP: använder %s (ingen publik IP hittad)" % ip) + values = { + "psu-ip-address": ip, + "psu-user-agent": browser.get("user_agent") or "Mozilla/5.0", + "psu-referer": "http://localhost:%d/" % CFG["port"], + "psu-accept": browser.get("accept") or "text/html", + "psu-accept-charset": browser.get("accept_charset") or "utf-8", + "psu-accept-encoding": browser.get("accept_encoding") or "gzip, deflate", + "psu-accept-language": browser.get("accept_language") or "sv-SE,sv;q=0.9", + } + wanted = ["psu-ip-address", "psu-user-agent"] + aspsp = chosen_aspsp(st) + for h in (aspsp or {}).get("required_psu_headers") or []: + lh = str(h).lower() + if lh not in wanted: + wanted.append(lh) + headers = {} + for lh in wanted: + if lh in values: + headers[PSU_HEADER_CANON.get(lh, lh)] = values[lh] + else: + note.append("kan inte fylla %s" % lh) + return headers, note + + +def error_code(res): + if isinstance(res.data, dict): + return res.data.get("error") or res.data.get("code") + return res.error and "network" + + +def action_fetch(st, mode, browser): + uid = st.get("account_uid") + if not uid: + set_last("error", "Hämta", message="Inget konto valt.", section="hamta") + return + s = st["settings"] + date_from = (dt.date.today() - dt.timedelta(days=int(s["history_days"]))).isoformat() + query = {"date_from": date_from} + if s.get("strategy") == "longest": + query["strategy"] = "longest" + headers, notes = psu_headers(st, browser) if mode == "user" else ({}, []) + txs = [] + pages = 0 + empty_pages = 0 + key = None + truncated = False + res = None + path = "/accounts/%s/transactions" % urllib.parse.quote(uid, safe="") + while True: + q = dict(query) + if key: + q["continuation_key"] = key + res = api("GET", path, "/accounts/{uid}/transactions", query=q, headers=headers) + pages += 1 + if not res.ok or not isinstance(res.data, dict): + break + page = res.data.get("transactions") or [] + if not page: + empty_pages += 1 + txs.extend(page) + key = res.data.get("continuation_key") + if not key: + break + if pages >= MAX_PAGES: + truncated = True + break + entry = { + "at": now_iso(), + "mode": mode, + "strategy": s.get("strategy"), + "history_days": s["history_days"], + "date_from": date_from, + "psu_headers": sorted(headers.keys()), + "status": res.status if res else 0, + "pages": pages, + "empty_pages": empty_pages, + "truncated": truncated, + } + if res is None or not res.ok: + entry["error"] = error_code(res) if res else "none" + entry["ok"] = False + st["fetch_log"].append(entry) + label = "Uppdatera" if mode == "user" else "Bakgrundshämtning" + fail("%s: sida %d misslyckades" % (label, pages), res, "hamta") + if notes: + LAST["message"] = "; ".join(notes) + return + prev = st.get("transactions") or [] + d = diff_transactions(prev, txs) + dates = [parse_date(date_of(t)) for t in txs] + dates = [x for x in dates if x] + entry.update( + { + "ok": True, + "count": len(txs), + "oldest": min(dates).isoformat() if dates else None, + "newest": max(dates).isoformat() if dates else None, + "new": len(d["new"]), + "changed": len(d["changed"]), + "disappeared": len(d["disappeared"]), + "had_previous": d["had_previous"], + } + ) + entry.update(d["stats"]) + st["fetch_log"].append(entry) + st["previous_transactions"] = prev + st["transactions"] = txs + st["transactions_meta"] = { + "at": entry["at"], + "mode": mode, + "date_from": date_from, + "pages": pages, + "strategy": s.get("strategy"), + } + st["last_diff"] = d + msg = "%d transaktioner, %d sidor. Nya %d, ändrade %d, borta %d." % ( + len(txs), + pages, + len(d["new"]), + len(d["changed"]), + len(d["disappeared"]), + ) + if notes: + msg += " (" + "; ".join(notes) + ")" + if truncated: + msg += " Avbröt efter %d sidor." % MAX_PAGES + label = "Uppdatera" if mode == "user" else "Bakgrundshämtning" + set_last("ok", label + " (HTTP %d)" % res.status, message=msg, section="kop") + + +def action_balances(st, browser): + uid = st.get("account_uid") + if not uid: + set_last("error", "Saldo", message="Inget konto valt.", section="hamta") + return + headers, notes = psu_headers(st, browser) + res = api( + "GET", + "/accounts/%s/balances" % urllib.parse.quote(uid, safe=""), + "/accounts/{uid}/balances", + headers=headers, + ) + entry = { + "at": now_iso(), + "mode": "balances", + "status": res.status, + "ok": res.ok, + "psu_headers": sorted(headers.keys()), + } + if not res.ok or not isinstance(res.data, dict): + entry["error"] = error_code(res) + st["fetch_log"].append(entry) + fail("Saldo misslyckades", res, "hamta") + return + bal = res.data.get("balances") or [] + entry["balance_types"] = sorted({str(b.get("balance_type")) for b in bal}) + st["fetch_log"].append(entry) + st["balances"] = {"at": entry["at"], "balances": bal} + set_last("ok", "Saldo (HTTP %d)" % res.status, message="%d saldon." % len(bal), section="hamta") + if notes: + LAST["message"] += " (" + "; ".join(notes) + ")" + + +def action_disconnect(st): + sess = st.get("session") or {} + msg = "Lokalt tillstånd rensat." + if sess.get("session_id"): + res = api( + "DELETE", + "/sessions/" + urllib.parse.quote(sess["session_id"], safe=""), + "/sessions/{id}", + ) + if not res.ok: + fail("DELETE /sessions/{id} misslyckades (lokalt tillstånd rensat ändå)", res, "koppla") + else: + set_last("ok", "Koppla från (HTTP %d)" % res.status, message=msg, section="koppla") + else: + set_last("ok", "Koppla från", message=msg, section="koppla") + # keep settings, the ASPSP list and the aggregate fetch log (no personal content) + keep = { + "settings": st["settings"], + "aspsps": st.get("aspsps") or [], + "aspsps_fetched_at": st.get("aspsps_fetched_at"), + "fetch_log": st.get("fetch_log") or [], + } + st.clear() + st.update(default_state()) + st.update(keep) + st["fetch_log"].append({"at": now_iso(), "mode": "disconnect", "ok": True}) + + +# ----------------------------------------------------------------------------- analysis + + +def pct(n, d): + return "%.1f" % (100.0 * n / d) if d else "-" + + +def analysis_text(st): + lines = [] + txs = st.get("transactions") or [] + meta = st.get("transactions_meta") or {} + aspsp = chosen_aspsp(st) or {} + sess = st.get("session") or {} + lines.append("K0 analys, genererad %s" % now_iso()) + if aspsp: + mcv = aspsp.get("maximum_consent_validity") + lines.append( + "ASPSP: %s (%s), beta %s, maximum_consent_validity %s s (%s d), psu_types %s" + % ( + aspsp.get("name"), + aspsp.get("country"), + aspsp.get("beta"), + mcv, + round(mcv / 86400.0, 1) if isinstance(mcv, (int, float)) else "?", + ",".join(aspsp.get("psu_types") or []), + ) + ) + lines.append( + "auth_methods: %s; required_psu_headers: %s" + % ( + "; ".join( + "%s/%s/%s" % (m.get("name"), m.get("approach"), m.get("psu_type")) + for m in aspsp.get("auth_methods") or [] + ) + or "-", + ",".join(aspsp.get("required_psu_headers") or []) or "-", + ) + ) + else: + lines.append("ASPSP: inte vald") + if sess.get("session_id"): + accts = sess.get("accounts") or [] + lines.append( + "Session: giltig till %s; konton %d; valutor %s; cash_account_type %s" + % ( + (sess.get("valid_until") or "?")[:10], + len(accts), + ",".join(sorted({str(a.get("currency")) for a in accts})), + ",".join(sorted({str(a.get("cash_account_type")) for a in accts})), + ) + ) + else: + lines.append("Session: ingen") + lines.append("") + n = len(txs) + lines.append( + "Senaste lyckade hämtning: %s, läge %s, strategi %s, sidor %s" + % (meta.get("at", "-"), meta.get("mode", "-"), meta.get("strategy", "-"), meta.get("pages", "-")) + ) + lines.append("Transaktioner: %d" % n) + if n: + dates = sorted(x for x in (parse_date(date_of(t)) for t in txs) if x) + bdates = sorted(x for x in (parse_date(t.get("booking_date")) for t in txs) if x) + req = parse_date(meta.get("date_from")) + oldest = dates[0] if dates else None + lines.append( + "Begärt från %s (%s d); äldsta nådda (transaction/booking/value) %s; äldsta booking_date %s; nyaste %s; täckning %s d av begärda %s d" + % ( + meta.get("date_from"), + st["settings"].get("history_days"), + oldest, + bdates[0] if bdates else "-", + dates[-1] if dates else "-", + (dt.date.today() - oldest).days if oldest else "-", + (dt.date.today() - req).days if req else "-", + ) + ) + by_status = {} + by_dir = {} + for t in txs: + by_status[str(t.get("status"))] = by_status.get(str(t.get("status")), 0) + 1 + k = str(t.get("credit_debit_indicator")) + by_dir[k] = by_dir.get(k, 0) + 1 + lines.append("Status: " + ", ".join("%s %d" % kv for kv in sorted(by_status.items()))) + lines.append("Riktning: " + ", ".join("%s %d" % kv for kv in sorted(by_dir.items()))) + non_sek = [t for t in txs if currency_of(t) and currency_of(t) != "SEK"] + lines.append( + "Icke-SEK transaction_amount: %d (exchange_rate ifyllt i %d; valutor %s)" + % ( + len(non_sek), + sum(1 for t in non_sek if present(t.get("exchange_rate"))), + ",".join(sorted({currency_of(t) for t in non_sek})) or "-", + ) + ) + with_er = [t for t in txs if _er(t) is not None] + ers = [_er(t) for t in with_er] + er_by_status = {} + for t in txs: + s = str(t.get("status")) + tot, has = er_by_status.get(s, (0, 0)) + er_by_status[s] = (tot + 1, has + (1 if _er(t) is not None else 0)) + lines.append( + "entry_reference: finns i %d/%d (%s); unika %d; dubbletter %d" + % ( + len(with_er), + n, + ", ".join("%s %d/%d" % (s, h, tt) for s, (tt, h) in sorted(er_by_status.items())), + len(set(ers)), + len(ers) - len(set(ers)), + ) + ) + lines.append( + "transaction_id: finns i %d/%d" + % (sum(1 for t in txs if present(t.get("transaction_id"))), n) + ) + lines.append( + "Text tom (creditor/debtor-namn, remittance_information, note): %d/%d" + % (sum(1 for t in txs if not text_of(t)), n) + ) + lines.append("") + lines.append("Fältnärvaro (%% av %d rader, ifyllt):" % n) + for k, c in sorted(field_presence(txs).items()): + lines.append(" %-45s %6s%%" % (k, pct(c, n))) + log_entries = st.get("fetch_log") or [] + tx_fetches = [e for e in log_entries if e.get("mode") in ("user", "background")] + ok_fetches = [e for e in tx_fetches if e.get("ok")] + lines.append("") + lines.append( + "Hämtningar: %d transaktionsanrop (%d lyckade), %d saldoanrop" + % ( + len(tx_fetches), + len(ok_fetches), + sum(1 for e in log_entries if e.get("mode") == "balances"), + ) + ) + tot = { + "pdng_to_book_by_entry_reference": 0, + "pdng_to_book_by_amount_date": 0, + "er_same": 0, + "er_changed": 0, + "er_appeared": 0, + "er_prev_unmatched": 0, + } + for e in ok_fetches: + for k in tot: + tot[k] += int(e.get(k) or 0) + lines.append( + "PDNG->BOOK (summerat över hämtningar): via entry_reference %d, via belopp+datum (exakt eller ±%d d) %d" + % ( + tot["pdng_to_book_by_entry_reference"], + MATCH_WINDOW_DAYS, + tot["pdng_to_book_by_amount_date"], + ) + ) + lines.append( + "entry_reference mellan hämtningar: samma %d, ändrad %d, dök upp (saknades förut) %d, föregående med ER som försvann %d" + % (tot["er_same"], tot["er_changed"], tot["er_appeared"], tot["er_prev_unmatched"]) + ) + for mode in ("user", "background"): + last = [e for e in tx_fetches if e.get("mode") == mode] + if last: + e = last[-1] + lines.append( + "Senaste %s: HTTP %s%s, PSU-headers [%s]" + % ( + "Uppdatera" if mode == "user" else "Bakgrundshämtning", + e.get("status"), + (" " + str(e.get("error"))) if e.get("error") else "", + ",".join(e.get("psu_headers") or []), + ) + ) + bal = [e for e in log_entries if e.get("mode") == "balances"] + if bal: + e = bal[-1] + lines.append( + "Senaste Saldo: HTTP %s%s, balance_type %s" + % ( + e.get("status"), + (" " + str(e.get("error"))) if e.get("error") else "", + ",".join(e.get("balance_types") or []) or "-", + ) + ) + if log_entries: + lines.append("") + lines.append("Logg (UTC; läge, HTTP, sidor/tomma, rader, från, äldsta, ny/ändr/borta, PDNG->BOOK er/datum):") + for e in log_entries[-40:]: + if e.get("mode") in ("user", "background"): + lines.append( + " %s %-10s %s sidor %s/%s rader %s från %s äldsta %s ny %s ändr %s borta %s pb %s/%s%s" + % ( + e.get("at"), + e.get("mode"), + e.get("status"), + e.get("pages"), + e.get("empty_pages"), + e.get("count", "-"), + e.get("date_from"), + e.get("oldest", "-"), + e.get("new", "-"), + e.get("changed", "-"), + e.get("disappeared", "-"), + e.get("pdng_to_book_by_entry_reference", "-"), + e.get("pdng_to_book_by_amount_date", "-"), + (" fel " + str(e.get("error"))) if e.get("error") else "", + ) + ) + else: + lines.append( + " %s %-10s %s%s" + % ( + e.get("at"), + e.get("mode"), + e.get("status", ""), + (" fel " + str(e.get("error"))) if e.get("error") else "", + ) + ) + return "\n".join(lines) + + +# ----------------------------------------------------------------------------- HTML + +CSS = """ +:root{--fg:#1b1b1b;--bg:#fafafa;--mut:#666;--hl:#fff3c4;--ok:#e6f4ea;--err:#fde7e7;--bd:#ddd} +@media (prefers-color-scheme: dark){:root{--fg:#eee;--bg:#161616;--mut:#999;--hl:#4a3f12;--ok:#173322;--err:#4a1c1c;--bd:#333}} +body{font:14px/1.45 system-ui,sans-serif;color:var(--fg);background:var(--bg);max-width:1200px;margin:0 auto;padding:16px} +h1{font-size:20px}h2{font-size:16px;margin-top:28px;border-bottom:1px solid var(--bd);padding-bottom:4px} +table{border-collapse:collapse;width:100%;margin:8px 0}td,th{border-bottom:1px solid var(--bd);padding:4px 6px;text-align:left;vertical-align:top} +th{font-weight:600}.num{text-align:right;font-variant-numeric:tabular-nums;white-space:nowrap} +.hl{background:var(--hl)}.ok{background:var(--ok);padding:8px}.err{background:var(--err);padding:8px} +.mut{color:var(--mut)}pre{white-space:pre-wrap;word-break:break-word;font-size:12px;margin:4px 0} +textarea{width:100%;font:12px ui-monospace,monospace}button{margin:2px 4px 2px 0} +.tag{font-size:11px;padding:1px 5px;border-radius:3px;border:1px solid var(--bd)} +.scroll{max-height:420px;overflow:auto} +.paste{border:2px solid var(--bd);padding:8px 12px;margin:12px 0}.paste.active{border-color:#c99a00;background:var(--hl)} +""" + + +def e(s): + return html.escape("" if s is None else str(s), quote=True) + + +def render_last(): + if not LAST: + return "" + cls = "ok" if LAST.get("kind") == "ok" else "err" + out = ['
%s' % (cls, e(LAST.get("title")))] + if "status" in LAST: + out.append(" HTTP %s" % e(LAST.get("status"))) + out.append(" %s" % e(LAST.get("at"))) + if LAST.get("message"): + out.append("
%s
" % e(LAST["message"])) + if LAST.get("body") and LAST.get("kind") != "ok": + out.append("
%s
" % e(LAST["body"])) + out.append("
") + return "".join(out) + + +def render_status(st): + app_id, key_path, problems = load_config() + sess = st.get("session") or {} + rows = [ + ("Application id", e(app_id) if app_id else "saknas"), + ( + "Nyckel", + ("hittad (%s)" % e(os.path.basename(key_path))) + if key_path and os.path.isfile(key_path) + else "saknas", + ), + ("openssl", e(openssl_bin())), + ("API", e(CFG["api_base"])), + ("Bank", e(st["settings"].get("aspsp_name")) or "inte vald"), + ] + if sess.get("session_id"): + vu = sess.get("valid_until") or "" + left = "" + try: + vdt = dt.datetime.fromisoformat(vu.replace("Z", "+00:00")) + left = " (%d dagar kvar)" % (vdt - dt.datetime.now(dt.timezone.utc)).days + except ValueError: + pass + rows.append(("Session", "finns, giltig till %s%s" % (e(vu), e(left)))) + else: + rows.append(("Session", "ingen")) + acct = current_account(st) + rows.append( + ( + "Konto", + e("%s %s (%s)" % (acct.get("name") or "", acct.get("product") or "", acct.get("currency") or "")) + if acct + else "inte valt", + ) + ) + if st.get("pending_auth"): + rows.append(("Väntar på", "callback (startad %s)" % e(st["pending_auth"].get("started_at")))) + out = [""] + for k, v in rows: + out.append("" % (e(k), v)) + out.append("
%s%s
") + if problems: + out.append('
%s
' % "
".join(e(p) for p in problems)) + return "".join(out) + + +def current_account(st): + uid = st.get("account_uid") + for a in (st.get("session") or {}).get("accounts") or []: + if a.get("uid") == uid: + return a + return None + + +def render_aspsps(st): + out = [ + '
' + ' GET /aspsps?country=SE %s
' + % (("(hämtad " + e(st.get("aspsps_fetched_at")) + ")") if st.get("aspsps_fetched_at") else "") + ] + aspsps = st.get("aspsps") or [] + if aspsps: + chosen = st["settings"].get("aspsp_name") + ordered = sorted( + aspsps, key=lambda a: (0 if HIGHLIGHT.search(a.get("name") or "") else 1, a.get("name") or "") + ) + out.append('
') + out.append( + "" + "" + ) + for a in ordered: + name = a.get("name") or "" + mcv = a.get("maximum_consent_validity") + methods = "; ".join( + "%s/%s/%s" % (m.get("name"), m.get("approach"), m.get("psu_type")) + for m in a.get("auth_methods") or [] + ) + out.append( + '' + '' + % ( + "hl" if HIGHLIGHT.search(name) else "", + e(name), + " checked" if name == chosen else "", + e(name), + e(a.get("beta")), + e("%.0f d" % (mcv / 86400.0)) if isinstance(mcv, (int, float)) else "?", + e(",".join(a.get("psu_types") or [])), + e(methods), + e(",".join(a.get("required_psu_headers") or [])), + ) + ) + out.append("
NamnbetaMax samtyckepsu_typesauth_methodsrequired_psu_headers
%s%s%s%s%s%s
") + a = chosen_aspsp(st) + if a: + out.append("

Vald: %s

%s
" % (e(a.get("name")), e(json.dumps(a, ensure_ascii=False, indent=2)))) + return "".join(out) + + +def render_connect(st): + s = st["settings"] + a = chosen_aspsp(st) + secs = consent_seconds(a) + days_opts = "".join( + '' % (d, " selected" if d == s["history_days"] else "", d) + for d in HISTORY_CHOICES + ) + redirect = s.get("redirect_url") + radios = [] + for r in REDIRECT_CHOICES: + radios.append( + '
' + % (e(r), " checked" if r == redirect else "", e(r)) + ) + custom = redirect not in REDIRECT_CHOICES + radios.append( + '' + '' + % (" checked" if custom else "", e(redirect if custom else "")) + ) + methods = [m for m in (a or {}).get("auth_methods") or [] if m.get("psu_type") in (None, "personal")] + m_opts = [''] + for m in methods: + m_opts.append( + '' + % (e(m.get("name")), " selected" if m.get("name") == s.get("auth_method") else "", e(m.get("name")), e(m.get("approach"))) + ) + strat = "".join( + '' % (v, " selected" if v == s.get("strategy") else "", v) + for v in ("default", "longest") + ) + out = [ + '
', + "

Historik:   strategi: " + ' (longest = Enable Bankings strategy=longest)

' % (days_opts, strat), + "

Redirect URL (måste finnas registrerad i Enable Banking-appen):
%s

" % "".join(radios), + "

auth_method:

" % "".join(m_opts), + "

access.valid_until = nu + %s dagar (%s, max 180) − 5 min; psu_type personal; aspsp {%s, SE}

" + % ( + "%.1f" % (secs / 86400.0), + "bankens maximum_consent_validity" if a else "reserv 90 d, bankens max okänt", + e(s.get("aspsp_name") or "?"), + ), + ' ', + "
", + ] + pending = st.get("pending_auth") or {} + box_cls = "paste active" if pending else "paste" + out.append('

Klistra in adressen du hamnade på

' % box_cls) + out.append("

%s

" % e(PASTE_HINT)) + if pending.get("url"): + out.append( + '

Bankens sida öppnades i en ny flik (startad %s). ' + 'Öppna bankens sida igen

' + % (e(pending.get("started_at")), e(pending["url"])) + ) + out.append( + '
' + "
" + ) + out.append( + "

Reserv: http://localhost:%d/callback hanteras också direkt om den adressen " + "någon gång kan registreras (kontrollpanelen nekar http i dag).

" % CFG["port"] + ) + return "".join(out) + + +def render_accounts(st): + sess = st.get("session") or {} + if not sess.get("session_id"): + return "

Ingen session ännu.

" + out = [ + "

Session %s, giltig till %s, aspsp %s, psu_type %s, skapad %s

" + % ( + e(mask(sess.get("session_id"))), + e(sess.get("valid_until")), + e((sess.get("aspsp") or {}).get("name")), + e(sess.get("psu_type")), + e(sess.get("created_at")), + ), + '
GET /sessions/{id}
', + '
' + "", + ] + for a in sess.get("accounts") or []: + out.append( + "" + "" + % ( + e(a.get("uid")), + " checked" if a.get("uid") == st.get("account_uid") else "", + e(a.get("uid")), + e(a.get("name")), + e(" / ".join(x for x in (a.get("product"), a.get("details")) if x)), + e(a.get("currency")), + e(" / ".join(str(x) for x in (a.get("cash_account_type"), a.get("usage")) if x)), + e(json.dumps(mask_ids(a.get("account_id")), ensure_ascii=False)), + e(json.dumps(mask_ids(a.get("all_account_ids")), ensure_ascii=False)), + e(mask(a.get("identification_hash") or "")), + "finns" if present(a.get("credit_limit")) else "-", + ) + ) + out.append("
uidnameproduct / detailsvalutacash_account_type / usageaccount_id (maskerat)all_account_idsidentification_hashcredit_limit
%s%s%s%s%s
%s
%s
%s%s
") + return "".join(out) + + +def render_fetch(st): + s = st["settings"] + meta = st.get("transactions_meta") or {} + out = [ + '
', + "

PSU IP-adress: (tom = hämtas från api.ipify.org vid första Uppdatera)

" + % e(s.get("psu_ip")), + "

Från %s (idag − %d dagar), strategi %s

" + % ( + e((dt.date.today() - dt.timedelta(days=int(s["history_days"]))).isoformat()), + int(s["history_days"]), + e(s.get("strategy")), + ), + " med PSU-headers (användaren närvarande)
", + " utan PSU-headers
", + " GET /accounts/{uid}/balances med PSU-headers", + "
", + ] + if meta: + out.append( + "

Senast hämtat %s (%s, %s sidor).

" + % (e(meta.get("at")), e(meta.get("mode")), e(meta.get("pages"))) + ) + bal = st.get("balances") + if bal: + out.append("") + for b in bal.get("balances") or []: + ba = b.get("balance_amount") or {} + out.append( + "" + % ( + e(b.get("name")), + e(b.get("balance_type")), + e(ba.get("amount")), + e(ba.get("currency")), + e(b.get("reference_date")), + e(b.get("last_change_date_time")), + ) + ) + out.append("
namebalance_typebeloppvalutareference_datelast_change_date_time
%s%s%s%s%s%s

Saldo hämtat %s

" % e(bal.get("at"))) + return "".join(out) + + +def tx_row(t, tag, raw_id): + amt = amount_of(t) + return ( + "%s%s%s%s%s%s%s" + "%s
rå
%s
" + % ( + e(t.get("booking_date") or "-") + "
" + e(t.get("transaction_date") or "-") + "", + tag, + e(text_of(t) or "-"), + e(fmt_kr(amt)) + " kr", + e(currency_of(t)), + e(t.get("credit_debit_indicator")), + e(t.get("status")), + e(t.get("entry_reference") or "-"), + e(json.dumps(t, ensure_ascii=False, indent=2)), + ) + ) + + +def render_purchases(st): + txs = st.get("transactions") or [] + prev = st.get("previous_transactions") or [] + d = st.get("last_diff") or {} + if not txs and not d: + return "

Inget hämtat ännu.

" + new = set(d.get("new") or []) + changed = {c["idx"]: c for c in d.get("changed") or []} + out = [] + if d.get("had_previous"): + out.append( + "

Jämfört med förra hämtningen: %d nya, %d ändrade, %d borta.

" + % (len(new), len(changed), len(d.get("disappeared") or [])) + ) + else: + out.append("

Första hämtningen; inget att jämföra med.

") + out.append( + "" + "" + ) + order = sorted(range(len(txs)), key=lambda j: (str(date_of(txs[j]) or ""), j), reverse=True) + for j in order: + t = txs[j] + tag = "" + if d.get("had_previous") and j in new: + tag = "ny" + elif j in changed: + c = changed[j] + p = prev[c["prev_idx"]] if c["prev_idx"] < len(prev) else {} + st_change = "" + if p.get("status") != t.get("status"): + st_change = "%s→%s " % (p.get("status"), t.get("status")) + tag = "ändrad %s
%s; via %s" % ( + e(st_change), + e(",".join(c["fields"])), + e(c["by"]), + ) + out.append(tx_row(t, tag, j)) + out.append("
booking_date
transaction_date		difftextbeloppvalutariktningstatusentry_reference
") + gone = d.get("disappeared") or [] + if gone: + out.append("

Försvunna sedan förra hämtningen

") + for i in gone: + if i < len(prev): + out.append(tx_row(prev[i], "borta", "p%d" % i)) + out.append("
") + return "".join(out) + + +def render_page(st): + text = analysis_text(st) + return """ +K0 Enable Banking +

K0: Enable Banking-test

+%s +

1. Status

%s +

2. Banker

%s +

3–4. Anslut

%s +

5. Konton

%s +

6. Hämta

%s +

7. Köp

%s +

8. Analys

+

Bara aggregat; inga texter, belopp eller id:n.

+ +

9. Koppla från

+
+DELETE /sessions/{id}; rensar session, konton och transaktioner lokalt (inställningar, bankernas lista och den aggregerade loggen behålls)
+""" % ( + CSS, + render_last(), + render_status(st), + render_aspsps(st), + render_connect(st), + render_accounts(st), + render_fetch(st), + render_purchases(st), + e(text), + ) + + +# ----------------------------------------------------------------------------- server + + +class Handler(BaseHTTPRequestHandler): + server_version = "K0/1" + + def log_request(self, code="-", size="-"): + path = self.path.split("?", 1)[0] + log("LOCAL %s %s %s" % (self.command, path, code)) + + def log_message(self, fmt, *args): # silence the default request-line logging + return + + def send_html(self, body, status=200): + raw = body.encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(raw))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(raw) + + def redirect(self, location): + self.send_response(303) + self.send_header("Location", location) + self.send_header("Content-Length", "0") + self.end_headers() + + def browser_info(self): + return { + "client_ip": self.client_address[0], + "user_agent": self.headers.get("User-Agent"), + "accept": self.headers.get("Accept"), + "accept_charset": self.headers.get("Accept-Charset"), + "accept_encoding": self.headers.get("Accept-Encoding"), + "accept_language": self.headers.get("Accept-Language"), + } + + def do_GET(self): + parsed = urllib.parse.urlparse(self.path) + if parsed.path == "/": + with LOCK: + st = load_state() + self.send_html(render_page(st)) + return + if parsed.path in ("/callback", "/cards/callback"): + params = extract_code_state("?" + parsed.query) + with LOCK: + st = load_state() + complete_auth(st, params, "callback") + save_state(st) + self.redirect("/#svar") + return + if parsed.path == "/favicon.ico": + self.send_response(204) + self.end_headers() + return + self.send_html("

Hittades inte.

", 404) + + def same_origin(self): + origin = self.headers.get("Origin") or self.headers.get("Referer") or "" + if not origin: + return True # curl and some browsers omit both on same-origin form posts + host = urllib.parse.urlparse(origin).netloc + return host in ("localhost:%d" % CFG["port"], "127.0.0.1:%d" % CFG["port"]) + + def do_POST(self): + parsed = urllib.parse.urlparse(self.path) + if not parsed.path.startswith("/action/") or not self.same_origin(): + self.send_html("

Nekad.

", 403) + return + length = int(self.headers.get("Content-Length") or 0) + raw = self.rfile.read(length).decode("utf-8", "replace") if length else "" + form = {k: v[0] for k, v in urllib.parse.parse_qs(raw, keep_blank_values=True).items()} + name = parsed.path[len("/action/"):] + target = "/#svar" + with LOCK: + st = load_state() + if name == "list_aspsps": + action_list_aspsps(st) + elif name == "choose_aspsp": + action_choose_aspsp(st, form) + elif name == "settings": + save_settings(st, form) + if form.get("op") == "auth": + url = action_auth(st) + if url: + # the bank page opens in its own tab; this tab shows the paste box + if CFG["open_browser"]: + webbrowser.open_new_tab(url) + target = "/#klistra" + else: + set_last("ok", "Inställningar sparade", section="anslut") + elif name == "paste": + complete_auth(st, extract_code_state(form.get("url", "")), "inklistrad adress") + elif name == "choose_account": + action_choose_account(st, form) + elif name == "check_session": + action_check_session(st) + elif name == "fetch": + st["settings"]["psu_ip"] = form.get("psu_ip", "").strip() + op = form.get("op") + if op in ("user", "background"): + action_fetch(st, op, self.browser_info()) + elif op == "balances": + action_balances(st, self.browser_info()) + elif name == "disconnect": + action_disconnect(st) + else: + self.send_html("

Okänd åtgärd.

", 404) + return + save_state(st) + self.redirect(target) + + +class V6Server(ThreadingHTTPServer): + address_family = socket.AF_INET6 + + +def main(argv=None): + p = argparse.ArgumentParser(description="K0 Enable Banking test app") + p.add_argument("--port", type=int, default=DEFAULT_PORT) + p.add_argument("--local-dir", default=DEFAULT_LOCAL_DIR, help="config and state folder") + p.add_argument("--api-base", default=DEFAULT_API_BASE, help="for tests against a fake API") + p.add_argument("--no-browser", action="store_true") + args = p.parse_args(argv) + CFG["port"] = args.port + CFG["local_dir"] = os.path.abspath(args.local_dir) + CFG["api_base"] = args.api_base + CFG["open_browser"] = not args.no_browser + os.makedirs(CFG["local_dir"], exist_ok=True) + try: + os.chmod(CFG["local_dir"], 0o700) + except OSError: + pass + server = ThreadingHTTPServer(("127.0.0.1", args.port), Handler) + # browsers may resolve localhost to ::1 first; serve the loopback there too if possible + try: + server6 = V6Server(("::1", args.port), Handler) + threading.Thread(target=server6.serve_forever, daemon=True).start() + except OSError: + pass + url = "http://localhost:%d/" % args.port + log("K0 listening on %s (state in %s)" % (url, CFG["local_dir"])) + if not args.no_browser: + threading.Timer(0.5, lambda: webbrowser.open(url)).start() + try: + server.serve_forever() + except KeyboardInterrupt: + log("stopped") + return 0 + + +if __name__ == "__main__": + sys.exit(main())