450 lines
21 KiB
Python
450 lines
21 KiB
Python
"""Account HTTP API; trusted adapters alone hold Hermes connection secrets."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hmac
|
|
import inspect
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import time
|
|
from collections import defaultdict, deque
|
|
from contextlib import asynccontextmanager
|
|
from dataclasses import dataclass, field
|
|
from typing import Protocol
|
|
from urllib.parse import urlsplit
|
|
|
|
from fastapi import Depends, FastAPI, HTTPException, Request, Response
|
|
|
|
from .accounts import Accounts, password_hash, password_matches, public_user, validate_password, validate_username
|
|
|
|
COOKIE = 'hope_session'
|
|
MAX_BODY = 8192
|
|
|
|
|
|
class Provisioner(Protocol):
|
|
"""All methods async. remove MUST verify Hope/user ownership before deletion.
|
|
|
|
ensure is idempotent and returns {container_name, hermes_url, hermes_api_key}.
|
|
ready checks the owned Hermes API health, not just container existence.
|
|
Private home/workspace must be retained by remove unless explicitly erased.
|
|
"""
|
|
async def ensure(self, user: dict) -> dict: ...
|
|
async def remove(self, user: dict) -> None: ...
|
|
async def ready(self, user: dict) -> bool: ...
|
|
|
|
|
|
@dataclass
|
|
class Settings:
|
|
db_path: str = field(default_factory=lambda: os.getenv('CONTROL_DB_PATH', '/data/control/accounts.sqlite3'))
|
|
admin_username: str = field(default_factory=lambda: os.getenv('CONTROL_ADMIN_USERNAME', 'admin'))
|
|
admin_password: str | None = field(default_factory=lambda: os.getenv('CONTROL_ADMIN_PASSWORD'))
|
|
service_secret: str | None = field(default_factory=lambda: os.getenv('CONTROL_SERVICE_SECRET'))
|
|
allowed_origins: tuple[str, ...] = field(default_factory=lambda: tuple(x.strip().rstrip('/') for x in os.getenv('CONTROL_ALLOWED_ORIGINS', '').split(',') if x.strip()))
|
|
session_ttl: int = 86400
|
|
reconcile_interval: float = 30
|
|
login_limit: int = 10
|
|
login_window: int = 300
|
|
|
|
|
|
async def bounded_json(request):
|
|
if request.headers.get('content-type', '').split(';', 1)[0].strip().lower() != 'application/json':
|
|
raise HTTPException(415, 'Content-Type application/json required')
|
|
data = bytearray()
|
|
async for chunk in request.stream():
|
|
data.extend(chunk)
|
|
if len(data) > MAX_BODY:
|
|
raise HTTPException(413, 'Request body too large')
|
|
try:
|
|
value = json.loads(data)
|
|
except (ValueError, UnicodeDecodeError):
|
|
raise HTTPException(400, 'Invalid JSON') from None
|
|
if not isinstance(value, dict):
|
|
raise HTTPException(400, 'JSON object required')
|
|
return value
|
|
|
|
|
|
def bearer_token(request):
|
|
authorization = request.headers.get('authorization')
|
|
if authorization is None:
|
|
return None
|
|
parts = authorization.split()
|
|
if len(parts) != 2 or parts[0].lower() != 'bearer' or not 20 <= len(parts[1]) <= 256:
|
|
raise HTTPException(401, 'Invalid authorization')
|
|
return parts[1]
|
|
|
|
|
|
def check_origin(request, *, login=False):
|
|
"""No forwarded-header trust; explicit origins required behind TLS proxy.
|
|
|
|
Native bearer clients are exempt. Login without Origin is allowed only for
|
|
JSON native clients with no session cookie and no browser Fetch Metadata.
|
|
"""
|
|
if bearer_token(request) is not None:
|
|
return
|
|
origin = request.headers.get('origin')
|
|
if not origin:
|
|
if login and not request.cookies.get(COOKIE) and not request.headers.get('sec-fetch-site'):
|
|
return
|
|
raise HTTPException(403, 'Origin required')
|
|
settings = request.app.state.settings
|
|
allowed = settings.allowed_origins or (str(request.base_url).rstrip('/'),)
|
|
if origin not in allowed or origin == 'null':
|
|
raise HTTPException(403, 'Origin not allowed')
|
|
|
|
|
|
async def current_user(request: Request):
|
|
token = bearer_token(request) or request.cookies.get(COOKIE)
|
|
if not token:
|
|
raise HTTPException(401, 'Authentication required')
|
|
user = await asyncio.to_thread(request.app.state.accounts.resolve_session, token)
|
|
if user is None:
|
|
raise HTTPException(401, 'Session invalid or expired')
|
|
if request.method not in ('GET', 'HEAD', 'OPTIONS'):
|
|
check_origin(request)
|
|
request.state.session_token = token
|
|
return public_user(user)
|
|
|
|
|
|
async def administrator(user=Depends(current_user)):
|
|
if user['role'] != 'admin':
|
|
raise HTTPException(403, 'Administrator required')
|
|
return user
|
|
|
|
|
|
def internal_authorized(request: Request):
|
|
secret = request.app.state.settings.service_secret
|
|
supplied = request.headers.get('x-control-secret', '')
|
|
if not secret or len(secret) < 32 or not supplied or not hmac.compare_digest(secret.encode('utf-8'), supplied.encode('utf-8')):
|
|
raise HTTPException(401, 'Internal authorization required')
|
|
|
|
|
|
class Lifecycle:
|
|
def __init__(self, accounts, provisioner):
|
|
self.accounts = accounts
|
|
self.provisioner = provisioner
|
|
self.tasks = {}
|
|
self.closed = False
|
|
|
|
def schedule(self, user_id):
|
|
if not self.closed and (user_id not in self.tasks or self.tasks[user_id].done()):
|
|
self.tasks[user_id] = asyncio.create_task(self.run(user_id))
|
|
|
|
async def call(self, method, user):
|
|
# Native async adapters must offload their own blocking Docker/HTTP calls.
|
|
# Also tolerate synchronous test/legacy adapters without blocking ASGI.
|
|
func = getattr(self.provisioner, method)
|
|
if inspect.iscoroutinefunction(func):
|
|
return await func(user)
|
|
result = await asyncio.to_thread(func, user)
|
|
return await result if inspect.isawaitable(result) else result
|
|
|
|
async def run(self, user_id):
|
|
user = await asyncio.to_thread(self.accounts.get, user_id)
|
|
if user is None:
|
|
return
|
|
try:
|
|
if user['desired'] == 'deleted':
|
|
await self.call('remove', user)
|
|
await asyncio.to_thread(self.accounts.finish_delete, user_id)
|
|
return
|
|
await asyncio.to_thread(self.accounts.update, user_id, status='provisioning', error=None)
|
|
metadata = await self.call('ensure', user)
|
|
if not isinstance(metadata, dict):
|
|
raise ValueError('Invalid provisioning metadata')
|
|
url = metadata.get('hermes_url')
|
|
key = metadata.get('hermes_api_key')
|
|
parsed = urlsplit(url or '')
|
|
if parsed.scheme not in ('http', 'https') or not parsed.hostname or not isinstance(key, str) or not key:
|
|
raise ValueError('Invalid provisioning metadata')
|
|
# Deterministic name cannot be overridden by adapter metadata.
|
|
if metadata.get('container_name', user['container_name']) != user['container_name']:
|
|
raise ValueError('Unexpected container ownership')
|
|
await asyncio.to_thread(self.accounts.update, user_id, hermes_url=url, hermes_api_key=key)
|
|
user.update(hermes_url=url, hermes_api_key=key)
|
|
if not await self.call('ready', user):
|
|
raise RuntimeError('Hermes not healthy')
|
|
current = await asyncio.to_thread(self.accounts.get, user_id)
|
|
if current and current['desired'] == 'deleted':
|
|
await self.call('remove', current)
|
|
await asyncio.to_thread(self.accounts.finish_delete, user_id)
|
|
else:
|
|
await asyncio.to_thread(self.accounts.update, user_id, status='ready', error=None)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
# Exception messages can contain upstream URLs, keys or Docker env.
|
|
current = await asyncio.to_thread(self.accounts.get, user_id)
|
|
if current:
|
|
deleting = current['desired'] == 'deleted'
|
|
await asyncio.to_thread(self.accounts.update, user_id, status='deleting' if deleting else 'error', error='Container removal failed; retry pending' if deleting else 'Hermes provisioning failed; retry pending')
|
|
|
|
async def reconcile(self):
|
|
for user in await asyncio.to_thread(self.accounts.users):
|
|
if user['desired'] == 'deleted' or user['status'] != 'ready':
|
|
self.schedule(user['id'])
|
|
elif user['id'] not in self.tasks or self.tasks[user['id']].done():
|
|
try:
|
|
healthy = await self.call('ready', user)
|
|
except Exception:
|
|
healthy = False
|
|
if not healthy:
|
|
await asyncio.to_thread(self.accounts.update, user['id'], status='error', error='Hermes health check failed; retry pending')
|
|
self.schedule(user['id'])
|
|
|
|
async def loop(self, interval):
|
|
while True:
|
|
await self.reconcile()
|
|
await asyncio.sleep(interval)
|
|
|
|
async def stop(self):
|
|
self.closed = True
|
|
tasks = list(self.tasks.values())
|
|
for task in tasks:
|
|
task.cancel()
|
|
await asyncio.gather(*tasks, return_exceptions=True)
|
|
|
|
|
|
def create_app(settings=None, provisioner=None, model_catalog=None):
|
|
settings = settings or Settings()
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app):
|
|
store = Accounts(settings.db_path)
|
|
app.state.accounts = store
|
|
try:
|
|
if not store.users():
|
|
if not settings.admin_password:
|
|
raise RuntimeError('CONTROL_ADMIN_PASSWORD required to bootstrap an empty account database')
|
|
await asyncio.to_thread(store.create, settings.admin_username, 'Administrator', 'admin', settings.admin_password)
|
|
adapter = provisioner
|
|
if adapter is None:
|
|
from .provisioner import Provisioner as HermesProvisioner
|
|
adapter = HermesProvisioner()
|
|
from .model_catalog import ModelCatalog
|
|
catalog = model_catalog if model_catalog is not None else ModelCatalog()
|
|
app.state.model_catalog = catalog
|
|
app.state.lifecycle = Lifecycle(store, adapter)
|
|
# Revalidate even previously-ready containers after control restart.
|
|
for user in store.users():
|
|
app.state.lifecycle.schedule(user['id'])
|
|
app.state.dummy_hash = await asyncio.to_thread(password_hash, 'timing-only-password')
|
|
loop = asyncio.create_task(app.state.lifecycle.loop(settings.reconcile_interval))
|
|
try:
|
|
yield
|
|
finally:
|
|
loop.cancel()
|
|
await asyncio.gather(loop, return_exceptions=True)
|
|
await app.state.lifecycle.stop()
|
|
if hasattr(catalog, 'aclose'):
|
|
await catalog.aclose()
|
|
if hasattr(adapter, 'aclose'):
|
|
await adapter.aclose()
|
|
finally:
|
|
store.close()
|
|
|
|
app = FastAPI(title='Hope account control', lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None)
|
|
app.state.settings = settings
|
|
attempts = defaultdict(deque)
|
|
|
|
@app.middleware('http')
|
|
async def no_store(request, call_next):
|
|
response = await call_next(request)
|
|
response.headers['Cache-Control'] = 'no-store'
|
|
return response
|
|
|
|
@app.post('/api/auth/login')
|
|
async def login(request: Request, response: Response):
|
|
check_origin(request, login=True)
|
|
body = await bounded_json(request)
|
|
try:
|
|
username = validate_username(body.get('username'))
|
|
password = validate_password(body.get('password'))
|
|
except ValueError as exc:
|
|
raise HTTPException(400, str(exc)) from None
|
|
peer = request.client.host if request.client else 'unknown'
|
|
# nginx /api/ overwrites X-Real-IP with $remote_addr. Only its direct
|
|
# loopback connection is trusted; never trust X-Forwarded-For.
|
|
try:
|
|
if ipaddress.ip_address(peer).is_loopback:
|
|
forwarded = request.headers.get('x-real-ip')
|
|
if forwarded:
|
|
peer = str(ipaddress.ip_address(forwarded))
|
|
except ValueError:
|
|
pass
|
|
now = time.monotonic()
|
|
for key in list(attempts):
|
|
bucket = attempts[key]
|
|
while bucket and bucket[0] <= now - settings.login_window:
|
|
bucket.popleft()
|
|
if not bucket:
|
|
del attempts[key]
|
|
scope = (peer, username)
|
|
bucket = attempts.get(scope)
|
|
if bucket is not None and len(bucket) >= settings.login_limit:
|
|
raise HTTPException(429, 'Too many login attempts', headers={'Retry-After': str(settings.login_window)})
|
|
if bucket is None:
|
|
if len(attempts) >= 4096:
|
|
raise HTTPException(429, 'Too many login attempts')
|
|
bucket = attempts[scope]
|
|
# Reserve a slot while scrypt is in flight to prevent concurrent bypass.
|
|
# Successful requests release their own slot; only failures persist.
|
|
slot = time.monotonic()
|
|
bucket.append(slot)
|
|
store = app.state.accounts
|
|
user = await asyncio.to_thread(store.by_username, username)
|
|
valid = await asyncio.to_thread(password_matches, password, user['password_hash'] if user else app.state.dummy_hash)
|
|
if not valid or not user or user['desired'] == 'deleted':
|
|
raise HTTPException(401, 'Invalid username or password')
|
|
try:
|
|
token, expires = await asyncio.to_thread(store.issue_session, user['id'], settings.session_ttl, user['password_hash'])
|
|
except ValueError:
|
|
raise HTTPException(401, 'Account unavailable') from None
|
|
try:
|
|
bucket.remove(slot)
|
|
except ValueError:
|
|
pass # Expired while the request was in flight.
|
|
if not bucket and attempts.get(scope) is bucket:
|
|
attempts.pop(scope, None)
|
|
response.set_cookie(COOKIE, token, max_age=settings.session_ttl, httponly=True, secure=True, samesite='strict', path='/')
|
|
return {'token': token, 'user': public_user(user), 'expires_at': expires}
|
|
|
|
@app.post('/api/auth/logout')
|
|
async def logout(request: Request, response: Response, user=Depends(current_user)):
|
|
await asyncio.to_thread(app.state.accounts.revoke, request.state.session_token)
|
|
response.delete_cookie(COOKIE, path='/', secure=True, httponly=True, samesite='strict')
|
|
return {'ok': True}
|
|
|
|
@app.get('/api/auth/me')
|
|
async def me(user=Depends(current_user)):
|
|
return {'user': user}
|
|
|
|
@app.get('/api/settings/model')
|
|
async def model_settings(user=Depends(current_user)):
|
|
try:
|
|
saved = await asyncio.to_thread(app.state.accounts.get_model_settings, user['id'])
|
|
except KeyError:
|
|
raise HTTPException(401, 'Account unavailable') from None
|
|
try:
|
|
models = await app.state.model_catalog.models()
|
|
available, message = True, ''
|
|
except Exception:
|
|
models, available = [], False
|
|
message = 'Model list unavailable. Your saved settings are unchanged; try again.'
|
|
return {'settings': saved, 'models': models, 'catalog_available': available,
|
|
'message': message, 'thinking_note': 'Thinking levels are requests. Some models may limit or ignore them.'}
|
|
|
|
@app.put('/api/settings/model')
|
|
async def save_model_settings(request: Request, user=Depends(current_user)):
|
|
body = await bounded_json(request)
|
|
if set(body) != {'model', 'thinking_level'}:
|
|
raise HTTPException(400, 'Only model and thinking_level are accepted')
|
|
try:
|
|
models = await app.state.model_catalog.models()
|
|
except Exception:
|
|
raise HTTPException(503, 'Model list unavailable. Settings were not changed; try again.') from None
|
|
selected = next((m for m in models if m['id'] == body.get('model')), None)
|
|
if selected is None or body.get('thinking_level') not in selected['thinking_levels']:
|
|
raise HTTPException(400, 'Choose an available model and thinking level')
|
|
try:
|
|
saved = await asyncio.to_thread(app.state.accounts.set_model_settings, user['id'], body['model'], body['thinking_level'])
|
|
except ValueError:
|
|
raise HTTPException(400, 'Invalid model settings') from None
|
|
except KeyError:
|
|
raise HTTPException(401, 'Account unavailable') from None
|
|
return {'settings': saved, 'message': 'Saved for your account. Applies to your next request.'}
|
|
|
|
@app.get('/api/admin/users')
|
|
async def list_users(user=Depends(administrator)):
|
|
return {'users': [public_user(u) for u in await asyncio.to_thread(app.state.accounts.users)]}
|
|
|
|
@app.post('/api/admin/users', status_code=201)
|
|
async def create_user(request: Request, admin=Depends(administrator)):
|
|
body = await bounded_json(request)
|
|
try:
|
|
user = await asyncio.to_thread(app.state.accounts.create, body.get('username'), body.get('display_name'), body.get('role', 'user'), body.get('password'))
|
|
except ValueError as exc:
|
|
raise HTTPException(400, str(exc)) from None
|
|
except sqlite3.IntegrityError:
|
|
raise HTTPException(409, 'Username already exists') from None
|
|
app.state.lifecycle.schedule(user['id'])
|
|
return {'user': public_user(user)}
|
|
|
|
@app.delete('/api/admin/users/{user_id}', status_code=202)
|
|
async def delete_user(user_id: str, admin=Depends(administrator)):
|
|
try:
|
|
user = await asyncio.to_thread(app.state.accounts.begin_delete, user_id)
|
|
except KeyError:
|
|
raise HTTPException(404, 'User not found') from None
|
|
except ValueError as exc:
|
|
raise HTTPException(409, str(exc)) from None
|
|
app.state.lifecycle.schedule(user_id)
|
|
return {'user': public_user(user)}
|
|
|
|
@app.post('/api/admin/users/{user_id}/password')
|
|
async def reset_password(user_id: str, request: Request, admin=Depends(administrator)):
|
|
body = await bounded_json(request)
|
|
try:
|
|
await asyncio.to_thread(app.state.accounts.reset_password, user_id, body.get('password'))
|
|
except ValueError as exc:
|
|
raise HTTPException(400, str(exc)) from None
|
|
except KeyError:
|
|
raise HTTPException(404, 'User not found') from None
|
|
return {'ok': True}
|
|
|
|
@app.post('/api/admin/users/{user_id}/retry', status_code=202)
|
|
async def retry(user_id: str, admin=Depends(administrator)):
|
|
user = await asyncio.to_thread(app.state.accounts.get, user_id)
|
|
if not user:
|
|
raise HTTPException(404, 'User not found')
|
|
app.state.lifecycle.schedule(user_id)
|
|
return {'user': public_user(user)}
|
|
|
|
def connection(user):
|
|
if not user or user['desired'] == 'deleted':
|
|
raise HTTPException(404, 'User not found')
|
|
if user['status'] != 'ready' or not user['hermes_url'] or not user['hermes_api_key']:
|
|
raise HTTPException(503, 'User Hermes is not ready')
|
|
return {'user': public_user(user), 'hermes_url': user['hermes_url'], 'hermes_api_key': user['hermes_api_key'],
|
|
'model_settings': app.state.accounts.get_model_settings(user['id'])}
|
|
|
|
@app.get('/internal/users/{user_id}', dependencies=[Depends(internal_authorized)])
|
|
async def internal_user(user_id: str):
|
|
return connection(await asyncio.to_thread(app.state.accounts.get, user_id))
|
|
|
|
@app.post('/internal/resolve-session', dependencies=[Depends(internal_authorized)])
|
|
async def resolve_session(request: Request):
|
|
body = await bounded_json(request)
|
|
user = await asyncio.to_thread(app.state.accounts.resolve_session, body.get('token'))
|
|
if user is None:
|
|
raise HTTPException(401, 'Session invalid or expired')
|
|
return connection(user)
|
|
|
|
@app.post('/internal/rooms', dependencies=[Depends(internal_authorized)], status_code=201)
|
|
async def create_room(request: Request):
|
|
body = await bounded_json(request)
|
|
user_id, room = body.get('user_id'), body.get('room_name')
|
|
ttl = body.get('ttl', 3600)
|
|
if not isinstance(user_id, str) or not isinstance(room, str) or not 1 <= len(room) <= 128 or not all(c.isascii() and (c.isalnum() or c in '-_') for c in room):
|
|
raise HTTPException(400, 'Invalid room grant')
|
|
if type(ttl) is not int or not 1 <= ttl <= 86400:
|
|
raise HTTPException(400, 'Invalid room expiry')
|
|
try:
|
|
await asyncio.to_thread(app.state.accounts.grant_room, user_id, room, ttl, body.get('session_token'))
|
|
except ValueError:
|
|
raise HTTPException(409, 'Account not ready') from None
|
|
except sqlite3.IntegrityError:
|
|
raise HTTPException(409, 'Room already granted') from None
|
|
return {'room_name': room, 'user_id': user_id}
|
|
|
|
@app.get('/internal/rooms/{room_name}', dependencies=[Depends(internal_authorized)])
|
|
async def resolve_room(room_name: str):
|
|
return connection(await asyncio.to_thread(app.state.accounts.resolve_room, room_name))
|
|
|
|
@app.get('/health')
|
|
async def health():
|
|
return {'status': 'ok'}
|
|
|
|
return app
|