Test for LeighCurran/AuroraPlus#2

auroraplus-oauth.py

#!/usr/bin/env python

import base64
import hashlib
import json
import logging
import random
import requests
import string
import uuid

from requests_oauthlib import OAuth2Session
from oauthlib.oauth2.rfc6749.errors import MissingTokenError

logging.basicConfig(level=logging.DEBUG)

AUTHORIZE_URL = 'https://customers.auroraenergy.com.au/' \
    + 'auroracustomers1p.onmicrosoft.com/b2c_1a_sign_in/oauth2/v2.0/authorize'
TOKEN_URL = 'https://customers.auroraenergy.com.au/' \
    + 'auroracustomers1p.onmicrosoft.com/b2c_1a_sign_in/oauth2/v2.0/token'
BEARER_TOKEN_URL = 'https://api.auroraenergy.com.au/api/identity/LoginToken'


def _get_token():
    client_id = '2ff9da64-8629-4a92-a4b6-850a3f02053d'
    redirect_uri = 'https://my.auroraenergy.com.au/login/redirect'
    client_secret = ''

    scope = ['openid', 'profile', 'offline_access']

    oauth = OAuth2Session(client_id, redirect_uri=redirect_uri, scope=scope)

    state = {
        "id": str(uuid.uuid4()),  # "c28abb6e-a8db-42a2-8605-82149e19a5d3",
        "meta": {"interactionType": "redirect"},
    }

    code_verifier = ''.join([random.choice(
        string.ascii_letters
        + string.digits
        + '-_')
        for i in range(43)])
    code_challenge = base64.urlsafe_b64encode(
        hashlib.sha256(code_verifier.encode()).digest()
    ).strip(b'=')

    authorization_url, state = oauth.authorization_url(
        AUTHORIZE_URL,
        client_request_id=uuid.uuid4(),
        # response_mode='fragment',
        # x_client_SKU='msal.js.browser',
        # x_client_VER='2.38.2',
        client_info=1,
        code_challenge=code_challenge,
        code_challenge_method='S256',
        # nonce=uuid.uuid4(),
        state=base64.encodebytes(json.dumps(state).encode()))

    print(f'Please go to {authorization_url} and authorize access.')
    authorization_response = input('Enter the full callback URL: ')

    # useful if copying URL from the web app, or when response_mode='fragment'
    # authorization_response = '?'.join(authorization_response.split('#'))

    def _include_access_token(r):
        rjs = r.json()
        id_token = rjs.get('id_token')

        atr = requests.post(BEARER_TOKEN_URL,
                            json={'token': id_token}
                            )
        access_token = atr.json().get('accessToken')

        rjs.update({
            'access_token': access_token.split()[1],
            'scope': 'openid profile offline_access',
        })

        r._content = json.dumps(rjs).encode()

        return r

    oauth.register_compliance_hook(
        'access_token_response',
        _include_access_token)

    token = oauth.fetch_token(
        TOKEN_URL,
        authorization_response=authorization_response,
        code_verifier=code_verifier,
    )

    return oauth


if __name__ == '__main__':
    oauth = _get_token()

    headers = {
        'user-agent': 'not that',
    }

    r = oauth.get('https://api.auroraenergy.com.au/api/customers/current',
                  headers=headers)

    customer_id = r.json()[0]['CustomerID']
    service_id = r.json()[0]['Premises'][0]['ServiceAgreementID']

    r = oauth.get('https://api.auroraenergy.com.au/api/usage/day'
                  + f'?serviceAgreementID={service_id}'
                  + f'&customerId={customer_id}'
                  + '&index=-1',
                  headers=headers)

    print(r.json())