idds/tests/test_token_refresh.py

0001 import os
0002 import sys
0003 import logging
0004 import json
0005 import base64
0006 import datetime
0007
0008 from pandaclient import Client
0009
0010
0011 logging.basicConfig(stream=sys.stderr,
0012                     level=logging.DEBUG,
0013                     format='%(asctime)s\t%(threadName)s\t%(name)s\t%(levelname)s\t%(message)s')
0014
0015 logger = logging.getLogger('my_logger')
0016
0017
0018 # open_id = OpenIdConnect_Utils(None, log_stream=None, verbose=verbose)
0019 oidc = Client._Curl().get_oidc(logger)
0020 oidc.verbose = True
0021 print(oidc)
0022
0023
0024 def get_id_token():
0025     # oidc.cleanup()
0026     # this one should automatically refresh token
0027     status, output = oidc.run_device_authorization_flow()
0028     print(status)
0029     print(output)
0030     return output
0031
0032
0033 def show_token_info(id_token):
0034     enc = id_token.split('.')[1]
0035     enc += '=' * (-len(enc) % 4)
0036     dec = json.loads(base64.urlsafe_b64decode(enc.encode()))
0037     print(dec)
0038     exp_time = datetime.datetime.utcfromtimestamp(dec['exp'])
0039     print(exp_time)
0040
0041
0042 def get_refresh_token():
0043     # manually refresh token
0044     refresh_token = None
0045     token_file = oidc.get_token_path()
0046     if os.path.exists(token_file):
0047         with open(token_file) as f:
0048             data = json.load(f)
0049             if 'refresh_token' in data:
0050                 refresh_token = data['refresh_token']
0051     return refresh_token
0052
0053
0054 def get_config():
0055     auth_config = None
0056     endpoint_config = None
0057     # get auth config
0058     s, o = oidc.fetch_page(oidc.auth_config_url)
0059     if s:
0060         auth_config = o
0061         # get endpoint config
0062         s_1, o_1 = oidc.fetch_page(auth_config['oidc_config_url'])
0063         if s_1:
0064             endpoint_config = o_1
0065     return auth_config, endpoint_config
0066
0067
0068 auth_config, endpoint_config = get_config()
0069 # print(endpoint_config)
0070 refresh_token = get_refresh_token()
0071
0072 if endpoint_config and auth_config and refresh_token:
0073     status, output = oidc.refresh_token(endpoint_config['token_endpoint'], auth_config['client_id'],
0074                                         auth_config['client_secret'], refresh_token)
0075     print(status)
0076     print(output)
0077     show_token_info(output)
0078 else:
0079     output = get_id_token()
0080     print(output)
0081     show_token_info(output)