538 lines
21 KiB
Python
538 lines
21 KiB
Python
"""
|
|
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 <access_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 <tok> --account <account_id>
|
|
|
|
4. Fetch more history:
|
|
python scripts/test_teller.py --token <tok> --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 = """<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<title>Teller Connect — PFM Test</title>
|
|
<script src="https://cdn.teller.io/connect/connect.js"></script>
|
|
<style>
|
|
body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif;
|
|
background: #0f172a; color: #f1f5f9; display: flex;
|
|
align-items: center; justify-content: center; height: 100vh; margin: 0; }}
|
|
.card {{ background: #1e293b; border-radius: 12px; padding: 40px; text-align: center;
|
|
max-width: 400px; width: 100%; box-shadow: 0 8px 32px rgba(0,0,0,.4); }}
|
|
h2 {{ margin: 0 0 8px; font-size: 20px; }}
|
|
p {{ color: #94a3b8; font-size: 14px; margin: 0 0 28px; }}
|
|
button {{ background: #10b981; color: #fff; border: none; border-radius: 8px;
|
|
padding: 12px 28px; font-size: 15px; font-weight: 600; cursor: pointer;
|
|
width: 100%; transition: background .2s; }}
|
|
button:hover {{ background: #059669; }}
|
|
button:disabled {{ background: #334155; color: #64748b; cursor: not-allowed; }}
|
|
#status {{ margin-top: 16px; font-size: 13px; color: #94a3b8; min-height: 20px; }}
|
|
.success {{ color: #10b981; }} .error {{ color: #ef4444; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<div class="card">
|
|
<h2>🏦 Connect Your Bank</h2>
|
|
<p>Sign in to your bank via Teller Connect.<br>
|
|
Your credentials go directly to your bank — never to this script.</p>
|
|
<button id="connectBtn" onclick="startConnect()">Connect a Bank Account</button>
|
|
<div id="status"></div>
|
|
</div>
|
|
<script>
|
|
function setStatus(msg, cls) {{
|
|
const el = document.getElementById('status');
|
|
el.textContent = msg;
|
|
el.className = cls || '';
|
|
}}
|
|
|
|
function startConnect() {{
|
|
document.getElementById('connectBtn').disabled = true;
|
|
setStatus('Opening Teller Connect…');
|
|
|
|
const teller = TellerConnect.setup({{
|
|
applicationId: '{app_id}',
|
|
environment: '{env}',
|
|
products: ['transactions'],
|
|
onSuccess: function(enrollment) {{
|
|
setStatus('✓ Connected! Sending token to test script…', 'success');
|
|
fetch('/capture', {{
|
|
method: 'POST',
|
|
headers: {{'Content-Type': 'application/json'}},
|
|
body: JSON.stringify({{
|
|
accessToken: enrollment.accessToken,
|
|
enrollmentId: enrollment.enrollment.id,
|
|
institution: enrollment.enrollment.institution.name,
|
|
}}),
|
|
}}).then(() => {{
|
|
setStatus('✓ Token captured. You can close this tab.', 'success');
|
|
}}).catch(() => {{
|
|
setStatus('Capture failed — check terminal output.', 'error');
|
|
}});
|
|
}},
|
|
onFailure: function(failure) {{
|
|
setStatus('Connection failed: ' + (failure.message || JSON.stringify(failure)), 'error');
|
|
document.getElementById('connectBtn').disabled = false;
|
|
}},
|
|
onExit: function() {{
|
|
if (!document.getElementById('status').classList.contains('success')) {{
|
|
setStatus('Cancelled.');
|
|
document.getElementById('connectBtn').disabled = false;
|
|
}}
|
|
}},
|
|
}});
|
|
teller.open();
|
|
}}
|
|
</script>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
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()
|