Teller Connect — PFM Test

""" 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, datetime, 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' # ── Logger ──────────────────────────────────────────────────────────────────── _log_file = None # set in main() when --log-file is given def _log(label, data, status_code=None, url=None): """Append a structured JSON entry to the log file if one is configured.""" if _log_file is None: return entry = { 'timestamp': datetime.now().isoformat(timespec='seconds'), 'label': label, 'url': url, 'status': status_code, 'response': data, } with open(_log_file, 'a', encoding='utf-8') as f: f.write(json.dumps(entry, default=str) + '\n') # ── 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

""" def connect_bank(cert_path, key_path, port=8765, days=30, log_path=None): """ 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)}') _log('teller_connect_enrollment', { 'enrollment_id': enrollment_id, 'institution': institution, 'access_token': f'{access_token[:8]}…(redacted)', }) # 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 global _log_file if log_path: _log_file = log_path 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') url = f'{TELLER_BASE}/accounts' try: resp = session.get(url, timeout=15) if resp.status_code == 200: accounts = resp.json() _log('get_accounts', accounts, resp.status_code, url) _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: _log('get_accounts', resp.text, resp.status_code, url) _fail('Unauthorized (401) — check access token and mTLS certs') _info(f'Response: {resp.text[:500]}') else: _log('get_accounts', resp.text, resp.status_code, url) _fail(f'HTTP {resp.status_code}') _info(f'Response: {resp.text[:500]}') except Exception as e: _log('get_accounts', str(e), None, url) _fail(f'Request failed: {e}') return [] def test_balances(session, account_id): _head(f'2. Balance — GET /accounts/{account_id}/balances') url = f'{TELLER_BASE}/accounts/{account_id}/balances' try: resp = session.get(url, timeout=15) if resp.status_code == 200: data = resp.json() _log('get_balances', data, resp.status_code, url) _ok(f'Ledger: {data.get("ledger", "n/a")} | Available: {data.get("available", "n/a")}') _info(_pretty(data)) return data else: _log('get_balances', resp.text, resp.status_code, url) _fail(f'HTTP {resp.status_code} — {resp.text[:300]}') except Exception as e: _log('get_balances', str(e), None, url) _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() url = f'{TELLER_BASE}/accounts/{account_id}/transactions' try: resp = session.get(url, params={'start_date': start, 'end_date': end}, timeout=30) if resp.status_code == 200: txns = resp.json() _log('get_transactions', txns, resp.status_code, url) _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: _log('get_transactions', resp.text, resp.status_code, url) _fail(f'HTTP {resp.status_code} — {resp.text[:300]}') except Exception as e: _log('get_transactions', str(e), None, url) _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}') url = f'{TELLER_BASE}/accounts/{account_id}/transactions/{txn_id}' try: resp = session.get(url, timeout=15) if resp.status_code == 200: data = resp.json() _log('get_transaction_detail', data, resp.status_code, url) _ok('Retrieved successfully') _info(_pretty(data)) return data else: _log('get_transaction_detail', resp.text, resp.status_code, url) _fail(f'HTTP {resp.status_code} — {resp.text[:300]}') except Exception as e: _log('get_transaction_detail', str(e), None, url) _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)') parser.add_argument('--log-file', metavar='PATH', default=None, help='Append all API responses as JSON lines to this file ' '(default: logs/teller_test_YYYYMMDD_HHMMSS.log)') args = parser.parse_args() cert = args.cert key = args.key # ── Set up log file ─────────────────────────────────────────────────────── global _log_file log_path = args.log_file if log_path is None: log_dir = os.path.join(_root, 'logs') os.makedirs(log_dir, exist_ok=True) stamp = datetime.now().strftime('%Y%m%d_%H%M%S') log_path = os.path.join(log_dir, f'teller_test_{stamp}.log') _log_file = log_path _info(f'Logging responses to: {_log_file}') 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, log_path=log_path) 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()