Source code for pykubegrader.tokens.validate_token
import os
import sys
from typing import Optional
import requests
from requests.auth import HTTPBasicAuth
[docs]
def get_jhub_user():
"""
Fetches the JupyterHub user from the environment.
"""
jhub_user = os.getenv("JUPYTERHUB_USER")
if jhub_user is None:
raise ValueError("JupyterHub user not found")
return jhub_user
[docs]
class TokenValidationError(Exception):
"""
Custom exception raised when the token validation fails.
"""
def __init__(self, message: Optional[str] = None):
"""
Initialize the exception with an optional message.
Args:
message (str, optional): The error message to display. Defaults to None.
"""
super().__init__(message)
[docs]
def get_credentials() -> dict[str, str]:
"""
Fetch the username and password from environment variables.
Returns:
dict: A dictionary containing 'username' and 'password'.
"""
username = os.getenv("user_name_student")
password = os.getenv("keys_student")
if not username or not password:
raise ValueError(
"Environment variable 'user_name_student' or 'keys_student' not set"
)
return {"username": username, "password": password}
[docs]
def validate_token(
token: Optional[str] = None,
assignment: Optional[str] = None,
) -> None:
if token:
os.environ["TOKEN"] = token # If token passed, set env var
else:
token = os.getenv("TOKEN") # Otherwise try to get from env
if not token:
print("Error: No token provided", file=sys.stderr)
sys.exit(1)
# Get endpoint URL
base_url = os.getenv("DB_URL")
if not base_url:
print("Error: Environment variable 'DB_URL' not set", file=sys.stderr)
sys.exit(1)
# Construct endpoint with optional parameters
endpoint = f"{base_url.rstrip('/')}/validate-token/{token}"
# Build query parameters
params = {}
if assignment:
params["assignment"] = assignment
params["student_id"] = get_jhub_user()
# Get credentials
try:
credentials = get_credentials()
except ValueError as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
username = credentials["username"]
password = credentials["password"]
basic_auth = HTTPBasicAuth(username, password)
try:
# Send request with optional query parameters
response = requests.get(
url=endpoint, auth=basic_auth, timeout=10, params=params
)
response.raise_for_status()
detail = response.json().get("detail", response.text)
return
except requests.exceptions.HTTPError as e:
detail = e.response.json().get("detail", e.response.text)
print(f"Error: {detail}", file=sys.stderr)
except requests.exceptions.RequestException as e:
print(f"Error: Request failed: {e}", file=sys.stderr)
except Exception as e:
print(f"Unexpected error: {e}", file=sys.stderr)
sys.exit(1) # If we reached here, something went wrong
# Example usage
if __name__ == "__main__":
token = "test"
try:
validate_token(token)
print("Token is valid")
except TokenValidationError as e:
print(f"Token validation failed: {e}")