Source code for diva.data.cache_b.get_token

#! /usr/bin/env python3

from pathlib import Path
from urllib.parse import parse_qs, urlparse

import requests
from lxml import html

user = "christophe.taillandier"
password = None


# password sur teams



def get_token_cache_b(user, password):
    SERVICE_URL = "https://cacheb.dcms.destine.eu/"
    IAM_URL = 'https://auth.destine.eu'
    IAM_REALM = 'desp'
    IAM_CLIENT = 'edh-public'

    with requests.Session() as s:

        # Get the auth url
        response = s.get(
            url=IAM_URL
                + "/realms/"
                + IAM_REALM
                + "/protocol/openid-connect/auth",
            params={
                "client_id": IAM_CLIENT,
                "redirect_uri": SERVICE_URL,
                "scope": "openid offline_access",
                "response_type": "code",
            },
        )
        response.raise_for_status()
        auth_url = html.fromstring(response.content.decode()).forms[0].action

        # Login and get auth code
        login = s.post(
            auth_url,
            data={
                "username": user,
                "password": password,
            },
            allow_redirects=False,
        )

        # We expect a 302, a 200 means we got sent back to the login page and there's probably an error message
        if login.status_code == 200:
            tree = html.fromstring(login.content)
            error_message_element = tree.xpath('//span[@id="input-error"]/text()')
            error_message = (
                error_message_element[0].strip()
                if error_message_element
                else "Error message not found"
            )
            return 1, error_message

        if login.status_code != 302:
            return 1, "Login failed"

        auth_code = parse_qs(urlparse(login.headers["Location"]).query)["code"][0]

        # Use the auth code to get the token
        response = requests.post(
            IAM_URL
            + "/realms/"
            + IAM_REALM
            + "/protocol/openid-connect/token",
            data={
                "client_id": IAM_CLIENT,
                "redirect_uri": SERVICE_URL,
                "code": auth_code,
                "grant_type": "authorization_code",
                "scope": "",
            },
        )

        if response.status_code != 200:
            return 1, "Failed to get token"

        # instead of storing the access token, we store the offline_access (kind of "refresh") token
        token = response.json()["refresh_token"]

        with open(Path.home() / ".netrc", "w") as fp:
            txt = f"""# Authenticating on https://auth.destine.eu with user {user}

machine cacheb.dcms.destine.eu
    login anonymous
    password {token}
"""
            fp.write(txt)

        return 0, "Token successfully updated!"