From d09c739288c8c714ac1e1f2685b33f08d4d0038a Mon Sep 17 00:00:00 2001 From: Nguyen Ngo Date: Mon, 1 Jun 2026 20:54:16 -0400 Subject: [PATCH] 06/01 Add teller test tool --- .gitignore | 1 + scripts/test_teller.py | 485 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 486 insertions(+) create mode 100644 scripts/test_teller.py diff --git a/.gitignore b/.gitignore index fda3c23..e4b669a 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # Environment .env +.teller_token # Python __pycache__/ diff --git a/scripts/test_teller.py b/scripts/test_teller.py new file mode 100644 index 0000000..7975e6d --- /dev/null +++ b/scripts/test_teller.py @@ -0,0 +1,485 @@ +""" +Teller API Test Script +====================== +Standalone script to verify Teller credentials, API connectivity, and bank +enrollment — no Flask app required. + +Modes +----- +1. Test with an existing access token: + python scripts/test_teller.py --token + +2. Connect a new bank via Teller Connect (opens browser, captures token): + python scripts/test_teller.py --connect + python scripts/test_teller.py --connect --port 8765 # custom port + + After completing the bank login in the browser, the script captures the + access token and enrollment ID, then automatically runs all API tests. + +3. Target a specific account: + python scripts/test_teller.py --token --account + +4. Fetch more history: + python scripts/test_teller.py --token --days 90 + +Reads TELLER_APP_ID, TELLER_ENV, TELLER_CERT_PATH, TELLER_KEY_PATH, and +TELLER_WEBHOOK_SECRET from .env automatically. +Override cert paths with --cert / --key flags if needed. +""" + +import argparse +import json +import os +import sys +from datetime import date, timedelta + +# ── Load .env from project root ─────────────────────────────────────────────── + +_here = os.path.dirname(os.path.abspath(__file__)) +_root = os.path.dirname(_here) +_env = os.path.join(_root, '.env') + +if os.path.exists(_env): + with open(_env) as _f: + for _line in _f: + _line = _line.strip() + if _line and not _line.startswith('#') and '=' in _line: + _k, _v = _line.split('=', 1) + os.environ.setdefault(_k.strip(), _v.strip().strip('"').strip("'")) + +# ── Teller constants ────────────────────────────────────────────────────────── + +TELLER_BASE = 'https://api.teller.io' +TELLER_VERSION = '2020-10-12' + +# ── Console helpers ─────────────────────────────────────────────────────────── + +def _ok(msg): print(f' \033[32m✓\033[0m {msg}') +def _fail(msg): print(f' \033[31m✗\033[0m {msg}') +def _info(msg): print(f' \033[34m→\033[0m {msg}') +def _warn(msg): print(f' \033[33m!\033[0m {msg}') +def _head(msg): print(f'\n\033[1m{msg}\033[0m') + + +def _pretty(data, indent=4): + return json.dumps(data, indent=indent, default=str) + + +# ── mTLS session ────────────────────────────────────────────────────────────── + +def _build_session(access_token, cert_path, key_path): + import requests + if not os.path.exists(cert_path): + raise FileNotFoundError(f'Cert not found: {cert_path}') + if not os.path.exists(key_path): + raise FileNotFoundError(f'Key not found: {key_path}') + + s = requests.Session() + s.cert = (cert_path, key_path) + s.auth = (access_token, '') + s.headers.update({'Teller-Version': TELLER_VERSION, 'Accept': 'application/json'}) + return s + + +# ── Bank connection via Teller Connect ─────────────────────────────────────── + +_CONNECT_HTML = """ + + + + Teller Connect — PFM Test + + + + +
+

🏦 Connect Your Bank

+

Sign in to your bank via Teller Connect.
+ Your credentials go directly to your bank — never to this script.

