Add multi-user support - each API key gets own GeoGuessr session #5

Merged
NyxiumYuuki merged 1 commit from claude/add-mcp-authentication-01V5tbppGEtXc3tvjRGoTcfh into master 2025-11-29 23:43:07 +01:00
8 changed files with 551 additions and 93 deletions

View file

@ -12,6 +12,12 @@ A Model Context Protocol (MCP) server for analyzing GeoGuessr game statistics wi
## 🌟 Key Features ## 🌟 Key Features
### Multi-User Support
- **Independent Sessions**: Each API key gets its own GeoGuessr session
- **Multiple Accounts**: Different users can access their own GeoGuessr accounts
- **Single Server**: No need to deploy separate instances per user
- **Automatic Context**: User sessions are automatically managed per request
### Dynamic API Monitoring ### Dynamic API Monitoring
- **Automatic endpoint discovery**: Monitors GeoGuessr API endpoints daily - **Automatic endpoint discovery**: Monitors GeoGuessr API endpoints daily
- **Schema change detection**: Automatically detects when API response formats change - **Schema change detection**: Automatically detects when API response formats change
@ -101,12 +107,14 @@ Add to your Claude Desktop configuration:
## 🔐 Authentication ## 🔐 Authentication
The server supports two types of authentication: The server supports two types of authentication with **multi-user support**:
### MCP Server Authentication (Controls Access to the MCP Server) ### MCP Server Authentication (Controls Access to the MCP Server)
Secures who can connect to your MCP server. When enabled, clients must provide a valid API key. Secures who can connect to your MCP server. When enabled, clients must provide a valid API key.
**Multi-User Support:** Each API key can have its own GeoGuessr session, allowing multiple users to use the same MCP server instance with their own accounts!
**Enable in `.env`:** **Enable in `.env`:**
```bash ```bash
MCP_AUTH_ENABLED=true MCP_AUTH_ENABLED=true
@ -133,9 +141,19 @@ openssl rand -hex 32
} }
``` ```
**Multi-User Example:**
```bash
# Give each user their own API key
MCP_API_KEYS=alice_key_abc123,bob_key_def456,charlie_key_ghi789
# Alice connects with Authorization: Bearer alice_key_abc123
# Bob connects with Authorization: Bearer bob_key_def456
# Each can login to their own GeoGuessr account!
```
### GeoGuessr API Authentication (Access GeoGuessr Data) ### GeoGuessr API Authentication (Access GeoGuessr Data)
The server also needs authentication to access GeoGuessr's API. Multiple methods supported: The server also needs authentication to access GeoGuessr's API. In multi-user mode, **each API key holder can login to their own GeoGuessr account:**
### Option 1: Login via Claude (Recommended) ### Option 1: Login via Claude (Recommended)
Simply ask Claude: Simply ask Claude:
@ -150,6 +168,75 @@ GEOGUESSR_NCFA_COOKIE=your_cookie_value_here
### Option 3: Manual Cookie ### Option 3: Manual Cookie
Use the `set_ncfa_cookie` tool with a cookie extracted from your browser. Use the `set_ncfa_cookie` tool with a cookie extracted from your browser.
## 👥 Multi-User Mode
The server supports multiple users, each with their own GeoGuessr account, using a single MCP server instance.
### How It Works
1. **API Keys**: Each user gets a unique API key
2. **Independent Sessions**: Each API key has its own GeoGuessr login session
3. **Automatic Routing**: The server automatically routes requests to the correct user's session
4. **No Interference**: Users don't affect each other's sessions
### Setup Example
**1. Configure Multiple API Keys:**
```bash
# .env file
MCP_AUTH_ENABLED=true
MCP_API_KEYS=alice_key,bob_key,charlie_key
```
**2. Restart Server:**
```bash
# Development
docker compose restart
# Production
docker compose -f docker-compose.prod.yml restart
```
**3. Each User Connects:**
```json
// Alice's Claude Desktop config
{
"mcpServers": {
"geoguessr": {
"url": "https://your-domain.com/mcp",
"headers": {"Authorization": "Bearer alice_key"}
}
}
}
// Bob's Claude Desktop config
{
"mcpServers": {
"geoguessr": {
"url": "https://your-domain.com/mcp",
"headers": {"Authorization": "Bearer bob_key"}
}
}
}
```
**4. Each User Logs In:**
- Alice asks Claude: "Login to GeoGuessr with my credentials"
- Bob asks Claude: "Login to GeoGuessr with my credentials"
- Sessions are completely independent!
### Adding New Users
To add a new user to an existing deployment:
1. Edit `.env` and add the new API key to `MCP_API_KEYS`
2. Restart the server: `docker compose restart`
3. Share the new API key with the user
4. User configures their Claude Desktop with the API key
5. User logs in to their GeoGuessr account via Claude
**The server restarts in ~2-3 seconds** and all existing users remain logged in!
## 📊 Available Tools ## 📊 Available Tools
### Authentication ### Authentication

