mra/random-access
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

significant restructuring, format all code

Browse source
This commit is contained in:
Micha R. Albert 2025-07-08 11:04:36 -04:00
parent 16e840fb78
commit 43b2e33551
Signed by: mra
SSH key fingerprint: SHA256:2JB0fGfy7m2HQXAzvSXXKm7wPTj9Z60MOjFOQGM2Y/E
16 changed files with 698 additions and 449 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.env.example
View file

@ -13,6 +13,7 @@ AIRTABLE_USERS_TABLE=your_users_table_id_here
AIRTABLE_SESSIONS_TABLE=your_sessions_table_id_here
AIRTABLE_ITEMS_TABLE=your_items_table_id_here
AIRTABLE_ITEM_ADDONS_TABLE=your_item_addons_table_id_here
AIRTABLE_ITEM_INSTANCES_TABLE=your_item_instances_table_id_here
# Application Settings
APP_BASE_URL=http://localhost:8000

1
Dockerfile
View file

@ -21,6 +21,7 @@ COPY pyproject.toml ./
COPY LICENSE ./
COPY README.md ./
COPY src/ ./src/
COPY templates/ ./templates/
# Install project and dependencies using Hatch
RUN hatch build -t wheel && \

2
src/random_access/__init__.py
View file

@ -1,4 +1,4 @@
"""Random Access - A FastAPI app with secure authentication, Redis caching, and Airtable integration."""
"""Random Access - A FastAPI backend for Hack Club's Random Access grant initiative."""
__version__ = "0.0.1"
__author__ = "Micha R. Albert"

49
src/random_access/auth.py
View file

