xero_list_bank_accounts_alt.py

import hashlib
import base64
import secrets
import uuid
import re
from urllib.parse import urlencode, urlparse, parse_qs
from curl_cffi import requests

# Injected at runtime; fallback for local testing
try:
    BASE_URL
except NameError:
    BASE_URL = "https://go.xero.com"


def run(headers, user_input):
    """
    List all bank accounts for the authenticated Xero organization.

    Returns a list of bank accounts with their IDs and names.
    """
    cookies = headers.get("Cookie", "")

    # Extract organization ID from cookies
    org_id = _extract_org_id(cookies)
    if not org_id:
        return {'status_code': 400, 'body': {'error': 'Organization ID not found in cookies'}}

    # Get fresh Bearer token via OAuth2 PKCE flow
    access_token, error = _get_bearer_token(headers)
    if not access_token:
        if error and "Session expired" in error:
            return {'status_code': 401, 'body': {'error': 'Session expired - reauthentication required'}}
        return {'status_code': 401, 'body': {'error': f'Failed to obtain access token: {error}'}}

    try:
        accounts = _fetch_bank_accounts(headers, org_id, access_token)
        return {
            'status_code': 200,
            'body': {
                'organisation_id': org_id,
                'bank_accounts': accounts,
                'total_count': len(accounts)
            }
        }
    except Exception as e:
        return {'status_code': 500, 'body': {'error': str(e)}}


# === PRIVATE ===

def _extract_org_id(cookies):
    """Extract organization ID from X-OrgID cookie."""
    match = re.search(r'X-OrgID=([a-f0-9-]+)', cookies)
    if match:
        return match.group(1)
    return None


def _get_bearer_token(headers):
    """
    Perform silent OIDC flow to obtain a fresh Bearer token.
    Uses PKCE (Proof Key for Code Exchange) for secure auth code exchange.
    """
    cookies = headers.get("Cookie", "")

    # Generate PKCE code verifier and challenge
    code_verifier = secrets.token_hex(32) + secrets.token_hex(32)
    code_challenge = base64.urlsafe_b64encode(
        hashlib.sha256(code_verifier.encode()).digest()
    ).decode().rstrip('=')

    # Build authorization URL for silent auth
    state = secrets.token_hex(16)
    auth_params = {
        'client_id': 'xero_business_go',
        'redirect_uri': 'https://go.xero.com/oidc/silent.html',
        'response_type': 'code',
        'scope': 'openid profile email xero_frontend-apis xero_frontend-platform-apis',
        'state': state,
        'code_challenge': code_challenge,
        'code_challenge_method': 'S256',
        'prompt': 'none'  # Silent auth - no user interaction
    }

    auth_url = f"https://login.xero.com/identity/connect/authorize?{urlencode(auth_params)}"

    # Make auth request - follow redirects to get the code
    response = requests.get(
        auth_url,
        headers={"Cookie": cookies},
        allow_redirects=True,
        impersonate="chrome131",
        timeout=30
    )

    # Extract auth code from final URL
    final_url = str(response.url)
    parsed = urlparse(final_url)
    query_params = parse_qs(parsed.query)

    if 'code' not in query_params:
        # Check if redirected to login page (session expired)
        if 'login' in final_url.lower() or response.status_code == 401:
            return None, "Session expired - login required"
        return None, f"No auth code in response URL: {final_url}"

    auth_code = query_params['code'][0]

    # Exchange code for token
    token_data = {
        'client_id': 'xero_business_go',
        'code': auth_code,
        'redirect_uri': 'https://go.xero.com/oidc/silent.html',
        'code_verifier': code_verifier,
        'grant_type': 'authorization_code'
    }

    token_response = requests.post(
        'https://identity.xero.com/connect/token',
        data=token_data,
        headers={
            'Content-Type': 'application/x-www-form-urlencoded',
            'Origin': 'https://go.xero.com',
            'Referer': 'https://go.xero.com/'
        },
        impersonate="chrome131",
        timeout=30
    )

    if token_response.status_code != 200:
        return None, f"Token exchange failed: {token_response.status_code}"

    token_json = token_response.json()
    return token_json.get('access_token'), None


def _fetch_bank_accounts(headers, org_id, access_token):
    """Fetch bank accounts from Xero API."""
    cookies = headers.get("Cookie", "")
    correlation_id = str(uuid.uuid4())

    api_headers = {
        "Cookie": cookies,
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "Accept": "application/json",
        "xero-correlation-id": correlation_id,
        "xero-organisation-id": org_id,
        "x-client": "xero-secure-fetch"
    }

    response = requests.get(
        f"{BASE_URL}/api/manage-bank-accounts/v2/bootstrap/organisations/{org_id}/",
        headers=api_headers,
        impersonate="chrome131",
        timeout=30
    )

    if response.status_code == 401:
        raise Exception('Authentication failed')

    if response.status_code != 200:
        raise Exception(f'API request failed: {response.text}')

    data = response.json()

    # Extract and format bank accounts from the response
    accounts = []
    bank_accounts = data.get('bankAccounts', [])

    for account in bank_accounts:
        accounts.append({
            'account_id': account.get('accountId'),
            'name': account.get('name')
        })

    return accounts