View file

@ -16,6 +16,7 @@ import httpx
from .dynamic_response import DynamicResponse from .dynamic_response import DynamicResponse
from .endpoints import EndpointInfo from .endpoints import EndpointInfo
from ..auth import get_current_user_context
from ..auth.session import SessionManager from ..auth.session import SessionManager
from ..config import settings from ..config import settings
from ..monitoring.schema.schema_registry import schema_registry from ..monitoring.schema.schema_registry import schema_registry
@ -46,8 +47,21 @@ class GeoGuessrClient:
self, self,
session_token: Optional[str] = None, session_token: Optional[str] = None,
) -> httpx.AsyncClient: ) -> httpx.AsyncClient:
"""Get an authenticated HTTP client.""" """
session = await self.session_manager.get_session(session_token) Get an authenticated HTTP client.
In multi-user mode, if no session_token is provided, uses the current user's context
to get their session automatically.
"""
# Try to get session from current user context (multi-user mode)
user_context = get_current_user_context()
if user_context and user_context.is_authenticated:
# Use the session from the user's context
session = user_context.session
else:
# Fall back to session manager (legacy mode or no user context)
session = await self.session_manager.get_session(session_token)
if not session: if not session:
raise ValueError( raise ValueError(
"No valid session available. Please login first or set GEOGUESSR_NCFA_COOKIE." "No valid session available. Please login first or set GEOGUESSR_NCFA_COOKIE."

View file

@ -1,8 +1,21 @@
"""Auth module for GeoGuessr session.""" """Auth module for GeoGuessr session."""
from .multi_user_session import MultiUserSessionManager, multi_user_session_manager
from .request_context import (
get_current_user_context,
require_user_context,
set_current_user_context,
)
from .session import SessionManager, UserSession from .session import SessionManager, UserSession
from .user_context import UserContext
__all__ = [ __all__ = [
"UserSession", "UserSession",
"SessionManager", "SessionManager",
"UserContext",
"MultiUserSessionManager",
"multi_user_session_manager",
"get_current_user_context",
"require_user_context",
"set_current_user_context",
] ]

View file

@ -0,0 +1,192 @@
"""
Multi-user session management.
This module provides session management for multiple users,
where each API key can have its own GeoGuessr session.
"""
import asyncio
import logging
from typing import Optional
from ..config import settings
from .session import SessionManager, UserSession
from .user_context import UserContext
logger = logging.getLogger(__name__)
class MultiUserSessionManager:
"""
Manages GeoGuessr sessions for multiple users.
Each API key can have its own GeoGuessr session, allowing
multiple users to access their own accounts through the same
MCP server instance.
"""
def __init__(self):
"""Initialize the multi-user session manager."""
# Map API keys to their session managers
self._user_managers: dict[str, SessionManager] = {}
self._lock = asyncio.Lock()
# Create default session manager if default cookie is configured
if settings.DEFAULT_NCFA_COOKIE:
logger.info("Default GeoGuessr cookie configured - will be used as fallback")
async def get_user_context(self, api_key: str) -> UserContext:
"""
Get or create a user context for an API key.
Args:
api_key: The API key identifying the user
Returns:
UserContext: The context for this user
"""
async with self._lock:
# Get or create session manager for this user
if api_key not in self._user_managers:
# Create new session manager with default cookie as fallback
self._user_managers[api_key] = SessionManager(
default_cookie=settings.DEFAULT_NCFA_COOKIE
)
logger.info(f"Created new session manager for API key {api_key[:8]}...")
manager = self._user_managers[api_key]
# Get the session (may return default session if no user login)
session = await manager.get_session()
# Create user context
context = UserContext(api_key=api_key, session=session)
return context
async def login_user(
self, api_key: str, email: str, password: str
) -> tuple[str, UserSession]:
"""
Login a specific user (API key) to their GeoGuessr account.
Args:
api_key: The API key identifying the user
email: GeoGuessr email
password: GeoGuessr password
Returns:
tuple[str, UserSession]: (session_token, UserSession)
Raises:
ValueError: If login fails
"""
async with self._lock:
if api_key not in self._user_managers:
self._user_managers[api_key] = SessionManager()
manager = self._user_managers[api_key]
# Perform login
session_token, session = await manager.login(email, password)
logger.info(
f"User {session.username} logged in successfully for API key {api_key[:8]}..."
)
return session_token, session
async def logout_user(self, api_key: str, session_token: str) -> bool:
"""
Logout a specific user's session.
Args:
api_key: The API key identifying the user
session_token: The session token to logout
Returns:
bool: True if logout successful, False otherwise
"""
async with self._lock:
if api_key not in self._user_managers:
return False
manager = self._user_managers[api_key]
success = await manager.logout(session_token)
if success:
logger.info(f"User logged out for API key {api_key[:8]}...")
return success
async def set_user_cookie(self, api_key: str, cookie: str) -> bool:
"""
Set a GeoGuessr cookie for a specific user.
Args:
api_key: The API key identifying the user
cookie: The NCFA cookie value
Returns:
bool: True if cookie is valid, False otherwise
"""
# Validate cookie first
profile = await SessionManager.validate_cookie(cookie)
if not profile:
return False
async with self._lock:
if api_key not in self._user_managers:
self._user_managers[api_key] = SessionManager()
manager = self._user_managers[api_key]
await manager.set_default_cookie(cookie)
logger.info(
f"Cookie set for user {profile.get('nick', 'unknown')} (API key {api_key[:8]}...)"
)
return True
async def get_session_for_api_key(self, api_key: str) -> Optional[UserSession]:
"""
Get the active session for a specific API key.
Args:
api_key: The API key identifying the user
Returns:
UserSession if available, None otherwise
"""
async with self._lock:
if api_key not in self._user_managers:
return None
manager = self._user_managers[api_key]
return await manager.get_session()
async def get_auth_status(self, api_key: str) -> dict:
"""
Get authentication status for a specific user.
Args:
api_key: The API key identifying the user
Returns:
dict: Authentication status information
"""
context = await self.get_user_context(api_key)
return {
"authenticated": context.is_authenticated,
"user_id": context.user_id if context.is_authenticated else None,
"username": context.username if context.is_authenticated else None,
"api_key": f"{api_key[:8]}..." if len(api_key) > 8 else "***",
}
# Global multi-user session manager instance
multi_user_session_manager = MultiUserSessionManager()

View file

@ -0,0 +1,61 @@
"""
Request context utilities for accessing user context in tools.
This module provides utilities to access the current user context
from within MCP tools, allowing each tool to operate on behalf of
the authenticated user making the request.
"""
from contextvars import ContextVar
from typing import Optional
from .user_context import UserContext
# Context variable to store the current user context
_current_user_context: ContextVar[Optional[UserContext]] = ContextVar(
"current_user_context", default=None
)
def set_current_user_context(context: UserContext) -> None:
"""
Set the current user context for this request.
This should be called by middleware to set the context
for the duration of the request.
Args:
context: The UserContext for the current request
"""
_current_user_context.set(context)
def get_current_user_context() -> Optional[UserContext]:
"""
Get the current user context.
This can be called from within tools to access the user context
for the current request.
Returns:
UserContext if available, None otherwise
"""
return _current_user_context.get()
def require_user_context() -> UserContext:
"""
Get the current user context, raising an error if not available.
Raises:
RuntimeError: If no user context is available
Returns:
UserContext: The current user context
"""
context = get_current_user_context()
if context is None:
raise RuntimeError(
"No user context available. This tool must be called through the MCP server."
)
return context

View file

@ -0,0 +1,58 @@
"""
User context for multi-user support.
This module provides the UserContext class that tracks which user
is making a request and their associated GeoGuessr session.
"""
from dataclasses import dataclass
from typing import Optional
from .session import UserSession
@dataclass
class UserContext:
"""
Context for a specific user making a request.
This class is attached to each request and contains information
about the authenticated user and their GeoGuessr session.
"""
api_key: str
"""The API key used to authenticate this request"""
session: Optional[UserSession] = None
"""The GeoGuessr session for this user (if logged in)"""
@property
def user_id(self) -> str:
"""Get the user ID for this context."""
if self.session:
return self.session.user_id
return f"anonymous_{hash(self.api_key) % 10000:04d}"
@property
def username(self) -> str:
"""Get the username for this context."""
if self.session:
return self.session.username
return f"User-{hash(self.api_key) % 10000:04d}"
@property
def ncfa_cookie(self) -> Optional[str]:
"""Get the NCFA cookie for this user."""
if self.session:
return self.session.ncfa_cookie
return None
@property
def is_authenticated(self) -> bool:
"""Check if this user has a valid GeoGuessr session."""
return self.session is not None and self.session.is_valid()
def __repr__(self) -> str:
"""String representation of user context."""
auth_status = "authenticated" if self.is_authenticated else "not authenticated"
return f"UserContext(user_id={self.user_id}, {auth_status})"

View file

@ -1,7 +1,8 @@
""" """
Authentication middleware for MCP server. Authentication middleware for MCP server.
Provides Bearer token authentication for HTTP-based MCP transports. Provides Bearer token authentication for HTTP-based MCP transports
and attaches user context for multi-user support.
""" """
import logging import logging
@ -11,6 +12,7 @@ from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request from starlette.requests import Request
from starlette.responses import JSONResponse, Response from starlette.responses import JSONResponse, Response
from ..auth import multi_user_session_manager, set_current_user_context
from ..config import settings from ..config import settings
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -40,8 +42,12 @@ class AuthenticationMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next) -> Response: async def dispatch(self, request: Request, call_next) -> Response:
"""Process the request and validate authentication if enabled.""" """Process the request and validate authentication if enabled."""
# Skip authentication if disabled # If authentication is disabled, create a default user context
if not self.enabled: if not self.enabled:
# Use a default API key for all unauthenticated requests
user_context = await multi_user_session_manager.get_user_context("default")
request.state.user_context = user_context
set_current_user_context(user_context)
return await call_next(request) return await call_next(request)
# Skip authentication for health check endpoint # Skip authentication for health check endpoint
@ -90,7 +96,16 @@ class AuthenticationMiddleware(BaseHTTPMiddleware):
) )
# Authentication successful # Authentication successful
logger.debug(f"Authenticated request from {request.client.host}") logger.debug(f"Authenticated request from {request.client.host} with API key {token[:8]}...")
# Get or create user context for this API key
user_context = await multi_user_session_manager.get_user_context(token)
# Attach user context to request state and context variable
request.state.user_context = user_context
set_current_user_context(user_context)
logger.debug(f"Request context: {user_context}")
# Proceed with the request # Proceed with the request
response = await call_next(request) response = await call_next(request)