@ -4,12 +4,19 @@ import hashlib
import hmac
from datetime import datetime, timedelta
from fastapi import HTTPException, status
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pyairtable.formulas import match
from random_access.security import validate_bearer_token_format, create_safe_error_response
from random_access.settings import settings
# Create HTTPBearer security scheme
security = HTTPBearer(
scheme_name="Bearer Token",
description="Authentication token from the game session",
auto_error=True,
)
def hash_token(token: str) -> str:
"""Hash a token using SHA256."""
@ -19,6 +26,7 @@ def hash_token(token: str) -> str:
async def get_session_by_token(token: str, sessions_table) -> dict | None:
"""Get a session by its hashed token (now using cached version)."""
from random_access.database import get_session_by_token_cached
return await get_session_by_token_cached(token, sessions_table)
@ -29,7 +37,7 @@ def is_session_expired(session: dict) -> bool:
return True
try:
created_time = datetime.fromisoformat(created_at.replace('Z', '+00:00'))
created_time = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
expiry_time = created_time + timedelta(hours=settings.session_ttl_hours)
return datetime.now().replace(tzinfo=created_time.tzinfo) > expiry_time
except (ValueError, TypeError):
@ -69,38 +77,35 @@ async def decode_oidc_state(state: str, sessions_table) -> tuple[str, str, str]:
return game_id, secure_token, session["id"]
async def extract_and_validate_auth(authorization: str, sessions_table, users_table):
async def extract_and_validate_auth(
credentials: HTTPAuthorizationCredentials, sessions_table, users_table
):
"""Extract and validate authentication token, return game_id, session, and user."""
try:
full_token = validate_bearer_token_format(authorization)
except HTTPException:
raise
# The credentials.credentials already contains the token without "Bearer " prefix
full_token = credentials.credentials
try:
game_id, token_id, session_rec_id = await decode_oidc_state(full_token, sessions_table)
game_id, token_id, session_rec_id = await decode_oidc_state(
full_token, sessions_table
)
except ValueError as e:
if settings.is_production:
detail = "Invalid authentication token"
else:
detail = str(e)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=detail
)
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=detail)
# Get session using the hashed token
session = sessions_table.first(formula=match({"Token": hash_token(token_id)}))
if not session:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid session token"
status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid session token"
)
# Check session expiration
if is_session_expired(session):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Session has expired"
status_code=status.HTTP_401_UNAUTHORIZED, detail="Session has expired"
)
# Get the user from the session
@ -108,17 +113,19 @@ async def extract_and_validate_auth(authorization: str, sessions_table, users_ta
if not user_id:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Session not linked to a user"
detail="Session not linked to a user",
)
user = users_table.get(user_id[0]) # User is stored as a list in Airtable
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
)
return game_id, session, user
# Removed unused SessionStore class for cleaner codebase
# Create a dependency function that combines HTTPBearer with our authentication logic
def get_auth_credentials(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""Dependency to get HTTPBearer credentials."""
return credentials

3
src/random_access/cli.py
View file

@ -1,12 +1,15 @@
import click
import uvicorn
from random_access.main import app
@click.group()
def cli():
"""Random Access Server CLI."""
pass
@cli.command()
@click.option("--host", default="127.0.0.1", help="Host to bind to.")
@click.option("--port", default=8000, help="Port to bind to.")

218
src/random_access/database.py
View file

@ -4,8 +4,8 @@ import asyncio
import datetime
import hashlib
import logging
import os
from typing import Any, Dict, List, Optional
import uuid
from typing import List, Optional
from aiocache import Cache, cached
from aiocache.serializers import PickleSerializer
@ -19,14 +19,118 @@ logger = logging.getLogger("uvicorn.error")
# Global queue for write operations
write_queue: asyncio.Queue = asyncio.Queue()
# Expected table schemas for validation
EXPECTED_SCHEMAS = {
"users": {
"required_fields": ["Slack ID", "Display Name", "Email"],
"optional_fields": ["Last Login", "Created"],
},
"sessions": {
"required_fields": ["Token"],
"optional_fields": ["User", "Game", "Created"],
},
"submissions": {
"required_fields": ["Game Name"],
"optional_fields": ["Description", "Created"],
},
"items": {
"required_fields": ["Name", "Type", "Level", "Rarity"],
"optional_fields": ["Description", "Game Name (from Games)"],
},
"item_instances": {
"required_fields": ["ID", "User", "Item"],
"optional_fields": ["Acquired"],
},
"item_addons": {
"required_fields": ["Name"],
"optional_fields": ["Description", "Item"],
},
}
def get_airtable_base():
"""Get the Airtable base instance."""
return AirtableApi(os.environ["AIRTABLE_PAT"]).base(os.environ["AIRTABLE_BASE"])
return AirtableApi(settings.airtable_pat).base(settings.airtable_base)
def get_table(base, name: str):
"""Get a specific Airtable table."""
return base.table(os.environ[f"AIRTABLE_{name.upper()}_TABLE"])
table = getattr(settings, f"airtable_{name}_table")
print(table)
return base.table(table)
async def validate_table_schema(table, table_name: str, expected_schema: dict) -> bool:
"""Validate that a table has the expected schema."""
try:
# Get a sample record to check field structure
# We'll try to get the first record, if none exist, we'll create a test record and delete it
records = table.all(max_records=1)
if not records:
logger.warning(
f"Table '{table_name}' is empty - cannot validate schema without records"
)
return True # Assume valid if empty
record = records[0]
available_fields = set(record.get("fields", {}).keys())
# Check required fields
required_fields = set(expected_schema.get("required_fields", []))
missing_required = required_fields - available_fields
if missing_required:
logger.error(
f"Table '{table_name}' missing required fields: {missing_required}"
)
return False
# Log optional fields that are missing (as info, not error)
optional_fields = set(expected_schema.get("optional_fields", []))
missing_optional = optional_fields - available_fields
if missing_optional:
logger.info(
f"Table '{table_name}' missing optional fields: {missing_optional}"
)
logger.info(f"Table '{table_name}' schema validation passed")
return True
except Exception as e:
logger.error(f"Error validating schema for table '{table_name}': {e}")
return False
async def validate_all_schemas() -> bool:
"""Validate schemas for all tables at startup."""
logger.info("Starting table schema validation...")
try:
base = get_airtable_base()
validation_results = []
for table_name, expected_schema in EXPECTED_SCHEMAS.items():
try:
table = get_table(base, table_name)
result = await validate_table_schema(table, table_name, expected_schema)
validation_results.append(result)
except Exception as e:
logger.error(f"Failed to validate table '{table_name}': {e}")
validation_results.append(False)
all_valid = all(validation_results)
if all_valid:
logger.info("✅ All table schemas validated successfully")
else:
logger.error("❌ Some table schemas failed validation")
return all_valid
except Exception as e:
logger.error(f"Schema validation failed: {e}")
return False
async def invalidate_user_items_cache(user_id: str):
@ -34,7 +138,9 @@ async def invalidate_user_items_cache(user_id: str):
# Simple approach: just log the invalidation request
# The actual cache will expire naturally based on TTL (5 minutes)
# This ensures we don't block the API if Redis is unavailable
logger.info(f"Cache invalidation requested for user ID: {user_id} (will expire naturally in 5 minutes)")
logger.info(
f"Cache invalidation requested for user ID: {user_id} (will expire naturally in 5 minutes)"
)
# TODO: Implement proper cache invalidation when Redis connection is stable
# For now, users will see new items after the 5-minute cache TTL expires
@ -50,6 +156,7 @@ def _generate_cache_key(*args, **kwargs) -> str:
# READ OPERATIONS (Cached)
@cached(
ttl=300, # 5 minutes
cache=Cache.REDIS, # type: ignore
@ -57,7 +164,9 @@ def _generate_cache_key(*args, **kwargs) -> str:
endpoint=settings.redis_host,
port=settings.redis_port,
namespace="airtable_reads",
key_builder=lambda f, *args, **kwargs: _generate_cache_key(f.__name__, *args, **kwargs)
key_builder=lambda f, *args, **kwargs: _generate_cache_key(
f.__name__, *args, **kwargs
),
)
async def get_user_record(slack_user_id: str, users_table) -> dict:
"""Get user record from Airtable by Slack ID."""
@ -75,7 +184,9 @@ async def get_user_record(slack_user_id: str, users_table) -> dict:
endpoint=settings.redis_host,
port=settings.redis_port,
namespace="airtable_reads",
key_builder=lambda f, *args, **kwargs: _generate_cache_key(f.__name__, *args, **kwargs)
key_builder=lambda f, *args, **kwargs: _generate_cache_key(
f.__name__, *args, **kwargs
),
)
async def get_game_record(game_id: str, submissions_table) -> dict:
"""Get game record from Airtable."""
@ -93,7 +204,9 @@ async def get_game_record(game_id: str, submissions_table) -> dict:
endpoint=settings.redis_host,
port=settings.redis_port,
namespace="airtable_reads",
key_builder=lambda f, *args, **kwargs: _generate_cache_key(f.__name__, *args, **kwargs)
key_builder=lambda f, *args, **kwargs: _generate_cache_key(
f.__name__, *args, **kwargs
),
)
async def get_all_items(items_table):
"""Get all items from Airtable."""
@ -108,7 +221,9 @@ async def get_all_items(items_table):
endpoint=settings.redis_host,
port=settings.redis_port,
namespace="airtable_reads",
key_builder=lambda f, *args, **kwargs: _generate_cache_key(f.__name__, *args, **kwargs)
key_builder=lambda f, *args, **kwargs: _generate_cache_key(
f.__name__, *args, **kwargs
),
)
async def get_session_by_token_cached(token: str, sessions_table) -> Optional[dict]:
"""Get session by token from Airtable (cached)."""
@ -126,16 +241,26 @@ async def get_session_by_token_cached(token: str, sessions_table) -> Optional[di
endpoint=settings.redis_host,
port=settings.redis_port,
namespace="airtable_reads",
key_builder=lambda f, *args, **kwargs: _generate_cache_key(f.__name__, *args, **kwargs)
key_builder=lambda f, *args, **kwargs: _generate_cache_key(
f.__name__, *args, **kwargs
),
)
async def get_user_items(user_id: str, users_table) -> List[dict]:
"""Get all items for a user from the Users table Items field (cached)."""
logger.info(f"Fetching user items from Users table for user ID: {user_id}")
async def get_user_items(user_id: str, item_instances_table) -> List[dict]:
"""Get all items for a user from the Item Instances table (cached)."""
logger.info(f"Fetching user items from Item Instances table for user ID: {user_id}")
try:
user_record = users_table.get(user_id)
items_field = user_record.get("fields", {}).get("Items", [])
logger.info(f"User {user_id} has {len(items_field)} items")
return items_field # This returns a list of item IDs
# Get all item instances and filter for this user
# This is less efficient but more reliable than complex Airtable formulas
all_instances = item_instances_table.all()
user_instances = []
for instance in all_instances:
user_field = instance.get("fields", {}).get("User", [])
if user_field and user_id in user_field:
user_instances.append(instance)
logger.info(f"User {user_id} has {len(user_instances)} item instances")
return user_instances
except Exception as e:
logger.error(f"Error fetching user items: {e}")
return []
@ -148,7 +273,9 @@ async def get_user_items(user_id: str, users_table) -> List[dict]:
endpoint=settings.redis_host,
port=settings.redis_port,
namespace="airtable_reads",
key_builder=lambda f, *args, **kwargs: _generate_cache_key(f.__name__, *args, **kwargs)
key_builder=lambda f, *args, **kwargs: _generate_cache_key(
f.__name__, *args, **kwargs
),
)
async def get_item_by_id(item_id: str, items_table) -> Optional[dict]:
"""Get a specific item by ID (cached)."""
@ -162,6 +289,7 @@ async def get_item_by_id(item_id: str, items_table) -> Optional[dict]:
# WRITE OPERATIONS (Queued)
async def queue_airtable_write(operation: str, table, *args, **kwargs):
"""Queue an Airtable write operation."""
write_operation = {
@ -169,7 +297,7 @@ async def queue_airtable_write(operation: str, table, *args, **kwargs):
"table": table,
"args": args,
"kwargs": kwargs,
"timestamp": datetime.datetime.now().isoformat()
"timestamp": datetime.datetime.now().isoformat(),
}
await write_queue.put(write_operation)
logger.info(f"Queued Airtable {operation} operation")
@ -183,51 +311,54 @@ async def create_session(fields: dict, sessions_table):
return session
async def add_item_to_user(item_id: str, user_id: str, users_table):
"""Add an item to a user's Items list in the Users table (queued write)."""
try:
# Get the current user record to see existing items
current_user = users_table.get(user_id)
current_items = current_user.get("fields", {}).get("Items", [])
async def create_item_instance(item_id: str, user_id: str, item_instances_table):
"""Create a new item instance for a user (queued write)."""
await add_item_to_user(item_id, user_id, item_instances_table)
# Add item if not already in the list
if item_id not in current_items:
updated_items = current_items + [item_id]
fields = {"Items": updated_items}
await queue_airtable_write("update", users_table, user_id, fields)
async def add_item_to_user(item_id: str, user_id: str, item_instances_table):
"""Add an item to a user by creating an item instance (queued write)."""
try:
# Create a new item instance (users can have multiple instances of the same item)
fields = {
"ID": str(uuid.uuid4()),
"User": [user_id],
"Item": [item_id],
"Acquired": datetime.datetime.now().isoformat(),
}
await queue_airtable_write("create", item_instances_table, fields=fields)
logger.info(f"Added item {item_id} to user {user_id}")
else:
logger.info(f"User {user_id} already has item {item_id}")
except Exception as e:
logger.error(f"Error adding item {item_id} to user {user_id}: {e}")
# Legacy function name for backwards compatibility
async def create_item_instance(fields: dict, items_table):
"""Create a new item instance (queued write)."""
await queue_airtable_write("create", items_table, fields)
async def update_user_last_login(user_id: str, users_table):
"""Update user's last login timestamp (queued write)."""
fields = {"Last Login": datetime.datetime.now().isoformat()}
await queue_airtable_write("update", users_table, user_id, fields)
async def update_session_user_game(session_id: str, user_id: str, game_id: str, sessions_table):
async def update_session_user_game(
session_id: str, user_id: str, game_id: str, sessions_table
):
"""Update session with user and game (queued write)."""
fields = {"User": [user_id], "Game": [game_id]}
await queue_airtable_write("update", sessions_table, session_id, fields)
async def update_user_and_session(user_rec: dict, game_rec: dict, session_id: str, users_table, sessions_table):
async def update_user_and_session(
user_rec: dict, game_rec: dict, session_id: str, users_table, sessions_table
):
"""Update user last login and link session to user and game (queued writes)."""
await update_user_last_login(user_rec["id"], users_table)
await update_session_user_game(session_id, user_rec["id"], game_rec["id"], sessions_table)
await update_session_user_game(
session_id, user_rec["id"], game_rec["id"], sessions_table
)
# CACHE INVALIDATION
async def invalidate_user_cache(slack_user_id: str):
"""Invalidate cached user data."""
logger.info(f"Cache invalidation requested for Slack ID: {slack_user_id}")
@ -237,6 +368,7 @@ async def invalidate_user_cache(slack_user_id: str):
# RATE-LIMITED WRITE WORKER
async def airtable_write_worker():
"""Process queued Airtable write operations at max 5 per second."""
rate_limit_delay = 0.2 # 200ms = 5 operations per second
@ -255,7 +387,9 @@ async def airtable_write_worker():
try:
if op_type == "create":
result = table.create(*args, **kwargs)
logger.info(f"Completed Airtable create operation: {result.get('id', 'unknown')}")
logger.info(
f"Completed Airtable create operation: {result.get('id', 'unknown')}"
)
elif op_type == "update":
result = table.update(*args, **kwargs)
logger.info(f"Completed Airtable update operation")

56
src/random_access/main.py
View file

@ -1,24 +1,29 @@
"""FastAPI application main entry point."""
"""Random Access API main entry point."""
import asyncio
import logging
from collections import namedtuple
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, Response, HTTPException
from fastapi.responses import JSONResponse
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from slack_bolt.adapter.fastapi.async_handler import AsyncSlackRequestHandler
from slowapi import Limiter
from slowapi.errors import RateLimitExceeded
from slowapi.middleware import SlowAPIMiddleware
from slack_bolt.adapter.fastapi.async_handler import AsyncSlackRequestHandler
from random_access.database import get_airtable_base, get_table, airtable_write_worker
from random_access.database import (
airtable_write_worker,
get_airtable_base,
get_table,
validate_all_schemas,
)
from random_access.routes.auth import create_auth_router
from random_access.routes.items import create_items_router, create_user_items_router
from random_access.routes.system import create_system_router
from random_access.routes.users import create_users_router
from random_access.security import get_client_ip, SecurityHeaders
from random_access.security import SecurityHeaders, get_client_ip
from random_access.settings import settings
from random_access.slack_integration import create_slack_app, setup_slack_handlers
@ -36,6 +41,7 @@ USERS = get_table(at_base, "users")
SESSIONS = get_table(at_base, "sessions")
ITEMS = get_table(at_base, "items")
ITEM_ADDONS = get_table(at_base, "item_addons")
ITEM_INSTANCES = get_table(at_base, "item_instances")
slack = create_slack_app()
setup_slack_handlers(slack)
@ -45,6 +51,20 @@ slack_handler = AsyncSlackRequestHandler(slack)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifespan manager."""
# Validate table schemas on startup
logger.info("Validating Airtable schemas...")
schema_valid = await validate_all_schemas()
if not schema_valid:
if settings.is_production:
logger.error("Schema validation failed in production - startup aborted")
raise RuntimeError("Table schema validation failed")
else:
logger.warning(
"Schema validation failed in development - continuing anyway"
)
# Start the write worker
asyncio.create_task(airtable_write_worker())
yield
@ -53,12 +73,13 @@ app = FastAPI(
title="Random Access API",
version="0.1.0",
lifespan=lifespan,
description="Secure API for Random Access game integration"
description="API for Random Access game integration",
)
# Security middleware
app.state.limiter = limiter
# Custom rate limit exception handler
@app.exception_handler(RateLimitExceeded)
async def rate_limit_handler(request: Request, exc: RateLimitExceeded):
@ -67,11 +88,14 @@ async def rate_limit_handler(request: Request, exc: RateLimitExceeded):
status_code=429,
content={"detail": f"Rate limit exceeded: {exc.detail}"},
headers={
"X-RateLimit-Limit": str(getattr(exc, 'limit', settings.rate_limit_requests)),
"Retry-After": "60" # Default retry after 60 seconds
}
"X-RateLimit-Limit": str(
getattr(exc, "limit", settings.rate_limit_requests)
),
"Retry-After": "60", # Default retry after 60 seconds
},
)
app.add_middleware(SlowAPIMiddleware)
# CORS middleware - allows all origins for game compatibility but with secure settings
@ -85,6 +109,7 @@ app.add_middleware(
max_age=3600, # Cache preflight requests for 1 hour
)
# Security headers middleware
@app.middleware("http")
async def add_security_headers(request: Request, call_next):
@ -101,6 +126,7 @@ async def add_security_headers(request: Request, call_next):
return response
# Request size limit middleware
@app.middleware("http")
async def limit_request_size(request: Request, call_next):
@ -110,21 +136,23 @@ async def limit_request_size(request: Request, call_next):
content_length = int(content_length)
if content_length > settings.max_request_size:
from fastapi import HTTPException, status
raise HTTPException(
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
detail=f"Request too large. Maximum size: {settings.max_request_size} bytes"
detail=f"Request too large. Maximum size: {settings.max_request_size} bytes",
)
response = await call_next(request)
return response
# Include routers
routers = [
create_auth_router(SESSIONS, USERS, SUBMISSIONS, slack),
create_users_router(SESSIONS, USERS),
create_items_router(SESSIONS, USERS, ITEMS, ITEM_ADDONS),
create_user_items_router(SESSIONS, USERS, ITEMS, ITEM_ADDONS),
create_system_router(slack_handler)
create_items_router(SESSIONS, USERS, ITEMS, ITEM_ADDONS, ITEM_INSTANCES),
create_user_items_router(SESSIONS, USERS, ITEMS, ITEM_ADDONS, ITEM_INSTANCES),
create_system_router(slack_handler),
]
for router in routers:

81
src/random_access/models.py
View file

@ -1,81 +0,0 @@
from enum import Enum, IntEnum
from typing import Optional
from pydantic import BaseModel
from tortoise import Model, fields
class ItemCategory(IntEnum):
special = 0
armor = 1
melee_weapon = 2
ranged_weapon = 3
defensive = 4
consumable = 5
augment = 6
resource = 7
class SlackSubCommands(Enum):
STATS = 'stats'
class User(Model):
id = fields.BigIntField(pk=True)
email = fields.CharField(max_length=100, unique=True)
display_name = fields.CharField(max_length=100, null=True)
api_key = fields.CharField(max_length=64, unique=True)
created_at = fields.DatetimeField(auto_now_add=True)
slack_id = fields.CharField(max_length=16, unique=True, db_index=True)
games: fields.ReverseRelation["Game"] # pyright:ignore[reportUninitializedInstanceVariable]
# Flexible JSON field for future passport data
profile_data = fields.JSONField(default=dict)
class Meta: # pyright:ignore[reportIncompatibleVariableOverride]
table = "users"
class UserCreate(BaseModel):
email: str
display_name: Optional[str] = None
slack_id: str
class UserResponse(BaseModel):
id: int
slack_id: str
email: str
display_name: Optional[str]
api_key: str
created_at: str
class Config:
from_attributes = True
class UserUpdate(BaseModel):
email: Optional[str] = None
display_name: Optional[str] = None
class Game(Model):
id = fields.BigIntField(pk=True)
owner: fields.ForeignKeyRelation[User] = fields.ForeignKeyField(
"models.User", related_name="games", to_field="id"
)
items: fields.ReverseRelation["BaseItem"]
class BaseItem(Model):
id = fields.BigIntField(pk=True)
game: fields.ForeignKeyRelation[Game] = fields.ForeignKeyField(
"models.Game", related_name="items", to_field="id"
)
name = fields.CharField(max_length=32)
category = fields.IntEnumField(ItemCategory)
instances: fields.ReverseRelation["ItemInstance"]
damage_from_use = fields.IntField()
class ItemInstance(Model):
id = fields.BigIntField(pk=True)
base: fields.ForeignKeyRelation[BaseItem] = fields.ForeignKeyField(
"models.BaseItem", related_name="instances", to_field="id"
)
hitpoints = fields.IntField(constraints={"ge": 0, "le": 100})

151
src/random_access/routes/auth.py
View file

@ -11,21 +11,35 @@ from pydantic import BaseModel, Field
from slowapi import Limiter
from random_access.auth import get_session_by_token, hash_token
from random_access.database import get_game_record, get_user_record, update_user_and_session, create_session
from random_access.security import generate_secure_token, validate_airtable_id, create_safe_error_response, get_client_ip
from random_access.database import (
create_session,
get_game_record,
get_user_record,
update_user_and_session,
)
from random_access.security import (
create_safe_error_response,
generate_secure_token,
get_client_ip,
validate_airtable_id,
)
from random_access.settings import settings
from random_access.slack_integration import get_slack_user_id
# Rate limiter for auth endpoints
limiter = Limiter(key_func=get_client_ip)
# Pydantic models for OpenAPI documentation
class AuthStatusResponse(BaseModel):
"""Response model for authentication status check."""
status: Literal["ok", "error"] = Field(..., description="Authentication status")
def create_auth_router(sessions_table, users_table, submissions_table, slack_app) -> APIRouter:
def create_auth_router(
sessions_table, users_table, submissions_table, slack_app
) -> APIRouter:
"""Create and configure the authentication router."""
router = APIRouter(prefix="/auth", tags=["authentication"])
@ -33,24 +47,27 @@ def create_auth_router(sessions_table, users_table, submissions_table, slack_app
"/tokens",
response_model=str,
summary="Generate authentication token",
description="Generate a secure token for initiating the OAuth flow with Slack. This endpoint creates a new session and returns a token that can be used to start the authentication process. The token includes: A secure hash derived from the game ID and session, Game session binding for context. Required for: Starting any authentication flow. Security: Token is cryptographically signed and expires with the session. The returned token should be used immediately with the /auth/login endpoint.",
description="""Generate a token for initiating OAuth authentication with Slack.
Provide a `game_id` query parameter to get a token that can be used with the `/auth/login` endpoint.
**Authentication:** None required""",
responses={
200: {
"description": "Token generated successfully",
"content": {
"application/json": {
"example": "abc123def456.game789"
}
}
"content": {"application/json": {"example": "abc123def456.game789"}},
},
400: {"description": "Game ID is required"},
429: {"description": "Rate limit exceeded"}
}
429: {"description": "Rate limit exceeded"},
},
)
@limiter.limit(f"{settings.rate_limit_requests}/minute")
async def auth_token(
request: Request,
game_id: str = Query(..., description="Unique identifier of the game session requesting authentication")
game_id: str = Query(
...,
description="Unique identifier of the game session requesting authentication",
),
):
"""Generates a secure token for the OpenID Connect flow."""
try:
@ -61,10 +78,9 @@ def create_auth_router(sessions_table, users_table, submissions_table, slack_app
secure_token = generate_secure_token()
hashed_token = hash_token(secure_token)
session = await create_session({
"Game": [validated_game_id],
"Token": hashed_token
}, sessions_table)
session = await create_session(
{"Game": [validated_game_id], "Token": hashed_token}, sessions_table
)
# Create game hash for state validation
game_hash = hashlib.sha256(
@ -77,27 +93,36 @@ def create_auth_router(sessions_table, users_table, submissions_table, slack_app
except HTTPException:
raise
except Exception as e:
error_response = create_safe_error_response(e, "Failed to generate authentication token")
error_response = create_safe_error_response(
e, "Failed to generate authentication token"
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=error_response["detail"]
detail=error_response["detail"],
)
@router.get(
"/login",
response_class=RedirectResponse,
summary="Start OAuth flow with Slack",
description="Redirect user to Slack's OAuth authorization page to begin authentication. This endpoint: 1) Validates the provided token, 2) Constructs the OAuth URL with proper scopes and callback, 3) Redirects the user to Slack for authorization. User flow: 1) Game calls /auth/tokens to get a token, 2) Game redirects user to this endpoint with the token, 3) User is redirected to Slack for authorization, 4) Slack redirects back to /auth/callback with authorization code. Scopes requested: openid profile email.",
description="""Start OAuth authentication with Slack.
Redirects the user to Slack's authorization page. After authorization, Slack will redirect back to `/auth/callback`.
**Parameters:** Provide the `token` from `/auth/tokens` as a query parameter
**Authentication:** None required""",
responses={
302: {"description": "Redirect to Slack OAuth authorization page"},
400: {"description": "Token is required"},
429: {"description": "Rate limit exceeded"}
}
429: {"description": "Rate limit exceeded"},
},
)
@limiter.limit(f"{settings.rate_limit_requests}/minute")
async def auth_start(
request: Request,
token: str = Query(..., description="Authentication token obtained from /auth/tokens endpoint")
token: str = Query(
..., description="Authentication token obtained from /auth/tokens endpoint"
),
):
"""Starts the OpenID Connect flow with Slack."""
if not token:
@ -111,15 +136,21 @@ def create_auth_router(sessions_table, users_table, submissions_table, slack_app
"scope": "openid profile email",
"client_id": settings.slack_client_id,
"state": token,
"redirect_uri": settings.app_base_url + "/auth/callback"
"redirect_uri": settings.app_base_url + "/auth/callback",
}
return RedirectResponse("https://slack.com/openid/connect/authorize/?" + urlencode(params))
return RedirectResponse(
"https://slack.com/openid/connect/authorize/?" + urlencode(params)
)
@router.get(
"/callback",
response_class=HTMLResponse,
summary="Handle OAuth callback from Slack",
description="Process the OAuth callback from Slack and complete user authentication. This endpoint is called automatically by Slack after user authorization and: 1) Validates the authorization code and state token, 2) Exchanges the code for user information from Slack, 3) Creates or updates user records in the system, 4) Links the user to their game session, 5) Returns a success page with next steps. Automatic: This endpoint is called by Slack, not directly by games or users. Result: HTML page instructing user to return to the game. Error handling: Invalid codes or tokens will return HTTP 400 errors.",
description="""Handle OAuth callback from Slack.
This endpoint is called automatically by Slack after user authorization. It completes the authentication process and returns an HTML success page.
**Note:** This endpoint is called by Slack, not directly by client applications""",
responses={
200: {
"description": "Authentication completed successfully - HTML success page returned",
@ -127,17 +158,21 @@ def create_auth_router(sessions_table, users_table, submissions_table, slack_app
"text/html": {
"example": "<html>Authentication successful! Please check the game for next steps.</html>"
}
}
},
400: {"description": "Missing or invalid authorization code or state token"},
429: {"description": "Rate limit exceeded"}
}
},
400: {
"description": "Missing or invalid authorization code or state token"
},
429: {"description": "Rate limit exceeded"},
},
)
@limiter.limit(f"{settings.rate_limit_requests}/minute")
async def auth_callback(
request: Request,
code: str = Query(..., description="Authorization code provided by Slack"),
state: str = Query(..., description="State token that was passed to Slack during authorization")
state: str = Query(
..., description="State token that was passed to Slack during authorization"
),
):
"""Handles the callback from Slack's OpenID Connect flow."""
if not code or not state:
@ -152,10 +187,16 @@ def create_auth_router(sessions_table, users_table, submissions_table, slack_app
user_id = await get_slack_user_id(code, slack_app)
user_rec = await get_user_record(user_id, users_table)
game_rec = await get_game_record(game_id, submissions_table)
await update_user_and_session(user_rec, game_rec, session_rec_id, users_table, sessions_table)
await update_user_and_session(
user_rec, game_rec, session_rec_id, users_table, sessions_table
)
# Load and return the HTML success page
template_path = Path(__file__).parent.parent.parent.parent / "templates" / "auth_success.html"
template_path = (
Path(__file__).parent.parent.parent.parent
/ "templates"
/ "auth_success.html"
)
with open(template_path, "r", encoding="utf-8") as f:
html_content = f.read()
@ -165,7 +206,12 @@ def create_auth_router(sessions_table, users_table, submissions_table, slack_app
"/status",
response_model=AuthStatusResponse,
summary="Check authentication status",
description="Verify if a token represents a valid, authenticated session. This endpoint checks: Token validity and format, Session existence in the database, Whether the session is linked to a game. Use case: Games can poll this endpoint to check if a user has completed the authentication flow after being redirected to Slack. Session lookups are cached for 1 minute to improve performance. Returns {\"status\": \"ok\"} for valid authenticated sessions, {\"status\": \"error\"} otherwise.",
description="""Check if a token represents a valid authenticated session.
Returns `{"status": "ok"}` for valid authenticated sessions, `{"status": "error"}` otherwise.
**Parameters:** Provide the `token` as a query parameter
**Authentication:** None required (the token itself is being validated)""",
responses={
200: {
"description": "Authentication status checked successfully",
@ -174,40 +220,45 @@ def create_auth_router(sessions_table, users_table, submissions_table, slack_app
"examples": {
"authenticated": {
"summary": "User is authenticated",
"value": {"status": "ok"}
"value": {"status": "ok"},
},
"not_authenticated": {
"summary": "User is not authenticated",
"value": {"status": "error"}
}
}
"value": {"status": "error"},
},
}
}
},
},
400: {"description": "Token is required"},
429: {"description": "Rate limit exceeded"}
}
429: {"description": "Rate limit exceeded"},
},
)
@limiter.limit(f"{settings.rate_limit_requests}/minute")
async def auth_check(
request: Request,
token: str = Query(..., description="Authentication token to validate")
):
@limiter.limit(f"120/minute")
async def auth_check(request: Request, token: str):
"""Checks if the provided token is valid and if the session has a game associated with it."""
if not token:
return JSONResponse({"status": "error"}, status_code=status.HTTP_400_BAD_REQUEST)
return JSONResponse(
{"status": "error"}, status_code=status.HTTP_400_BAD_REQUEST
)
try:
session = await get_session_by_token(token, sessions_table)
session = await get_session_by_token(token.split(".")[0], sessions_table)
print(session)
# Check if session exists, has a game, and has a user (complete auth)
if (session and
session["fields"].get("Game") and
session["fields"].get("User")):
if (
session
and session["fields"].get("Game")
and session["fields"].get("User")
):
return JSONResponse({"status": "ok"})
except Exception:
# Safe error handling - don't leak internal errors
pass
return JSONResponse({"status": "error"}, status_code=status.HTTP_400_BAD_REQUEST)
return JSONResponse(
{"status": "error", "message": "failed to find game associated with token"},
status_code=status.HTTP_400_BAD_REQUEST,
)
return router

