| 57 | |
| 58 | |
| 59 | class EsiAccess: |
| 60 | server_meta = {} |
| 61 | def __init__(self): |
| 62 | self.settings = EsiSettings.getInstance() |
| 63 | self.default_server_name = self.settings.get('server') |
| 64 | self.default_server_base = config.supported_servers[self.default_server_name] |
| 65 | # session request stuff |
| 66 | self._session = Session() |
| 67 | self._basicHeaders = { |
| 68 | 'Accept': 'application/json', |
| 69 | 'User-Agent': ( |
| 70 | 'pyfa v{}'.format(config.version) |
| 71 | ) |
| 72 | } |
| 73 | self._session.headers.update(self._basicHeaders) |
| 74 | self._session.proxies = NetworkSettings.getInstance().getProxySettingsInRequestsFormat() |
| 75 | |
| 76 | self.mem_cached_session = {} |
| 77 | |
| 78 | # Set up cached session. This is only used for SSO meta data for now, but can be expanded to actually handle |
| 79 | # various ESI caching (using ETag, for example) in the future |
| 80 | self.cached_session = CachedSession( |
| 81 | os.path.join(config.savePath, config.ESI_CACHE), |
| 82 | backend="sqlite", |
| 83 | cache_control=True, # Use Cache-Control headers for expiration, if available |
| 84 | expire_after=timedelta(days=1), # Otherwise expire responses after one day |
| 85 | stale_if_error=True, # In case of request errors, use stale cache data if possible |
| 86 | ) |
| 87 | self.cached_session.headers.update(self._basicHeaders) |
| 88 | self.cached_session.proxies = NetworkSettings.getInstance().getProxySettingsInRequestsFormat() |
| 89 | self.init(self.default_server_base) |
| 90 | |
| 91 | def init(self, server_base): |
| 92 | self.server_base: config.ApiServer = server_base |
| 93 | self.server_name = self.server_base.name |
| 94 | try: |
| 95 | meta_call = self.cached_session.get("https://%s/.well-known/oauth-authorization-server" % self.server_base.sso) |
| 96 | except: |
| 97 | # The http data of expire_after in evepc.163.com is -1 |
| 98 | meta_call = requests.get("https://%s/.well-known/oauth-authorization-server" % self.server_base.sso) |
| 99 | |
| 100 | meta_call.raise_for_status() |
| 101 | self.server_meta = meta_call.json() |
| 102 | |
| 103 | try: |
| 104 | jwks_call = self.cached_session.get(self.server_meta["jwks_uri"]) |
| 105 | except: |
| 106 | jwks_call = requests.get(self.server_meta["jwks_uri"]) |
| 107 | |
| 108 | jwks_call.raise_for_status() |
| 109 | self.jwks = jwks_call.json() |
| 110 | |
| 111 | @property |
| 112 | def sso_url(self): |
| 113 | return 'https://%s/v2' % self.server_base.sso |
| 114 | |
| 115 | @property |
| 116 | def esi_url(self): |