NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

discovery_hacks.py (2521B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\clientRefreshToken\discovery_hacks.py
      2 import requests
      3 import uthread2
      4 import logging
      5 import urlparse
      6 logger = logging.getLogger(__name__)
      7 WELL_KNOWN_PATH = '/.well-known/oauth-authorization-server/'
      8 DEFAULT_SCHEME = 'https'
      9 FALLBACK_FQDN = '{}://login.eveonline.com'.format(DEFAULT_SCHEME)
     10 FALLBACK_WELL_KNOWN_URL = urlparse.urljoin(FALLBACK_FQDN, WELL_KNOWN_PATH)
     11 FALLBACK_TOKEN_ENDPOINT = urlparse.urljoin(FALLBACK_FQDN, '/v2/oauth/token')
     12 FALLBACK_CLIENT_ID = 'eveLauncherTQ'
     13 
     14 def discover_refresh_parameters_from_user_token(user_token_payload):
     15     try:
     16         token_endpoint = discover_token_endpoint_from_user_token(user_token_payload)
     17         client_id = discover_client_id_from_user_token(user_token_payload)
     18         return (client_id, token_endpoint)
     19     except Exception as e:
     20         logger.exception('Failed to discover refresh parameters, using fallbacks', extra={'fallback_client_id': FALLBACK_CLIENT_ID,
     21          'fallback_token_endpoint': FALLBACK_TOKEN_ENDPOINT})
     22         return (FALLBACK_CLIENT_ID, FALLBACK_TOKEN_ENDPOINT)
     23 
     24 
     25 def discover_client_id_from_user_token(user_token_payload):
     26     return user_token_payload['azp']
     27 
     28 
     29 def discover_token_endpoint_from_user_token(user_token_payload):
     30     well_known_url = create_well_known_url_from_iss_claim(user_token_payload.get('iss'))
     31     max_retries = 10
     32     attempts = 0
     33     while True:
     34         attempts += 1
     35         if attempts >= 10:
     36             raise RuntimeError('max retries ({}) attempted'.format(max_retries))
     37         response = requests.get(well_known_url)
     38         if not response.ok:
     39             uthread2.sleep(1)
     40             continue
     41         content = response.json()
     42         token_endpoint = content['token_endpoint']
     43         if token_endpoint:
     44             return token_endpoint
     45 
     46 
     47 def create_well_known_url_from_iss_claim(issuer):
     48     iss_res = urlparse.urlparse(issuer)
     49     if not iss_res.netloc:
     50         if not iss_res.path:
     51             raise RuntimeError('Malformed `iss` claim in SSO token')
     52         iss_res = iss_res._replace(netloc=iss_res.path)
     53     if not iss_res.scheme:
     54         iss_res = iss_res._replace(scheme=DEFAULT_SCHEME)
     55     if iss_res.path != WELL_KNOWN_PATH:
     56         iss_res = iss_res._replace(path=WELL_KNOWN_PATH)
     57     return iss_res.geturl()
     58 
     59 
     60 if __name__ == '__main__':
     61     for test_val in ('', '?foo=bar', '#baz'):
     62         try:
     63             create_well_known_url_from_iss_claim(test_val)
     64         except RuntimeError as e:
     65             pass