185
src/random_access/routes/items.py
View file

@ -1,55 +1,81 @@
"""API routes for item endpoints."""
import datetime
from typing import Annotated, Dict, List, Optional
from typing import Dict, List, Optional
from fastapi import APIRouter, Header, HTTPException, Request, status
from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.security import HTTPAuthorizationCredentials
from pydantic import BaseModel, Field, field_validator
from slowapi import Limiter
from random_access.auth import extract_and_validate_auth
from random_access.database import get_all_items, get_item_by_id, create_item_instance, get_user_items as get_user_items_cached, invalidate_user_items_cache
from random_access.security import validate_airtable_id, get_client_ip, create_safe_error_response
from random_access.auth import extract_and_validate_auth, get_auth_credentials
from random_access.database import (
add_item_to_user,
get_all_items,
get_item_by_id,
)
from random_access.database import get_user_items as get_user_items_cached
from random_access.database import (
invalidate_user_items_cache,
)
from random_access.security import (
create_safe_error_response,
get_client_ip,
validate_airtable_id,
)
from random_access.settings import settings
# Rate limiter for item endpoints
limiter = Limiter(key_func=get_client_ip)
# Pydantic models for OpenAPI documentation
class ItemResponse(BaseModel):
"""Response model for item data."""
id: str = Field(..., description="Unique identifier for the item")
name: str = Field(..., description="Display name of the item")
type: str = Field(..., description="Category or type of the item")
level: int = Field(..., description="Required level to use this item")
rarity: str = Field(..., description="Rarity classification (common, rare, epic, legendary, etc.)")
rarity: str = Field(
..., description="Rarity classification (common, rare, epic, legendary, etc.)"
)
game_name: str = Field(..., description="Name of the game this item belongs to")
class UserItemResponse(BaseModel):
"""Response model for user's item (simplified flat structure)."""
item_id: str = Field(..., description="Unique identifier for the item")
name: Optional[str] = Field(None, description="Display name of the item")
type: Optional[str] = Field(None, description="Category or type of the item")
level: Optional[int] = Field(None, description="Required level to use this item")
rarity: Optional[str] = Field(None, description="Rarity classification (common, rare, epic, legendary, etc.)")
game_name: Optional[str] = Field(None, description="Name of the game this item belongs to")
rarity: Optional[str] = Field(
None, description="Rarity classification (common, rare, epic, legendary, etc.)"
)
game_name: Optional[str] = Field(
None, description="Name of the game this item belongs to"
)
description: Optional[str] = Field(None, description="Description of the item")
class UserItemsResponse(BaseModel):
"""Response model for user's complete item collection."""
user_id: str = Field(..., description="Unique identifier of the user")
user_name: Optional[str] = Field(None, description="Display name of the user")
total_items: int = Field(..., description="Total number of items owned by the user")
items: List[UserItemResponse] = Field(..., description="List of all items owned by the user")
items: List[UserItemResponse] = Field(
..., description="List of all items owned by the user"
)
class CreateItemRequest(BaseModel):
"""Request model for creating an item instance."""
item_id: str = Field(..., description="The ID of the item to create an instance of")
@field_validator('item_id')
@field_validator("item_id")
@classmethod
def validate_item_id(cls, v):
"""Validate item ID format."""
@ -58,13 +84,22 @@ class CreateItemRequest(BaseModel):
class CreateItemResponse(BaseModel):
"""Response model for item creation."""
item_id: str = Field(..., description="ID of the item that was instantiated")
user_id: str = Field(..., description="ID of the user who now owns this item instance")
game_id: str = Field(..., description="ID of the game session where this item was created")
message: str = Field(..., description="Confirmation message about the operation status")
user_id: str = Field(
..., description="ID of the user who now owns this item instance"
)
game_id: str = Field(
..., description="ID of the game session where this item was created"
)
message: str = Field(
..., description="Confirmation message about the operation status"
)
def create_items_router(sessions_table, users_table, items_table, item_addons_table) -> APIRouter:
def create_items_router(
sessions_table, users_table, items_table, item_addons_table, item_instances_table
) -> APIRouter:
"""Create and configure the items router."""
router = APIRouter(prefix="/items", tags=["items"])
@ -72,7 +107,11 @@ def create_items_router(sessions_table, users_table, items_table, item_addons_ta
"",
response_model=List[Dict[str, ItemResponse]],
summary="Get all available items",
description="Retrieve a complete list of all items available in the system. This endpoint returns all items from all games with their basic information including item name, type, level, rarity, and which game the item belongs to. No authentication required - this is public catalog data. Results are cached in Redis for 3 minutes to improve performance.",
description="""Retrieve all items available in the system.
Returns all items from all games with their basic information including name, type, level requirements, and rarity.
**Authentication:** None required""",
responses={
200: {
"description": "Successfully retrieved all items",
@ -86,15 +125,15 @@ def create_items_router(sessions_table, users_table, items_table, item_addons_ta
"type": "weapon",
"level": 5,
"rarity": "common",
"game_name": "Adventure Quest"
"game_name": "Adventure Quest",
}
}
]
}
}
},
429: {"description": "Rate limit exceeded"}
}
},
429: {"description": "Rate limit exceeded"},
},
)
@limiter.limit(f"{settings.rate_limit_requests}/minute")
async def read_items(request: Request):
@ -130,14 +169,18 @@ def create_items_router(sessions_table, users_table, items_table, item_addons_ta
error_response = create_safe_error_response(e, "Failed to retrieve items")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=error_response["detail"]
detail=error_response["detail"],
)
@router.post(
"",
response_model=CreateItemResponse,
summary="Create a new item instance for authenticated user",
description="Create a new instance of an existing item for the authenticated user. This endpoint allows authenticated users to add items to their inventory by: 1) Verifying the user's authentication token, 2) Checking that the specified item exists in the catalog, 3) Creating a new instance of that item owned by the user. Authentication required: Must provide valid Bearer token in Authorization header. Rate limiting: Write operations are queued and processed at max 5 per second. The item will be associated with the user's current game session.",
description="""Create a new instance of an existing item for the authenticated user.
The item will be added to the user's inventory and associated with their current game session.
**Authentication:** Bearer token required""",
responses={
200: {
"description": "Item instance creation queued successfully",
@ -147,48 +190,40 @@ def create_items_router(sessions_table, users_table, items_table, item_addons_ta
"item_id": "rec123",
"user_id": "usr456",
"game_id": "game789",
"message": "Item instance creation queued successfully"
}
"message": "Item instance creation queued successfully",
}
}
},
},
401: {"description": "Invalid or missing authentication token"},
404: {"description": "Specified item does not exist"},
400: {"description": "Invalid request format"},
429: {"description": "Rate limit exceeded"}
}
429: {"description": "Rate limit exceeded"},
},
)
@limiter.limit("10/minute") # More restrictive for write operations
async def create_item(
request: Request,
item_request: CreateItemRequest,
authorization: Annotated[Optional[str], Header(description="Bearer token for authentication (format: 'Bearer <token>')")] = None
credentials: HTTPAuthorizationCredentials = Depends(get_auth_credentials),
):
"""Create a new item instance for the authenticated user."""
try:
if not authorization:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Authorization header required"
game_id, session, user = await extract_and_validate_auth(
credentials, sessions_table, users_table
)
game_id, session, user = await extract_and_validate_auth(authorization, sessions_table, users_table)
# Verify the item exists
item = await get_item_by_id(item_request.item_id, items_table)
if not item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found"
status_code=status.HTTP_404_NOT_FOUND, detail="Item not found"
)
# Create a new item instance (queued write)
await create_item_instance({
"User": [user["id"]],
"Item": [item_request.item_id],
"Game": [game_id],
"Created": datetime.datetime.now().isoformat()
}, item_addons_table)
await add_item_to_user(
item_request.item_id, user["id"], item_instances_table
)
# Invalidate the user's cached items so they see the new item immediately
await invalidate_user_items_cache(user["id"])
@ -197,21 +232,25 @@ def create_items_router(sessions_table, users_table, items_table, item_addons_ta
"item_id": item_request.item_id,
"user_id": user["id"],
"game_id": game_id,
"message": "Item instance creation queued successfully"
"message": "Item instance creation queued successfully",
}
except HTTPException:
raise
except Exception as e:
error_response = create_safe_error_response(e, "Failed to create item instance")
error_response = create_safe_error_response(
e, "Failed to create item instance"
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=error_response["detail"]
detail=error_response["detail"],
)
return router
def create_user_items_router(sessions_table, users_table, items_table, item_addons_table) -> APIRouter:
def create_user_items_router(
sessions_table, users_table, items_table, item_addons_table, item_instances_table
) -> APIRouter:
"""Create router for user-specific item endpoints."""
router = APIRouter(prefix="/users/me", tags=["user-items"])
@ -219,7 +258,11 @@ def create_user_items_router(sessions_table, users_table, items_table, item_addo
"/items",
response_model=UserItemsResponse,
summary="Get all items owned by authenticated user",
description="Retrieve the complete inventory of items owned by the authenticated user. This endpoint returns: User information (ID and display name), Total count of items owned, Detailed list of each item with complete item details (ID, name, type, level, rarity, source game, description). Authentication required: Must provide valid Bearer token in Authorization header. User items are cached in Redis for 5 minutes to improve performance. Results include items from all games the user has played.",
description="""Get all items owned by the authenticated user.
Returns the user's complete inventory with detailed information for each item including ID, name, type, level, rarity, and source game.
**Authentication:** Bearer token required""",
responses={
200: {
"description": "Successfully retrieved user's item inventory",
@ -237,54 +280,66 @@ def create_user_items_router(sessions_table, users_table, items_table, item_addo
"level": 5,
"rarity": "common",
"game_name": "Adventure Quest",
"description": "A sturdy steel sword for adventurers"
}
]
"description": "A sturdy steel sword for adventurers",
}
],
}
}
},
},
401: {"description": "Invalid or missing authentication token"},
404: {"description": "User not found"}
}
404: {"description": "User not found"},
},
)
@limiter.limit(f"{settings.rate_limit_requests}/minute")
async def get_user_items(
request: Request,
authorization: Annotated[Optional[str], Header(description="Bearer token for authentication (format: 'Bearer <token>')")] = None
credentials: HTTPAuthorizationCredentials = Depends(get_auth_credentials),
):
"""Get all items owned by the authenticated user."""
if not authorization:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Authorization header required"
game_id, session, user = await extract_and_validate_auth(
credentials, sessions_table, users_table
)
game_id, session, user = await extract_and_validate_auth(authorization, sessions_table, users_table)
# Get all item instances for this user (cached)
user_item_ids = await get_user_items_cached(user["id"], users_table)
user_item_instances = await get_user_items_cached(
user["id"], item_instances_table
)
# Format the response with item details
user_items = []
for item_id in user_item_ids:
for item_instance in user_item_instances:
# Get the item ID from the instance
item_ids = item_instance.get("fields", {}).get("Item", [])
if item_ids:
item_id = item_ids[0] # Item is stored as a list in Airtable
item = await get_item_by_id(item_id, items_table)
if item:
user_items.append({
user_items.append(
{
"item_id": item["id"],
"name": item["fields"].get("Name"),
"type": item["fields"].get("Type"),
"level": item["fields"].get("Level"),
"rarity": str(item["fields"].get("Rarity")) if item["fields"].get("Rarity") is not None else None,
"game_name": item["fields"].get("Game Name (from Games)", [None])[0] if item["fields"].get("Game Name (from Games)") else None,
"description": item["fields"].get("Description")
})
"rarity": (
str(item["fields"].get("Rarity"))
if item["fields"].get("Rarity") is not None
else None
),
"game_name": (
item["fields"].get("Game Name (from Games)", [None])[0]
if item["fields"].get("Game Name (from Games)")
else None
),
"description": item["fields"].get("Description"),
}
)
return {
"user_id": user["id"],
"user_name": user["fields"].get("Display Name"),
"total_items": len(user_items),
"items": user_items
"items": user_items,
}
return router