View file

@ -2,48 +2,48 @@
Tools for handling authentication and session management with GeoGuessr. Tools for handling authentication and session management with GeoGuessr.
This module provides utilities for login, logout, setting authentication cookies, This module provides utilities for login, logout, setting authentication cookies,
and checking the current authentication status. Its primary purpose is to ensure and checking the current authentication status. With multi-user support, each
interactions with the GeoGuessr API are authenticated securely and conveniently. API key can have its own GeoGuessr session.
The module integrates with FastMCP to expose authentication methods as tools The module integrates with FastMCP to expose authentication methods as tools
and uses the `SessionManager` for session storage and validation. and uses the multi-user session manager for per-user session storage.
Functions Functions
--------- ---------
- register_auth_tools: Registers a set of authentication tools with FastMCP. - register_auth_tools: Registers a set of authentication tools with FastMCP.
- get_current_session_token: Returns the currently active session token.
""" """
import logging import logging
import secrets
from datetime import datetime, timedelta, UTC
from typing import Optional
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
from ..auth.session import SessionManager, UserSession from ..auth import get_current_user_context, multi_user_session_manager
from ..config import settings
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Global session token storage
_current_session_token: Optional[str] = None
def register_auth_tools(mcp: FastMCP, session_manager=None):
"""
Register authentication-related tools.
def register_auth_tools(mcp: FastMCP, session_manager: SessionManager): Note: session_manager parameter is kept for backward compatibility but not used.
"""Register authentication-related tools.""" The multi_user_session_manager is used instead.
"""
@mcp.tool() @mcp.tool()
async def login(email: str, password: str) -> dict: async def login(email: str, password: str) -> dict:
""" """
Authenticate with GeoGuessr using email and password. Authenticate with GeoGuessr using email and password.
Creates a session that will be used for all later API calls. Creates a session for YOUR account that will be used for all later API calls.
Credentials are only used to get an authentication token and are Your credentials are only used to get an authentication token and are
not stored on the server. not stored on the server.
Each API key gets its own session, so multiple users can use the same
MCP server with their own GeoGuessr accounts.
Args: Args:
email: Your GeoGuessr account email email: Your GeoGuessr account email
password: Your GeoGuessr account password password: Your GeoGuessr account password
@ -51,11 +51,19 @@ def register_auth_tools(mcp: FastMCP, session_manager: SessionManager):
Returns: Returns:
Session information including username and session token Session information including username and session token
""" """
global _current_session_token # Get the current user's context (API key)
user_context = get_current_user_context()
if not user_context:
return {
"success": False,
"error": "No user context available. Authentication is required.",
}
try: try:
session_token, session = await session_manager.login(email, password) # Login using the multi-user session manager
_current_session_token = session_token session_token, session = await multi_user_session_manager.login_user(
user_context.api_key, email, password
)
return { return {
"success": True, "success": True,
@ -64,6 +72,7 @@ def register_auth_tools(mcp: FastMCP, session_manager: SessionManager):
"user_id": session.user_id, "user_id": session.user_id,
"session_token": session_token, "session_token": session_token,
"expires_at": session.expires_at.isoformat() if session.expires_at else None, "expires_at": session.expires_at.isoformat() if session.expires_at else None,
"multi_user_note": "Your session is tied to your API key. Other users with different API keys can have their own sessions.",
} }
except ValueError as e: except ValueError as e:
return {"success": False, "error": str(e)} return {"success": False, "error": str(e)}
@ -72,23 +81,32 @@ def register_auth_tools(mcp: FastMCP, session_manager: SessionManager):
return {"success": False, "error": f"An unexpected error occurred: {str(e)}"} return {"success": False, "error": f"An unexpected error occurred: {str(e)}"}
@mcp.tool() @mcp.tool()
async def logout() -> dict: async def logout(session_token: str) -> dict:
""" """
Logout from the current GeoGuessr session. Logout from your current GeoGuessr session.
Invalidates the current session token. Invalidates your session token. This only affects your session
(tied to your API key), not other users.
Args:
session_token: The session token to logout (from login response)
Returns:
Success status
""" """
global _current_session_token # Get the current user's context
user_context = get_current_user_context()
if not user_context:
return {"success": False, "error": "No user context available"}
if _current_session_token: success = await multi_user_session_manager.logout_user(
success = await session_manager.logout(_current_session_token) user_context.api_key, session_token
_current_session_token = None )
return {
"success": success,
"message": "Successfully logged out" if success else "No active session found",
}
return {"success": False, "message": "No active session"} return {
"success": success,
"message": "Successfully logged out" if success else "No active session found",
}
@mcp.tool() @mcp.tool()
async def set_ncfa_cookie(cookie: str) -> dict: async def set_ncfa_cookie(cookie: str) -> dict:
@ -98,83 +116,83 @@ def register_auth_tools(mcp: FastMCP, session_manager: SessionManager):
Use this if you've manually extracted the cookie from your browser. Use this if you've manually extracted the cookie from your browser.
The cookie will be validated before being accepted. The cookie will be validated before being accepted.
This cookie will be tied to YOUR API key, so it won't affect other users.
Args: Args:
cookie: The _ncfa cookie value from your browser cookie: The _ncfa cookie value from your browser
Returns:
Success status and user information
""" """
global _current_session_token # Get the current user's context
user_context = get_current_user_context()
if not user_context:
return {"success": False, "error": "No user context available"}
# Validate the cookie # Set the cookie for this user
profile = await session_manager.validate_cookie(cookie) success = await multi_user_session_manager.set_user_cookie(user_context.api_key, cookie)
if not profile: if not success:
return {"success": False, "error": "Invalid cookie - authentication failed"} return {"success": False, "error": "Invalid cookie - authentication failed"}
# Create a session from the cookie # Get updated session to show user info
session = UserSession( session = await multi_user_session_manager.get_session_for_api_key(user_context.api_key)
ncfa_cookie=cookie,
user_id=profile.get("id", ""),
username=profile.get("nick", ""),
email="manual@cookie",
expires_at=datetime.now(UTC) + timedelta(days=30),
)
# Store as a session if session:
session_token = secrets.token_urlsafe(32) return {
async with session_manager._lock: "success": True,
session_manager._sessions[session_token] = session "message": f"Cookie set successfully. Authenticated as {session.username}",
session_manager._user_sessions[session.user_id] = session_token "username": session.username,
"user_id": session.user_id,
}
_current_session_token = session_token return {"success": True, "message": "Cookie set successfully"}
return {
"success": True,
"message": f"Cookie set successfully. Authenticated as {session.username}",
"username": session.username,
"user_id": session.user_id,
"session_token": session_token,
}
@mcp.tool() @mcp.tool()
async def get_auth_status() -> dict: async def get_auth_status() -> dict:
""" """
Check the current authentication status. Check your current authentication status.
Returns information about the current session or available Returns information about your session (tied to your API key).
authentication methods. Each user with a different API key has their own independent session.
Returns:
Authentication status and user information
""" """
global _current_session_token # Get the current user's context
user_context = get_current_user_context()
if not user_context:
return {
"authenticated": False,
"message": "No user context available",
}
# Check for active session # Get auth status for this user
if _current_session_token: status = await multi_user_session_manager.get_auth_status(user_context.api_key)
session = await session_manager.get_session(_current_session_token)
if session and session.is_valid():
return {
"authenticated": True,
"method": "session",
"username": session.username,
"user_id": session.user_id,
"expires_at": session.expires_at.isoformat() if session.expires_at else None,
}
# Check for environment variable if status["authenticated"]:
env_cookie = settings.DEFAULT_NCFA_COOKIE return {
if env_cookie: **status,
profile = await session_manager.validate_cookie(env_cookie) "message": f"Authenticated as {status['username']}",
if profile: "multi_user_info": "Your session is independent. Other API keys have their own sessions.",
return { }
"authenticated": True,
"method": "environment_variable",
"username": profile.get("nick", "Unknown"),
"user_id": profile.get("id", "Unknown"),
}
return { return {
"authenticated": False, **status,
"message": "Not authenticated. Use 'login' with credentials or 'set_ncfa_cookie' with a valid cookie.", "message": "Not authenticated. Use 'login' with credentials or 'set_ncfa_cookie' with a valid cookie.",
"available_methods": ["login(email, password)", "set_ncfa_cookie(cookie)"], "available_methods": ["login(email, password)", "set_ncfa_cookie(cookie)"],
} }
def get_current_session_token() -> Optional[str]: def get_current_session_token():
"""Get the current session token for use by other tools.""" """
return _current_session_token Deprecated: This function is no longer used in multi-user mode.
Sessions are now managed per-API-key automatically.
Use get_current_user_context() instead to access user-specific session data.
"""
logger.warning(
"get_current_session_token() is deprecated in multi-user mode. "
"Use get_current_user_context() instead."
)
return None