+ +
+
+ + + +""" + + +def connect_bank(cert_path, key_path, port=8765, days=30): + """ + Spin up a local HTTP server, open Teller Connect in the browser, + capture the access token, then run the full test suite. + """ + import threading + import webbrowser + from http.server import BaseHTTPRequestHandler, HTTPServer + + app_id = os.environ.get('TELLER_APP_ID', '') + env = os.environ.get('TELLER_ENV', 'development') + + if not app_id: + _fail('TELLER_APP_ID not set in .env — cannot launch Teller Connect') + sys.exit(1) + + captured = {} + server_ready = threading.Event() + capture_done = threading.Event() + + html = _CONNECT_HTML.format(app_id=app_id, env=env) + + class Handler(BaseHTTPRequestHandler): + def log_message(self, fmt, *args): + pass # silence access log + + def do_GET(self): + self.send_response(200) + self.send_header('Content-Type', 'text/html; charset=utf-8') + self.end_headers() + self.wfile.write(html.encode()) + + def do_POST(self): + if self.path == '/capture': + length = int(self.headers.get('Content-Length', 0)) + body = self.rfile.read(length) + try: + data = json.loads(body) + captured.update(data) + self.send_response(200) + self.send_header('Content-Type', 'application/json') + self.end_headers() + self.wfile.write(b'{"status":"ok"}') + capture_done.set() + except Exception: + self.send_response(400) + self.end_headers() + else: + self.send_response(404) + self.end_headers() + + server = HTTPServer(('127.0.0.1', port), Handler) + url = f'http://127.0.0.1:{port}' + + def _serve(): + server_ready.set() + server.serve_forever() + + t = threading.Thread(target=_serve, daemon=True) + t.start() + server_ready.wait() + + _head('Bank Connection — Teller Connect') + _info(f'Local server: {url}') + _info(f'App ID : {app_id}') + _info(f'Environment : {env}') + print() + print(' Opening browser. Complete the bank login, then return here.') + print(' Press Ctrl+C to cancel.\n') + + webbrowser.open(url) + + try: + capture_done.wait(timeout=300) # 5-minute window + except KeyboardInterrupt: + print('\n Cancelled.') + sys.exit(0) + finally: + server.shutdown() + + if not captured.get('accessToken'): + _fail('No token received — did you complete the bank login?') + sys.exit(1) + + access_token = captured['accessToken'] + enrollment_id = captured.get('enrollmentId', 'unknown') + institution = captured.get('institution', 'unknown') + + print() + _ok(f'Bank connected: {institution}') + _info(f'Enrollment ID : {enrollment_id}') + _info(f'Access token : {access_token[:8]}…{"*" * (len(access_token) - 8)}') + + # Save token to a local file for reuse + token_file = os.path.join(_root, '.teller_token') + with open(token_file, 'w') as f: + json.dump({ + 'access_token': access_token, + 'enrollment_id': enrollment_id, + 'institution': institution, + 'captured_at': date.today().isoformat(), + }, f, indent=2) + _info(f'Token saved to .teller_token (gitignored) for reuse with --token') + + # Run full test suite with the captured token + print() + try: + session = _build_session(access_token, cert_path, key_path) + except FileNotFoundError as e: + _fail(str(e)) + sys.exit(1) + + accounts = test_connectivity(session) + account_id = accounts[0]['id'] if accounts else None + + if account_id: + test_balances(session, account_id) + txns = test_transactions(session, account_id, days=days) + if txns: + test_transaction_detail(session, account_id, txns[0]['id']) + else: + _warn('No accounts returned — skipping balance/transaction tests') + + test_webhook_signature(os.environ.get('TELLER_WEBHOOK_SECRET', '')) + + +# ── Individual API tests ────────────────────────────────────────────────────── + +def test_connectivity(session): + _head('1. Connectivity — GET /accounts') + try: + resp = session.get(f'{TELLER_BASE}/accounts', timeout=15) + if resp.status_code == 200: + accounts = resp.json() + _ok(f'Connected — {len(accounts)} account(s) returned') + for a in accounts: + _info(f'{a["id"]} {a.get("institution", {}).get("name", "?")} ' + f'{a.get("name", "")} ({a.get("type", "")} / {a.get("subtype", "")})') + return accounts + elif resp.status_code == 401: + _fail('Unauthorized (401) — check access token and mTLS certs') + _info(f'Response: {resp.text[:500]}') + else: + _fail(f'HTTP {resp.status_code}') + _info(f'Response: {resp.text[:500]}') + except Exception as e: + _fail(f'Request failed: {e}') + return [] + + +def test_balances(session, account_id): + _head(f'2. Balance — GET /accounts/{account_id}/balances') + try: + resp = session.get(f'{TELLER_BASE}/accounts/{account_id}/balances', timeout=15) + if resp.status_code == 200: + data = resp.json() + _ok(f'Ledger: {data.get("ledger", "n/a")} | Available: {data.get("available", "n/a")}') + _info(_pretty(data)) + return data + else: + _fail(f'HTTP {resp.status_code} — {resp.text[:300]}') + except Exception as e: + _fail(f'Request failed: {e}') + return None + + +def test_transactions(session, account_id, days=30): + _head(f'3. Transactions — GET /accounts/{account_id}/transactions (last {days} days)') + start = (date.today() - timedelta(days=days)).isoformat() + end = date.today().isoformat() + try: + resp = session.get( + f'{TELLER_BASE}/accounts/{account_id}/transactions', + params={'start_date': start, 'end_date': end}, + timeout=30, + ) + if resp.status_code == 200: + txns = resp.json() + _ok(f'{len(txns)} transaction(s) from {start} to {end}') + for txn in txns[:5]: + amount = float(txn.get('amount', 0)) + sign = '-' if amount > 0 else '+' + _info(f'{txn.get("date", "")} {sign}${abs(amount):.2f}' + f' [{txn.get("status", "")}] {txn.get("description", "")}') + if len(txns) > 5: + _info(f'… and {len(txns) - 5} more') + return txns + else: + _fail(f'HTTP {resp.status_code} — {resp.text[:300]}') + except Exception as e: + _fail(f'Request failed: {e}') + return [] + + +def test_transaction_detail(session, account_id, txn_id): + _head(f'4. Transaction detail — GET /accounts/{account_id}/transactions/{txn_id}') + try: + resp = session.get( + f'{TELLER_BASE}/accounts/{account_id}/transactions/{txn_id}', + timeout=15, + ) + if resp.status_code == 200: + data = resp.json() + _ok('Retrieved successfully') + _info(_pretty(data)) + return data + else: + _fail(f'HTTP {resp.status_code} — {resp.text[:300]}') + except Exception as e: + _fail(f'Request failed: {e}') + return None + + +def test_webhook_signature(webhook_secret): + _head('5. Webhook HMAC verification (local simulation)') + if not webhook_secret: + _warn('TELLER_WEBHOOK_SECRET not set — skipping') + return + + import hmac + import hashlib + import time + + body = b'{"type":"transactions.processed","enrollment_id":"test_enroll_123"}' + ts = str(int(time.time())) + payload = f'{ts}.'.encode() + body + sig = hmac.new(webhook_secret.encode(), payload, hashlib.sha256).hexdigest() + header = f't={ts},v1={sig}' + + parts = dict(p.split('=', 1) for p in header.split(',') if '=' in p) + sigs = [v for p in header.split(',') if p.startswith('v1=') for _, v in [p.split('=', 1)]] + recomputed = hmac.new( + webhook_secret.encode(), + f'{parts["t"]}.'.encode() + body, + hashlib.sha256, + ).hexdigest() + + if any(hmac.compare_digest(recomputed, s) for s in sigs): + _ok('HMAC-SHA256 signature round-trip passed') + _info(f'Example Teller-Signature header: {header}') + else: + _fail('Signature mismatch — check TELLER_WEBHOOK_SECRET') + + +# ── Entry point ─────────────────────────────────────────────────────────────── + +def main(): + parser = argparse.ArgumentParser( + description='Test Teller API — use --connect to enroll a bank, ' + 'or --token to test an existing enrollment.', + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument('--connect', action='store_true', + help='Open Teller Connect in browser to enroll a bank, then run tests') + mode.add_argument('--token', metavar='ACCESS_TOKEN', + help='Test using an existing Teller access token') + + parser.add_argument('--account', default=None, + help='Teller account ID (optional; defaults to first account)') + parser.add_argument('--cert', default=os.environ.get('TELLER_CERT_PATH', ''), + help='Path to certificate.pem (default: TELLER_CERT_PATH from .env)') + parser.add_argument('--key', default=os.environ.get('TELLER_KEY_PATH', ''), + help='Path to private_key.pem (default: TELLER_KEY_PATH from .env)') + parser.add_argument('--days', type=int, default=30, + help='Days of transaction history to fetch (default: 30)') + parser.add_argument('--port', type=int, default=8765, + help='Local server port for --connect mode (default: 8765)') + args = parser.parse_args() + + cert = args.cert + key = args.key + + if not cert or not key: + _fail('Cert/key paths not found.\n' + ' Set TELLER_CERT_PATH / TELLER_KEY_PATH in .env, ' + 'or pass --cert / --key flags.') + sys.exit(1) + + if args.connect: + connect_bank(cert, key, port=args.port, days=args.days) + print('\n\033[1mDone.\033[0m\n') + return + + # ── --token mode ────────────────────────────────────────────────────────── + print(f'\n\033[1mTeller API Test\033[0m') + print(f' Base URL : {TELLER_BASE}') + print(f' Cert : {cert}') + print(f' Key : {key}') + print(f' Token : {args.token[:8]}…{"*" * max(0, len(args.token) - 8)}') + + try: + session = _build_session(args.token, cert, key) + except FileNotFoundError as e: + _fail(str(e)) + sys.exit(1) + + accounts = test_connectivity(session) + account_id = args.account + if not account_id and accounts: + account_id = accounts[0]['id'] + _info(f'No --account given; using first account: {account_id}') + + if account_id: + test_balances(session, account_id) + txns = test_transactions(session, account_id, days=args.days) + if txns: + test_transaction_detail(session, account_id, txns[0]['id']) + else: + _warn('No account ID available — skipping balance/transaction tests') + + test_webhook_signature(os.environ.get('TELLER_WEBHOOK_SECRET', '')) + + print('\n\033[1mDone.\033[0m\n') + + +if __name__ == '__main__': + main()