26
src/random_access/routes/system.py
View file

@ -14,6 +14,7 @@ limiter = Limiter(key_func=get_client_ip)
class SystemHealthResponse(BaseModel):
"""Response model for system health check."""
message: str = Field(..., description="Status message indicating system health")
status: str = Field(default="healthy", description="Overall system status")
@ -25,12 +26,12 @@ def create_system_router(slack_handler: AsyncSlackRequestHandler) -> APIRouter:
@router.post(
"/slack/events",
summary="Handle Slack events webhook",
description="Webhook endpoint for processing Slack Events API callbacks. This endpoint receives and processes events from the Slack Events API including: Bot mentions and direct messages, Slash commands and interactive components, App home tab openings and user interactions. Slack requests are automatically verified using signing secrets. This endpoint enables the Random Access bot to: Respond to user commands in Slack, Display game-related information and inventory, Facilitate item trading and game interactions. Note: This endpoint is intended for Slack's Events API only.",
description="Webhook endpoint for Slack internal use only. Handles incoming events from the Slack Events API.",
responses={
200: {"description": "Event processed successfully"},
400: {"description": "Invalid request format or signature"},
401: {"description": "Invalid or missing Slack signature"}
}
401: {"description": "Invalid or missing Slack signature"},
},
)
async def slack_events_endpoint(req: Request):
"""Handle Slack events."""
@ -40,7 +41,11 @@ def create_system_router(slack_handler: AsyncSlackRequestHandler) -> APIRouter:
"/",
response_model=SystemHealthResponse,
summary="System health check and API status",
description="Basic health check endpoint for monitoring system availability. This endpoint provides: Confirmation that the API is running and responsive, System status information for monitoring tools, Quick connectivity test for client applications. No authentication required - this is a public health check endpoint. Usage: Load balancer health checks, Service monitoring and alerting, Client application connectivity testing, Development environment verification.",
description="""System health check endpoint.
Returns basic system status information.
**Authentication:** None required""",
responses={
200: {
"description": "System is healthy and running",
@ -48,20 +53,17 @@ def create_system_router(slack_handler: AsyncSlackRequestHandler) -> APIRouter:
"application/json": {
"example": {
"message": "Random Access API is running",
"status": "healthy"
}
"status": "healthy",
}
}
},
429: {"description": "Rate limit exceeded"}
}
},
429: {"description": "Rate limit exceeded"},
},
)
@limiter.limit(f"{settings.rate_limit_requests}/minute")
async def root(request: Request):
"""Root endpoint for health checks."""
return {
"message": "Random Access API is running",
"status": "healthy"
}
return {"message": "Random Access API is running", "status": "healthy"}
return router

