XeroAPI/xero-python

Error code: 500 Error: unauthorized_client : Unknown client or client not enabled

mattjwbasterfield opened this issue · 3 comments

Hi

I have a Django app that had a working integration with the Xero API using the xero_python library. Suddenly on the 16th of March I had to re-auth the app and then was unable to login, receiving the "Error code: 500 Error: unauthorized_client : Unknown client or client not enabled" message.

I have checked and replaced client id and client secret.

I ran updates on that day so not sure if there was a change in the library that caused it to break.

Been troubleshooting this for a while so would appreciate any guidance.

views.py:

'import os

from django.core.cache import caches
from django.conf import settings
from django.shortcuts import redirect
from django.urls import reverse
import json
from functools import wraps
from datetime import timezone
import datetime
from django.http import HttpResponse
import urllib.parse
from authlib.integrations.django_client import OAuth

from xero_python.api_client import ApiClient, serialize
from xero_python.api_client.configuration import Configuration
from xero_python.api_client.oauth2 import OAuth2Token
from xero_python.exceptions import AccountingBadRequestException
from xero_python.api_client.serializer import serialize

from xero_python.accounting import AccountingApi, ContactPerson, Contacts
from xero_python.identity import IdentityApi

if settings.DEBUG is True:
    # Allow OAuth2 loop to run over HTTP (used for local testing only)
    os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1"

oauth = OAuth()
oauth.register(
    name='xero',
    client_id=settings.XERO_CLIENT_ID,
    client_secret=settings.XERO_CLIENT_SECRET,
    access_token_url="https://identity.xero.com/connect/token",
    access_token_params=None,
    refresh_token_url="https://identity.xero.com/connect/token",
    authorize_url="https://login.xero.com/identity/connect/authorize",
    authorize_params=None,
    api_base_url="https://api.xero.com/",
    client_kwargs= {'scope': 'offline_access openid profile email accounting.transactions accounting.transactions.read accounting.reports.read accounting.journals.read accounting.settings accounting.settings.read accounting.contacts accounting.contacts.read accounting.attachments accounting.attachments.read assets projects'},
    jwks_uri="https://login.xero.com/identity/.well-known/openid-configuration/jwks",
)

xero=oauth.create_client('xero')

api_client = ApiClient(
    Configuration(
        debug=settings.DEBUG,
        oauth2_token=OAuth2Token(
            client_id=settings.XERO_CLIENT_ID, client_secret=settings.XERO_CLIENT_SECRET,
        ),
    ),
    pool_threads=1,
)

def login(request):
    redirect_uri = request.build_absolute_uri('authorize/')
    return oauth.xero.authorize_redirect(request, redirect_uri)

def authorize(request):
    try:
        token = oauth.xero.authorize_access_token(request)
    except Exception as e:
        print(e)
        raise
    if token is None or token.get("access_token") is None:
        return HttpResponse("Access Denied")

    store_xero_oauth2_token(token)

    return redirect('/')

def logout(self):
    store_xero_oauth2_token(None)
    return redirect('/')


def xero_token_required(function):
    @wraps(function)
    def decorator(*args, **kwargs):

        token = obtain_xero_oauth2_token()
        time = datetime.datetime.now(timezone.utc)
        utc_time = time.replace(tzinfo=timezone.utc)
        utc_timestamp = utc_time.timestamp()

        if utc_timestamp > token['expires_at']:
            new_token = oauth.xero.fetch_access_token(
                refresh_token=token['refresh_token'],
                grant_type='refresh_token',
                client_id=settings.XERO_CLIENT_ID)
            store_xero_oauth2_token(new_token)
            token = new_token

        if not token:
            return redirect(reverse("login"))

        expires_in = token.get("expires_in", 0)
        now = time.time()

        now = datetime.datetime.now().timestamp()
        if now + expires_in > token.get("expires_at", now):
            token = oauth.xero.fetch_access_token(
                refresh_token=token['refresh_token'],
                grant_type='refresh_token',
                client_id=settings.XERO_CLIENT_ID)
            store_xero_oauth2_token(token)

        return function(*args, **kwargs)

    return decorator

@xero_token_required
def get_xero_tenant_id():
    identity_api = IdentityApi(api_client)
    for connection in identity_api.get_connections():
        if connection.tenant_type == "ORGANISATION":
            return connection.tenant_id

@xero_token_required
def get_invoices(request, i_ds=['']):
    try:
        xero_tenant_id = get_xero_tenant_id()
        accounting_api = AccountingApi(api_client)
        data = accounting_api.get_invoices(
            xero_tenant_id=xero_tenant_id, i_ds=i_ds
        )
    except AccountingBadRequestException as exception:
        output = "Error: " + exception.reason
        json = jsonify(exception.error_data)
    else:
        response = serialize_model(data)

    return HttpResponse(response)

@xero_token_required
def get_invoice(request, invoice_id):
    xero_tenant_id = get_xero_tenant_id()
    accounting_api = AccountingApi(api_client)

    data = accounting_api.get_invoice(
            xero_tenant_id, invoice_id
        )

    response = serialize_model(data)

    return HttpResponse(response)

@xero_token_required
def get_contacts(request, i_ds=['']):
    xero_tenant_id = get_xero_tenant_id()
    accounting_api = AccountingApi(api_client)

    data = accounting_api.get_contacts(
        xero_tenant_id=xero_tenant_id, i_ds=i_ds
    )

    response = serialize_model(data)

    return HttpResponse(response)

@xero_token_required
def get_contact(request, contact_id):
    xero_tenant_id = get_xero_tenant_id()
    accounting_api = AccountingApi(api_client)

    data = accounting_api.get_contact(
            xero_tenant_id, contact_id
        )

    response = serialize_model(data)

    return HttpResponse(response)

@xero_token_required
def get_items(request, i_ds=['']):
    xero_tenant_id = get_xero_tenant_id()
    accounting_api = AccountingApi(api_client)

    try:
        data = accounting_api.get_items(
            xero_tenant_id=xero_tenant_id, i_ds=i_ds
        )
    except AccountingBadRequestException as exception:
        output = "Error: " + exception.reason
        json_data = jsonify(exception.error_data)
        return HttpResponse(f"{output}\n{json_data}")
    else:
        response = serialize_model(data)

    return HttpResponse(response)

@api_client.oauth2_token_getter
def obtain_xero_oauth2_token():
    return caches['default'].get('xeroapi_token')

@api_client.oauth2_token_saver
def store_xero_oauth2_token(token):
    caches['default'].set('xeroapi_token', token, None)

class JSONEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, datetime):
            return o.isoformat()
        if isinstance(o, date):
            return o.isoformat()
        if isinstance(o, (uuid.UUID, Decimal)):
            return str(o)
        return super(JSONEncoder, self).default(o)

def serialize_model(model):
    return jsonify(serialize(model))


def jsonify(data):
    return json.dumps(data, sort_keys=True, indent=4, cls=JSONEncoder)`

PETOSS-285

Thanks for raising an issue, a ticket has been created to track your request

Resolved this issue. Was an env issue. Obvious in hindsight.