5 files changed, 614 insertions, 0 deletions
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()