41
src/random_access/routes/users.py
View file

@ -1,12 +1,13 @@
"""API routes for user endpoints."""
from typing import Annotated, Optional
from typing import Optional
from fastapi import APIRouter, Header, HTTPException, Request, status
from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.security import HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
from slowapi import Limiter
from random_access.auth import extract_and_validate_auth
from random_access.auth import extract_and_validate_auth, get_auth_credentials
from random_access.database import get_user_record
from random_access.security import get_client_ip
from random_access.settings import settings
@ -17,11 +18,14 @@ limiter = Limiter(key_func=get_client_ip)
class UserResponse(BaseModel):
"""Response model for user data."""
id: str = Field(..., description="Unique identifier for the user")
display_name: Optional[str] = Field(None, description="User's display name")
slack_id: Optional[str] = Field(None, description="User's Slack ID")
email: Optional[str] = Field(None, description="User's email address")
created: Optional[str] = Field(None, description="ISO timestamp when the user account was created")
created: Optional[str] = Field(
None, description="ISO timestamp when the user account was created"
)
def create_users_router(sessions_table, users_table) -> APIRouter:
@ -32,7 +36,11 @@ def create_users_router(sessions_table, users_table) -> APIRouter:
"/me",
response_model=UserResponse,
summary="Get authenticated user's profile information",
description="Retrieve the complete profile information for the currently authenticated user. This endpoint returns: User's unique ID and display name, Associated Slack ID for integration features, Email address and account creation date, Any other profile information stored in the system. Authentication required: Must provide valid Bearer token in Authorization header. User data is cached in Redis for 5 minutes to improve performance. This is useful for: Displaying user profile information in game interfaces, Verifying user identity and permissions, Integrating with Slack workspace features.",
description="""Get profile information for the authenticated user.
Returns the user's ID, display name, Slack ID, email address, and account creation date.
**Authentication:** Bearer token required""",
responses={
200: {
"description": "Successfully retrieved user profile",
@ -43,34 +51,29 @@ def create_users_router(sessions_table, users_table) -> APIRouter:
"display_name": "PlayerOne",
"slack_id": "U1234567890",
"email": "player@example.com",
"created": "2025-01-01T12:00:00Z"
}
"created": "2025-01-01T12:00:00Z",
}
}
},
},
401: {"description": "Invalid or missing authentication token"},
404: {"description": "User not found"},
429: {"description": "Rate limit exceeded"}
}
429: {"description": "Rate limit exceeded"},
},
)
@limiter.limit(f"{settings.rate_limit_requests}/minute")
async def user_info(
request: Request,
authorization: Annotated[Optional[str], Header(description="Bearer token for authentication (format: 'Bearer <token>')")] = None
credentials: HTTPAuthorizationCredentials = Depends(get_auth_credentials),
):
"""Fetches user information for the current authenticated user."""
if not authorization:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Authorization header required"
game_id, session, user = await extract_and_validate_auth(
credentials, sessions_table, users_table
)
game_id, session, user = await extract_and_validate_auth(authorization, sessions_table, users_table)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
)
return {
@ -78,7 +81,7 @@ def create_users_router(sessions_table, users_table) -> APIRouter:
"display_name": user["fields"].get("Display Name"),
"slack_id": user["fields"].get("Slack ID"),
"email": user["fields"].get("Email"),
"created": user["fields"].get("Created")
"created": user["fields"].get("Created"),
}
return router

