From 2c5b7da8165e3f8b90748e8ced1de0e3e8d52309 Mon Sep 17 00:00:00 2001 From: lazydancer Date: Sun, 22 Dec 2024 17:33:11 -0500 Subject: Initial --- backend/src/esi/api.py | 243 +++++++++++++++++++++++++++++++++++++++ backend/src/esi/esi_oauth_web.py | 87 ++++++++++++++ backend/src/esi/refresh_token.py | 43 +++++++ backend/src/esi/shared_flow.py | 144 +++++++++++++++++++++++ backend/src/esi/validate_jwt.py | 97 ++++++++++++++++ 5 files changed, 614 insertions(+) create mode 100644 backend/src/esi/api.py create mode 100644 backend/src/esi/esi_oauth_web.py create mode 100644 backend/src/esi/refresh_token.py create mode 100644 backend/src/esi/shared_flow.py create mode 100644 backend/src/esi/validate_jwt.py (limited to 'backend/src/esi') diff --git a/backend/src/esi/api.py b/backend/src/esi/api.py new file mode 100644 index 0000000..862b4bf --- /dev/null +++ b/backend/src/esi/api.py @@ -0,0 +1,243 @@ +""" +A simplified API for interacting with the EVE Online ESI API. +""" + +import datetime +import json +import logging +import os +import queue +import threading +import time +from pathlib import Path +from typing import Any + +import requests + +from .refresh_token import get_access_token_from_refresh_token + +# Configure logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# Constants +ESI_BASE_URL = "https://esi.evetech.net/latest" +DATASOURCE = "tranquility" +USER_AGENT_HEADER = {'User-Agent': 'Haul (Hyperspace Asset Unloading and Loading)'} +CACHE_DIR = Path(__file__).resolve().parent.parent / 'cache' +AUTH_FILE = CACHE_DIR / 'auth.json' +ERROR_LIMIT_THRESHOLD = 50 # ESI Error limit threshold +REQUEST_RATE_LIMIT = 0.05 # Time in seconds between requests to avoid rate limiting + + +def get_location() -> int: + """ + Gets the current location of the character. + + Returns: + The location ID of the character (station ID or solar system ID). + """ + character_id = os.getenv('character_id') + if not character_id: + logger.error("Character ID not found in environment variables.") + raise EnvironmentError("Character ID not set in environment variables.") + + url = f"{ESI_BASE_URL}/characters/{character_id}/location/" + + headers = { + "Accept": "application/json", + "Cache-Control": "no-cache", + "Authorization": f"Bearer {access_token()}", + **USER_AGENT_HEADER, + } + + params = { + "datasource": DATASOURCE, + } + + response = requests.get(url, params=params, headers=headers) + logger.debug(f"Request URL: {response.url}") + logger.debug(f"Response Status Code: {response.status_code}") + + if response.status_code == 401: + logger.info("Access token expired, refreshing token.") + access_token(force_refresh=True) + headers["Authorization"] = f"Bearer {access_token()}" + response = requests.get(url, params=params, headers=headers) + + if int(response.headers.get("X-Esi-Error-Limit-Remain", 100)) < ERROR_LIMIT_THRESHOLD: + logger.warning("Approaching ESI Error Limit.") + raise Exception("ESI Error Limit approaching threshold.") + + response.raise_for_status() + data: dict[str, Any] = response.json() + + location_id = data.get("station_id") or data.get("solar_system_id") + if not location_id: + logger.error("Location ID not found in response.") + raise ValueError("Unable to determine character's location.") + + logger.info(f"Character location ID: {location_id}") + return location_id + + +def set_waypoints(waypoints: list[int]) -> None: + """ + Sets the waypoints for the character's autopilot. + + Args: + waypoints: A list of waypoint IDs to set. + """ + if not waypoints: + logger.warning("No waypoints provided.") + return + + url = f"{ESI_BASE_URL}/ui/autopilot/waypoint/" + + headers = { + "Accept": "application/json", + "Cache-Control": "no-cache", + "Authorization": f"Bearer {access_token()}", + **USER_AGENT_HEADER, + } + + for index, waypoint in enumerate(waypoints): + params = { + "add_to_beginning": False, + "clear_other_waypoints": index == 0, + "destination_id": waypoint, + "datasource": DATASOURCE, + } + + try: + response = requests.post(url, params=params, headers=headers) + response.raise_for_status() + logger.info(f"Waypoint {waypoint} set successfully.") + time.sleep(1) # Rate limiting + except requests.HTTPError as e: + logger.error(f"Failed to set waypoint {waypoint}: {e}") + continue + + +def open_market_window(type_id: int) -> None: + """ + Opens the market window for a specific item type in the game client. + + Args: + type_id: The type ID of the item. + """ + url = f"{ESI_BASE_URL}/ui/openwindow/marketdetails/" + + headers = { + "Accept": "application/json", + "Cache-Control": "no-cache", + "Authorization": f"Bearer {access_token()}", + **USER_AGENT_HEADER, + } + + params = { + "type_id": type_id, + "datasource": DATASOURCE, + } + + try: + response = requests.post(url, params=params, headers=headers) + response.raise_for_status() + logger.info(f"Market window opened for type ID {type_id}.") + except requests.HTTPError as e: + logger.error(f"Failed to open market window for type ID {type_id}: {e}") + + +def get_region_orders(region_id: int) -> tuple[list[dict[str, Any]], str]: + """ + Retrieves all market orders for a given region. + + Args: + region_id: The ID of the region. + + Returns: + A tuple containing the list of orders and the expiry time. + """ + def request_orders(region: int, page: int, q: queue.Queue) -> None: + url = f"{ESI_BASE_URL}/markets/{region}/orders/" + params = { + 'datasource': DATASOURCE, + 'order_type': 'all', + 'page': page, + } + headers = USER_AGENT_HEADER + + try: + response = requests.get(url, params=params, headers=headers) + response.raise_for_status() + logger.debug(f"Retrieved page {page} for region {region}.") + q.put(response) + except requests.HTTPError as e: + logger.error(f"Error fetching page {page} for region {region}: {e}") + + order_queue = queue.Queue() + request_orders(region_id, 1, order_queue) # Get the first page + first_response = order_queue.get() + pages = int(first_response.headers.get('X-Pages', '1')) + expires = first_response.headers.get('Expires', '') + + if pages > 1: + threads = [] + for page in range(2, pages + 1): + thread = threading.Thread(target=request_orders, args=(region_id, page, order_queue)) + threads.append(thread) + thread.start() + time.sleep(REQUEST_RATE_LIMIT) # Rate limiting + + for thread in threads: + thread.join() + + results: list[dict[str, Any]] = first_response.json() + while not order_queue.empty(): + response = order_queue.get() + results.extend(response.json()) + + # Filter out orders with zero remaining volume + results = [order for order in results if order.get('volume_remain', 0) > 0] + + # Add region ID to each order + for order in results: + order['region_id'] = region_id + + logger.info(f"Retrieved {len(results)} orders for region {region_id}.") + return results, expires + + +def access_token(force_refresh: bool = False) -> str: + """ + Retrieves an active access token, refreshing it if necessary. + + Args: + force_refresh: If True, forces a refresh of the access token. + + Returns: + The active access token as a string. + """ + try: + with AUTH_FILE.open("r", encoding="utf-8") as f: + auth: dict[str, Any] = json.load(f) + except FileNotFoundError: + logger.error(f"Authentication file not found at {AUTH_FILE}.") + raise FileNotFoundError("Authentication file is missing.") + + expires_at = auth.get("expires_at", 0) + current_time = datetime.datetime.utcnow().timestamp() + if not force_refresh and expires_at > current_time: + return auth.get("access_token", "") + + logger.info("Access token expired or force refresh requested, refreshing token.") + new_auth = get_access_token_from_refresh_token(auth["refresh_token"]) + + # Update expiry time (assuming token is valid for 20 minutes) + new_auth["expires_at"] = current_time + 1200 + + with AUTH_FILE.open("w", encoding="utf-8") as f: + json.dump(new_auth, f) + + logger.info("Access token refreshed successfully.") + return new_auth["access_token"] diff --git a/backend/src/esi/esi_oauth_web.py b/backend/src/esi/esi_oauth_web.py new file mode 100644 index 0000000..5440bf9 --- /dev/null +++ b/backend/src/esi/esi_oauth_web.py @@ -0,0 +1,87 @@ +""" +Python 3 Web Application OAuth 2.0 Example. + +This script demonstrates how to perform the OAuth 2.0 authorization code flow +for a web-based application using the EVE Online SSO. + +Prerequisites: + * Create an SSO application at https://developers.eveonline.com with the + scope "esi-characters.read_blueprints.v1" and the callback URL + "https://localhost/callback/". Note: Never use localhost as a callback + in production applications. + * Have a Python 3 environment available (possibly by using a virtual + environment: https://virtualenv.pypa.io/en/stable/). + * Install the required packages by running: + pip install -r requirements.txt + +To run this example, make sure you have completed the prerequisites and then +run the following command: + +>>> python esi_oauth_web.py + +Then follow the prompts. +""" + +import base64 +import os +from typing import Optional + +from dotenv import load_dotenv + +from shared_flow import ( + handle_sso_token_response, + print_auth_url, + send_token_request, +) + + +def main() -> None: + """ + Guides you through a local example of the OAuth 2.0 web flow. + """ + print( + "This program will take you through an example OAuth 2.0 flow " + "that you should be using if you are building a web-based " + "application. Follow the prompts and enter the information requested." + ) + + client_id: Optional[str] = os.getenv('CLIENT_ID') + if not client_id: + raise ValueError("CLIENT_ID not found in environment variables.") + + print_auth_url(client_id) + + auth_code: str = input('Copy the "code" query parameter and enter it here: ') + + # Basic auth can be handled by the requests library with the auth parameter, + # but we're constructing it manually for educational purposes. + secret_key: Optional[str] = os.getenv('SECRET_KEY') + if not secret_key: + raise ValueError("SECRET_KEY not found in environment variables.") + + user_pass: str = f"{client_id}:{secret_key}" + basic_auth: str = base64.urlsafe_b64encode(user_pass.encode('utf-8')).decode() + auth_header: str = f"Basic {basic_auth}" + + form_values: dict[str, str] = { + "grant_type": "authorization_code", + "code": auth_code, + } + + headers: dict[str, str] = {"Authorization": auth_header} + + print( + "\nThe following request uses basic authentication, as a web-based " + "app you should do the same." + ) + + input("\nPress Enter to continue...") + + res = send_token_request(form_values, add_headers=headers) + + handle_sso_token_response(res) + + +if __name__ == "__main__": + load_dotenv() + main() diff --git a/backend/src/esi/refresh_token.py b/backend/src/esi/refresh_token.py new file mode 100644 index 0000000..0b3e4a3 --- /dev/null +++ b/backend/src/esi/refresh_token.py @@ -0,0 +1,43 @@ +import base64 +import datetime +import os + +from .shared_flow import send_token_request + +def get_access_token_from_refresh_token(refresh_token: str) -> dict[str, any]: + """ + Obtains a new access token using a refresh token. + + This function uses the provided refresh token to request a new access token + from the authentication server. It updates the expiration time of the token. + + Args: + refresh_token: The refresh token used to obtain a new access token. + + Returns: + A dictionary containing the new access token data, including 'expires_at'. + """ + client_id = os.getenv('CLIENT_ID') + secret_key = os.getenv('SECRET_KEY') + + if not client_id or not secret_key: + raise EnvironmentError("CLIENT_ID and SECRET_KEY must be set in environment variables.") + + user_pass = f"{client_id}:{secret_key}" + basic_auth = base64.urlsafe_b64encode(user_pass.encode('utf-8')).decode() + auth_header = f"Basic {basic_auth}" + + form_values = { + "grant_type": "refresh_token", + "refresh_token": refresh_token, + } + + headers = {"Authorization": auth_header} + + response = send_token_request(form_values, add_headers=headers) + data = response.json() + + current_timestamp = int(datetime.datetime.now().timestamp()) + data["expires_at"] = current_timestamp + data.get("expires_in", 0) + + return data diff --git a/backend/src/esi/shared_flow.py b/backend/src/esi/shared_flow.py new file mode 100644 index 0000000..1d89fb1 --- /dev/null +++ b/backend/src/esi/shared_flow.py @@ -0,0 +1,144 @@ +""" +Contains shared OAuth 2.0 flow functions for EVE Online examples. + +This module contains shared functions used between the two different OAuth 2.0 +flows recommended for web-based and mobile/desktop applications. The functions +found here are used by the OAuth 2.0 examples contained in this project. +""" + +import requests + +from .validate_jwt import validate_eve_jwt + + +def print_auth_url(client_id: str, code_challenge: str = "") -> None: + """ + Prints the URL to redirect users to for OAuth authorization. + + Args: + client_id: The client ID of an EVE SSO application. + code_challenge: A PKCE code challenge (optional). + """ + base_auth_url = "https://login.eveonline.com/v2/oauth/authorize/" + params = { + "response_type": "code", + "redirect_uri": "https://localhost/callback/", + "client_id": client_id, + "state": "unique-state", + } + + # Define the required scopes + scopes = [ + "esi-assets.read_assets.v1", + "esi-location.read_location.v1", + "esi-location.read_online.v1", + "esi-ui.open_window.v1", + "esi-ui.write_waypoint.v1", + ] + scope_param = "%20".join(scopes) + params["scope"] = scope_param + + if code_challenge: + params["code_challenge"] = code_challenge + params["code_challenge_method"] = "S256" + + # Construct the full authorization URL + request = requests.Request('GET', base_auth_url, params=params) + prepared_request = request.prepare() + full_auth_url = prepared_request.url + + print( + "\nOpen the following link in your browser:\n\n" + f"{full_auth_url}\n\n" + "Once you have logged in as a character, you will be redirected to " + "https://localhost/callback/." + ) + + +def send_token_request(form_values: dict[str, str], add_headers: dict[str, str] = {}) -> requests.Response: + """ + Sends a request for an authorization token to the EVE SSO. + + Args: + form_values: A dictionary containing the form-encoded values to send with the request. + add_headers: A dictionary containing additional headers to send (optional). + + Returns: + A requests.Response object containing the server's response. + """ + headers = { + "Content-Type": "application/x-www-form-urlencoded", + "Host": "login.eveonline.com", + } + headers.update(add_headers) + + response = requests.post( + "https://login.eveonline.com/v2/oauth/token", + data=form_values, + headers=headers, + ) + + print( + f"Request sent to URL {response.url} with headers {headers} and form values: {form_values}\n" + ) + response.raise_for_status() + + return response + + +def handle_sso_token_response(sso_response: requests.Response) -> None: + """ + Handles the authorization code response from the EVE SSO. + + Args: + sso_response: A requests.Response object from the EVE SSO /v2/oauth/token endpoint. + """ + if sso_response.status_code == 200: + data = sso_response.json() + access_token = data.get("access_token") + + print("\nVerifying access token JWT...") + + jwt = validate_eve_jwt(access_token) + character_id = jwt["sub"].split(":")[2] + character_name = jwt["name"] + blueprint_url = f"https://esi.evetech.net/latest/characters/{character_id}/blueprints/" + + print( + f"\nSuccess! Here is the payload received from the EVE SSO: {data}\n" + "You can use the access token to make an authenticated request to " + f"{blueprint_url}" + ) + + input("\nPress Enter to have this program make the request for you:") + + headers = { + "Authorization": f"Bearer {access_token}" + } + + res = requests.get(blueprint_url, headers=headers) + print( + f"\nMade request to {blueprint_url} with headers: {res.request.headers}" + ) + res.raise_for_status() + + blueprints = res.json() + print(f"\n{character_name} has {len(blueprints)} blueprints.") + else: + print( + "\nSomething went wrong! Please re-read the comments at the top of this " + "file and ensure you completed all the prerequisites, then try again. " + "Here's some debug info to help you out:" + ) + request_info = sso_response.request + print( + f"\nSent request with URL: {request_info.url}\n" + f"Body: {request_info.body}\n" + f"Headers: {request_info.headers}" + ) + print(f"\nSSO response code is: {sso_response.status_code}") + try: + error_data = sso_response.json() + print(f"\nSSO response JSON is: {error_data}") + except ValueError: + print("\nNo JSON response received from the SSO.") diff --git a/backend/src/esi/validate_jwt.py b/backend/src/esi/validate_jwt.py new file mode 100644 index 0000000..054c367 --- /dev/null +++ b/backend/src/esi/validate_jwt.py @@ -0,0 +1,97 @@ +""" +Validates a given JWT access token originating from the EVE SSO. + +Prerequisites: + * Have a Python 3 environment available (possibly using a virtual environment). + * Run pip install -r requirements.txt with this directory as your root. + +This script can be run by executing: + +>>> python validate_jwt.py + +and providing a JWT access token that you have retrieved from the EVE SSO. +""" + +import sys +from typing import Any + +import requests +from jose import jwt +from jose.exceptions import ExpiredSignatureError, JWTError + +SSO_META_DATA_URL = "https://login.eveonline.com/.well-known/oauth-authorization-server" +JWK_ALGORITHM = "RS256" +JWK_ISSUERS = ("login.eveonline.com", "https://login.eveonline.com") +JWK_AUDIENCE = "EVE Online" + + +def validate_eve_jwt(token: str) -> dict[str, Any]: + """ + Validates a JWT access token retrieved from the EVE SSO. + + Args: + token: A JWT access token originating from the EVE SSO. + + Returns: + The contents of the validated JWT access token. + + Raises: + RuntimeError: If required data is missing from the SSO endpoints. + jose.exceptions.JWTError: If the token is invalid or expired. + """ + response = requests.get(SSO_META_DATA_URL) + response.raise_for_status() + metadata = response.json() + + jwks_uri = metadata.get("jwks_uri") + if not jwks_uri: + raise RuntimeError(f"Invalid data received from the SSO metadata endpoint: {metadata}") + + response = requests.get(jwks_uri) + response.raise_for_status() + jwks_data = response.json() + + jwk_keys = jwks_data.get("keys") + if not jwk_keys: + raise RuntimeError(f"Invalid data received from the JWKS endpoint: {jwks_data}") + + jwk = next((item for item in jwk_keys if item["alg"] == JWK_ALGORITHM), None) + if not jwk: + raise RuntimeError(f"No JWK found with algorithm {JWK_ALGORITHM}") + + # Decode and validate the token + contents = jwt.decode( + token=token, + key=jwk, + algorithms=[JWK_ALGORITHM], + issuer=JWK_ISSUERS, + audience=JWK_AUDIENCE, + ) + return contents + + +def main() -> None: + """ + Entry point of the script. Prompts the user for a JWT token and validates it. + """ + token = input("Enter an access token to validate: ") + + try: + token_contents = validate_eve_jwt(token) + except ExpiredSignatureError: + print("The JWT token has expired.") + sys.exit(1) + except JWTError as e: + print(f"The JWT token is invalid: {e}") + sys.exit(1) + except RuntimeError as e: + print(f"An error occurred: {e}") + sys.exit(1) + else: + print("\nThe contents of the access token are:") + for key, value in token_contents.items(): + print(f"{key}: {value}") + + +if __name__ == "__main__": + main() -- cgit v1.3