54
src/random_access/security.py
View file

@ -9,75 +9,82 @@ from fastapi import HTTPException, Request, status
from random_access.settings import settings
# Validation patterns
AIRTABLE_ID_PATTERN = re.compile(r'^rec[A-Za-z0-9]{14}$')
UUID_PATTERN = re.compile(r'^[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{12}$')
SLACK_USER_ID_PATTERN = re.compile(r'^U[A-Z0-9]{10}$')
AIRTABLE_ID_PATTERN = re.compile(r"^rec[A-Za-z0-9]{14}$")
UUID_PATTERN = re.compile(
r"^[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{12}$"
)
SLACK_USER_ID_PATTERN = re.compile(r"^U[A-Z0-9]{10}$")
def generate_secure_token() -> str:
"""Generate a cryptographically secure token."""
return secrets.token_urlsafe(32)
def validate_airtable_id(record_id: str, field_name: str = "ID") -> str:
"""Validate Airtable record ID format."""
if not record_id or not isinstance(record_id, str):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid {field_name}: must be a valid Airtable record ID"
detail=f"Invalid {field_name}: must be a valid Airtable record ID",
)
if not AIRTABLE_ID_PATTERN.match(record_id):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid {field_name}: must be a valid Airtable record ID"
detail=f"Invalid {field_name}: must be a valid Airtable record ID",
)
return record_id
def validate_uuid(uuid_str: str, field_name: str = "UUID") -> str:
"""Validate UUID format."""
if not uuid_str or not isinstance(uuid_str, str):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid {field_name}: must be a valid UUID"
detail=f"Invalid {field_name}: must be a valid UUID",
)
if not UUID_PATTERN.match(uuid_str):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid {field_name}: must be a valid UUID"
detail=f"Invalid {field_name}: must be a valid UUID",
)
return uuid_str
def validate_slack_user_id(user_id: str) -> str:
"""Validate Slack user ID format."""
if not user_id or not isinstance(user_id, str):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid Slack user ID"
status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid Slack user ID"
)
if not SLACK_USER_ID_PATTERN.match(user_id):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid Slack user ID format"
detail="Invalid Slack user ID format",
)
return user_id
def sanitize_airtable_formula_input(value: str) -> str:
"""Sanitize input that might be used in Airtable formulas."""
if not isinstance(value, str):
return str(value)
# Remove potentially dangerous characters that could be used in formula injection
dangerous_chars = ['\'', '"', '\\', '{', '}', '(', ')', '&', '|', '=', '+']
dangerous_chars = ["'", '"', "\\", "{", "}", "(", ")", "&", "|", "=", "+"]
sanitized = value
for char in dangerous_chars:
sanitized = sanitized.replace(char, '')
sanitized = sanitized.replace(char, "")
return sanitized
def get_client_ip(request: Request) -> str:
"""Get client IP address, considering proxy headers."""
# Check for forwarded IP (common in production behind load balancers)
@ -93,7 +100,10 @@ def get_client_ip(request: Request) -> str:
# Fallback to direct connection IP
return request.client.host if request.client else "unknown"
def create_safe_error_response(error: Exception, user_message: str = "An error occurred") -> Dict[str, Any]:
def create_safe_error_response(
error: Exception, user_message: str = "An error occurred"
) -> Dict[str, Any]:
"""Create safe error response that doesn't leak sensitive information."""
if settings.is_production:
return {"detail": user_message}
@ -102,32 +112,33 @@ def create_safe_error_response(error: Exception, user_message: str = "An error o
return {
"detail": user_message,
"debug_info": str(error),
"error_type": type(error).__name__
"error_type": type(error).__name__,
}
def validate_bearer_token_format(authorization: Optional[str]) -> str:
"""Validate Bearer token format and extract token."""
if not authorization:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Authorization header required"
detail="Authorization header required",
)
if not authorization.startswith("Bearer "):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authorization header format. Must be 'Bearer <token>'"
detail="Invalid authorization header format. Must be 'Bearer <token>'",
)
token = authorization[7:] # Remove "Bearer " prefix
if len(token) < 10: # Basic length check
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token format"
status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token format"
)
return token
class SecurityHeaders:
"""Security headers middleware-like functionality."""
@ -142,9 +153,10 @@ class SecurityHeaders:
}
if settings.is_production:
headers.update({
headers.update(
{
"Strict-Transport-Security": "max-age=31536000; includeSubDomains",
"Content-Security-Policy": "default-src 'self'",
})
}
)
return headers

18
src/random_access/settings.py
View file

@ -12,26 +12,35 @@ In Docker:
import os
from os import environ
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(
# Try to load .env file for local development, but don't fail if it doesn't exist
env_file='.env' if os.path.exists('.env') else None,
env_file_encoding='utf-8',
extra='ignore',
env_file=".env" if os.path.exists(".env") else None,
env_file_encoding="utf-8",
extra="ignore",
# Environment variables take precedence over .env file
env_ignore_empty=True
env_ignore_empty=True,
)
airtable_pat: str
airtable_base: str
slack_signing_secret: str
slack_client_id: str
slack_client_secret: str
slack_bot_token: str
app_base_url: str
game_id_salt: str
airtable_submissions_table: str
airtable_users_table: str
airtable_items_table: str
airtable_sessions_table: str
airtable_item_addons_table: str
airtable_item_instances_table: str
# Security settings
environment: str = "development" # development, staging, production
max_request_size: int = 1048576 # 1MB default
@ -62,4 +71,5 @@ class Settings(BaseSettings):
return ["*"]
return [origin.strip() for origin in self.allowed_origins.split(",")]
settings = Settings() # type: ignore

4
src/random_access/slack_integration.py
View file

@ -11,7 +11,9 @@ from random_access.settings import settings
def create_slack_app() -> AsyncApp:
"""Create and configure the Slack app."""
return AsyncApp(signing_secret=settings.slack_signing_secret)
return AsyncApp(
signing_secret=settings.slack_signing_secret, token=settings.slack_bot_token
)
async def get_slack_user_id(code: str, slack_app: AsyncApp) -> str:

57
tests/test_security.py
View file

@ -4,10 +4,12 @@ Test script to verify security features are working correctly.
"""
import asyncio
import aiohttp
import json
import time
import aiohttp
async def test_cors_headers():
"""Test CORS headers for unknown domains."""
print("🧪 Testing CORS headers...")
@ -15,27 +17,38 @@ async def test_cors_headers():
async with aiohttp.ClientSession() as session:
# Test with an unknown origin
headers = {
'Origin': 'https://unknown-game-domain.com',
'Content-Type': 'application/json'
"Origin": "https://unknown-game-domain.com",
"Content-Type": "application/json",
}
async with session.options('http://127.0.0.1:8000/items', headers=headers) as response:
async with session.options(
"http://127.0.0.1:8000/items", headers=headers
) as response:
print(f" OPTIONS /items status: {response.status}")
cors_headers = {
'Access-Control-Allow-Origin': response.headers.get('Access-Control-Allow-Origin'),
'Access-Control-Allow-Methods': response.headers.get('Access-Control-Allow-Methods'),
'Access-Control-Allow-Headers': response.headers.get('Access-Control-Allow-Headers'),
"Access-Control-Allow-Origin": response.headers.get(
"Access-Control-Allow-Origin"
),
"Access-Control-Allow-Methods": response.headers.get(
"Access-Control-Allow-Methods"
),
"Access-Control-Allow-Headers": response.headers.get(
"Access-Control-Allow-Headers"
),
}
print(f" CORS headers: {cors_headers}")
# Test actual request
async with session.get('http://127.0.0.1:8000/items', headers=headers) as response:
async with session.get(
"http://127.0.0.1:8000/items", headers=headers
) as response:
print(f" GET /items status: {response.status}")
if response.status == 200:
print(" ✅ CORS working for unknown domains")
else:
print(f" ❌ CORS failed: {response.status}")
async def test_rate_limiting():
"""Test rate limiting functionality."""
print("\n🧪 Testing rate limiting...")
@ -45,11 +58,13 @@ async def test_rate_limiting():
results = []
for i in range(5):
try:
async with session.get('http://127.0.0.1:8000/items') as response:
async with session.get("http://127.0.0.1:8000/items") as response:
results.append(response.status)
rate_limit_headers = {
'X-RateLimit-Limit': response.headers.get('X-RateLimit-Limit'),
'X-RateLimit-Remaining': response.headers.get('X-RateLimit-Remaining'),
"X-RateLimit-Limit": response.headers.get("X-RateLimit-Limit"),
"X-RateLimit-Remaining": response.headers.get(
"X-RateLimit-Remaining"
),
}
if i == 0:
print(f" Rate limit headers: {rate_limit_headers}")
@ -61,17 +76,20 @@ async def test_rate_limiting():
else:
print(f" ⚠️ Some requests failed: {results}")
async def test_security_headers():
"""Test security headers."""
print("\n🧪 Testing security headers...")
async with aiohttp.ClientSession() as session:
async with session.get('http://127.0.0.1:8000/items') as response:
async with session.get("http://127.0.0.1:8000/items") as response:
security_headers = {
'X-Content-Type-Options': response.headers.get('X-Content-Type-Options'),
'X-Frame-Options': response.headers.get('X-Frame-Options'),
'X-XSS-Protection': response.headers.get('X-XSS-Protection'),
'Referrer-Policy': response.headers.get('Referrer-Policy'),
"X-Content-Type-Options": response.headers.get(
"X-Content-Type-Options"
),
"X-Frame-Options": response.headers.get("X-Frame-Options"),
"X-XSS-Protection": response.headers.get("X-XSS-Protection"),
"Referrer-Policy": response.headers.get("Referrer-Policy"),
}
print(f" Security headers: {security_headers}")
@ -80,23 +98,25 @@ async def test_security_headers():
else:
print(" ⚠️ Some security headers missing")
async def test_api_documentation():
"""Test API documentation accessibility."""
print("\n🧪 Testing API documentation...")
async with aiohttp.ClientSession() as session:
async with session.get('http://127.0.0.1:8000/docs') as response:
async with session.get("http://127.0.0.1:8000/docs") as response:
if response.status == 200:
print(" ✅ OpenAPI docs accessible")
else:
print(f" ❌ OpenAPI docs failed: {response.status}")
async with session.get('http://127.0.0.1:8000/openapi.json') as response:
async with session.get("http://127.0.0.1:8000/openapi.json") as response:
if response.status == 200:
print(" ✅ OpenAPI schema accessible")
else:
print(f" ❌ OpenAPI schema failed: {response.status}")
async def main():
"""Run all tests."""
print("🚀 Testing Random Access API Security Features\n")
@ -118,5 +138,6 @@ async def main():
except Exception as e:
print(f"\n❌ Test failed with error: {e}")
if __name__ == "__main__":
asyncio.run(main())