Upload 158 files

.dockerignore

@@ -2,12 +2,18 @@
 .venv
 .github
 .env
+.env.*
+!.env.example
 .pytest_cache
 .ruff_cache
 __pycache__
 *.py[cod]
 *.log
 *.zip
+*.db
+*.sqlite
+*.sqlite3
+data
 outputs/*
 !outputs/.gitkeep
 temp/*

.env.example

@@ -1,3 +1,4 @@
+APP_NAME=MediaRouter
 TEMP_DIR=./temp
 OUTPUT_DIR=./outputs
 TEMPLATE_DIR=./app/templates/categories
@@ -14,3 +15,18 @@ ALLOW_PRIVATE_URLS=false
 BASE_URL=
 FFMPEG_BINARY=ffmpeg
 FFPROBE_BINARY=ffprobe
+AUTH_ENABLED=true
+DATABASE_URL=sqlite+aiosqlite:///./data/mediarouter.db
+AUTH_ROLE_SCOPES={}
+AUTH_BOOTSTRAP_KEY_HASH=
+AUTH_BOOTSTRAP_KEY_PREFIX=
+AUTH_BOOTSTRAP_KEY_NAME=Bootstrap Administrator
+AUTH_BOOTSTRAP_ENVIRONMENT=live
+AUTH_LAST_USED_UPDATE_SECONDS=60
+AUTH_DEFAULT_REQUESTS_PER_MINUTE=100
+AUTH_DEFAULT_CONCURRENT_JOBS=10
+AUTH_DEFAULT_UPLOADS_PER_HOUR=20
+AUTH_DEFAULT_PROCESSING_BYTES_PER_DAY=107374182400
+AUTH_TRUST_PROXY_HEADERS=true
+# Required only when running the standalone stdio MCP transport with auth enabled.
+MCP_STDIO_API_KEY=

.gitignore

@@ -1,4 +1,6 @@
 .env
+.env.*
+!.env.example
 .venv/
 venv/
 __pycache__/
@@ -14,3 +16,7 @@ outputs/*
 temp/*
 !temp/.gitkeep
 *.zip
+*.db
+*.sqlite
+*.sqlite3
+data/

Dockerfile

@@ -5,8 +5,10 @@ ENV PYTHONDONTWRITEBYTECODE=1 \
     PIP_NO_CACHE_DIR=1 \
     PIP_DISABLE_PIP_VERSION_CHECK=1 \
     HF_HOME=/home/user/.cache/huggingface \
+    APP_NAME=MediaRouter \
     TEMP_DIR=/app/temp \
     OUTPUT_DIR=/app/outputs \
+    DATABASE_URL=sqlite+aiosqlite:////app/data/mediarouter.db \
     PORT=7860
 
 RUN apt-get update \
@@ -26,7 +28,7 @@ RUN python -m pip install --upgrade pip \
     && python -m pip install -r requirements.txt
 
 RUN useradd --create-home --uid 1000 user \
-    && mkdir -p /app/temp /app/outputs /home/user/.cache/huggingface \
+    && mkdir -p /app/temp /app/outputs /app/data /home/user/.cache/huggingface \
     && chown -R user:user /app /home/user
 
 COPY --chown=user:user app ./app

README.md

@@ -1,5 +1,5 @@
 ---
-title: Enterprise Media Processing API
+title: MediaRouter
 emoji: 🎬
 colorFrom: blue
 colorTo: purple
@@ -8,9 +8,9 @@ app_port: 7860
 pinned: false
 ---
 
-# Enterprise Media Processing API
+# MediaRouter
 
-A production-oriented, CPU-optimized REST, Model Context Protocol (MCP), and YAML workflow API for FFmpeg, FFprobe, yt-dlp, and faster-whisper. It is designed to run unchanged as a Hugging Face Docker Space and to act as a reusable media backend for AI clients, n8n, and other automation systems.
+MediaRouter is a production-oriented, CPU-optimized REST, Model Context Protocol (MCP), and YAML workflow API for FFmpeg, FFprobe, yt-dlp, and faster-whisper. It is designed to run unchanged as a Hugging Face Docker Space and to act as a reusable media backend for AI clients, n8n, and other automation systems.
 
 The service accepts multipart uploads, JSON URLs, JSON Base64, n8n binary objects, and streamed raw request bodies. Every input becomes an `InputMedia` before it enters the operation layer, so operations never need to know how media arrived.
 
@@ -21,6 +21,9 @@ REST /v1/*                       MCP stdio or /mcp/
      │                                  │
      ├───────────────┬──────────────────┘
                      ▼
+        API-key authentication + scopes
+          rate limits + audit logging
+                     │
             request-ID + JSON logging
                      │
                 InputResolver
@@ -57,6 +60,7 @@ media-api/
 │   ├── mcp/              # MCP server, registry, tools, resources, prompts
 │   ├── models/           # InputMedia and request/result models
 │   ├── operations/       # reusable FFmpeg operation functions
+│   ├── security/         # API keys, scopes, roles, limits, audit, migrations
 │   ├── services/         # FFmpeg, FFprobe, resolver, downloads, Whisper
 │   ├── templates/        # YAML schemas, loader, registry, executor, categories
 │   ├── workers/          # asynchronous expiry cleanup worker
@@ -90,6 +94,8 @@ python3.10 -m venv .venv
 . .venv/bin/activate
 pip install -r requirements.txt
 cp .env.example .env
+python -m app.security.cli generate-bootstrap --environment test
+# Put the printed AUTH_BOOTSTRAP_* values in .env and save API_KEY securely.
 uvicorn main:app --host 0.0.0.0 --port 7860
 ```
 
@@ -108,14 +114,139 @@ The image uses `python:3.10-slim`, installs FFmpeg, FFprobe, ImageMagick and the
 
 1. Create a new Space and select **Docker** as the SDK.
 2. Push the contents of this directory to the Space repository. Keep the YAML block at the top of this README; `app_port` is already `7860`.
-3. Add environment variables under **Settings → Variables and secrets** if the defaults need changing. Do not store secrets in `.env` in Git.
+3. Generate the initial administrator key locally with `python -m app.security.cli generate-bootstrap --environment live`. Store `API_KEY` in a password manager. Add only the printed `AUTH_BOOTSTRAP_KEY_HASH`, `AUTH_BOOTSTRAP_KEY_PREFIX`, and `AUTH_BOOTSTRAP_ENVIRONMENT` under **Settings → Secrets**. Do not commit them.
 4. Wait for the Docker build. The first Whisper call downloads the selected model to the Hugging Face cache. Persistent storage is optional, but avoids downloading models again after a cold rebuild.
-5. Check `https://<owner>-<space>.hf.space/health`, then use `/docs` or call `/v1/*` from n8n.
+5. Check the public `https://<owner>-<space>.hf.space/health`, then use the saved key for every `/v1/*` or `/mcp/` request.
 
 If using the supplied `media-api-huggingface.zip`, extract it first and push the extracted files so `Dockerfile` and this `README.md` are at the Space repository root. Do not commit the ZIP as the only repository file.
 
 Use one Uvicorn process in a CPU Space. FFmpeg and Whisper concurrency is managed in-process by `MAX_WORKERS`; multiple Uvicorn workers duplicate Whisper models and memory.
 
+For durable keys, audit records, and rate aggregates, attach Hugging Face persistent storage and set `DATABASE_URL=sqlite+aiosqlite:////data/mediarouter.db`. The default `./data/mediarouter.db` is appropriate locally but follows the Space filesystem lifecycle. The security layer uses SQLAlchemy so a future external database migration does not change authentication contracts; this release includes and supports the `aiosqlite` driver.
+
+## Authentication and authorization
+
+MediaRouter uses stateless opaque API keys. There are no passwords, login sessions, cookies, or JWTs. Except for the public endpoints below, every REST and MCP request must send:
+
+```http
+Authorization: Bearer mp_live_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
+```
+
+Keys contain at least 256 bits of cryptographically secure entropy and use an environment prefix (`mp_live_` or `mp_test_`). The database stores only a SHA-256 hash, a short display prefix, lifecycle state, scopes, role, limits, timestamps, and operator metadata. A plaintext secret is returned once at creation or rotation and cannot be recovered later.
+
+The only unauthenticated endpoints are `GET /`, `GET /health`, `GET /version`, `GET /docs`, `GET /openapi.json`, and `GET /redoc`. `GET /v1/auth/context` is protected but requires no product scope, allowing a client to validate a key and retrieve its safe effective authorization context.
+
+### Bootstrap and key creation
+
+Generate a first administrator without putting plaintext key material in the server configuration:
+
+```bash
+python -m app.security.cli generate-bootstrap --environment live
+```
+
+The command prints `API_KEY` once and hash-only `AUTH_BOOTSTRAP_*` values. Store `API_KEY` in a secret manager; configure the hash, prefix, and environment on the server. On first startup, and only when the `api_keys` table is empty, MediaRouter inserts that bootstrap administrator. Removing the bootstrap variables after the first successful startup is recommended. If direct database access is available, `python -m app.security.cli create --name "Recovery Admin" --role admin` can create a recovery key and likewise prints its secret once.
+
+The idempotent migration is documented in `app/security/migrations/0001_api_key_security.sql`. Startup applies equivalent SQLAlchemy metadata for `api_keys`, `audit_logs`, and `rate_limits`, including indexes on key prefix/hash, lifecycle state, expiration, audit request/key timestamps, and rate buckets. Back up the database before changing schema or moving persistent storage.
+
+An administrator can then create a narrower key:
+
+```bash
+curl -X POST "$MEDIAROUTER_URL/v1/api-keys" \
+  -H "Authorization: Bearer $MEDIAROUTER_API_KEY" \
+  -H 'Content-Type: application/json' \
+  -d '{
+    "name":"Production n8n",
+    "environment":"live",
+    "role":"operator",
+    "expires_in_seconds":2592000,
+    "notes":"Media publishing workflow"
+  }'
+```
+
+The `api_key` property in the response is the only copy of the new secret. The lifecycle API is:
+
+| Method | Endpoint | Purpose |
+|---|---|---|
+| `GET` | `/v1/api-keys` | List safe key metadata |
+| `POST` | `/v1/api-keys` | Create a key and return its secret once |
+| `GET` / `PATCH` / `DELETE` | `/v1/api-keys/{id}` | Read, rename/update, or irrevocably revoke |
+| `POST` | `/v1/api-keys/{id}/rotate` | Create a replacement; optional grace period up to 24 hours |
+| `POST` | `/v1/api-keys/{id}/disable` | Temporarily reject an active key |
+| `POST` | `/v1/api-keys/{id}/enable` | Re-enable a disabled, non-revoked key |
+| `GET` | `/v1/api-keys/capabilities` | Discover scopes, configured roles, and limit defaults |
+| `GET` | `/v1/audit-logs` | Read authenticated request audit records |
+
+Revocation is immediate and irreversible without restarting the service. Rotation creates a distinct record and secret. With `{"grace_period_seconds":300}`, both keys work for five minutes; without a grace period, the old key is rejected immediately. Expired keys always fail, including disabled keys that are later enabled.
+
+### Scopes and roles
+
+Scopes are enforced before route execution. Explicit scopes are combined with the selected role; `admin` grants all scopes.
+
+| Domain | Scopes |
+|---|---|
+| Templates | `templates:read`, `templates:run` |
+| Operations | `operations:read`, `operations:execute` |
+| Jobs | `jobs:read`, `jobs:create`, `jobs:cancel` |
+| Assets | `assets:read`, `assets:write`, `assets:delete` |
+| MCP | `mcp:read`, `mcp:execute` |
+| System | `system:read` |
+| Administration | `admin` |
+
+Built-in roles are `admin` (all access), `developer` (read/run/execute, job control, and asset writes), `operator` (read/run/execute and job control), and `viewer` (read-only). Override or add roles with an `AUTH_ROLE_SCOPES` JSON object, for example `{"publisher":["templates:read","templates:run","assets:read"]}`. Unknown scopes are rejected at startup or key validation rather than silently ignored.
+
+### Limits, auditing, and error contracts
+
+Each key has independently configurable requests per minute, concurrent processing jobs, uploads per hour, and processing bytes per UTC day. Defaults are 100, 10, 20, and 100 GiB. A limit failure returns `429 Too Many Requests` with a `Retry-After` header. Hugging Face Spaces should run one Uvicorn process, which also prevents duplicate in-memory concurrency and Whisper state.
+
+Every authenticated HTTP request records its request ID, key ID/name, trusted client IP, user agent, method, endpoint, status, processing time, and upload/download byte counts. Standalone stdio MCP calls generate equivalent tool audit records. Logs never contain plaintext API keys.
+
+Authentication failures are intentionally indistinguishable:
+
+```json
+{"error":"Unauthorized","message":"Invalid or expired API key."}
+```
+
+Missing scopes return `403` with `{"error":"Forbidden","message":"Missing required scope."}`. Limit failures return `429` with `{"error":"Rate limit exceeded","message":"Retry later."}`. Clients must never use response differences to infer whether a key ID or hash exists.
+
+### REST, SDK, and n8n clients
+
+All SDKs use the same bearer header. A Python SDK constructor can expose `MediaPlatform(base_url="...", api_key="mp_live_...")`; a JavaScript SDK should accept the same two values. Neither needs a token refresh flow.
+
+```python
+import os
+
+import httpx
+
+api_key = os.environ["MEDIAROUTER_API_KEY"]
+client = httpx.Client(
+    base_url="https://OWNER-SPACE.hf.space",
+    headers={"Authorization": f"Bearer {api_key}"},
+)
+templates = client.get("/v1/templates").raise_for_status().json()
+```
+
+```javascript
+const { MEDIAROUTER_URL: baseUrl, MEDIAROUTER_API_KEY: apiKey } = process.env;
+const response = await fetch(`${baseUrl}/v1/templates`, {
+  headers: { Authorization: `Bearer ${apiKey}` },
+});
+if (!response.ok) throw new Error(`MediaRouter returned ${response.status}`);
+const templates = await response.json();
+```
+
+In n8n, create a reusable **Header Auth** credential with header name `Authorization` and value `Bearer mp_live_...`. Each HTTP Request node then needs only the Base URL/path and that credential. The future official MediaRouter node uses two credential fields—Base URL and API Key—and sends this header automatically.
+
+Security recommendations: give each automation its own least-privilege key; use `mp_test_` outside production; prefer short expirations for temporary agents; rotate on a schedule; revoke on suspected exposure; never put keys in URLs, client logs, Git, `NEXT_PUBLIC_*` variables, or media metadata; and treat frontend validation only as UX because the backend remains authoritative.
+
+## Official SDKs
+
+Publish-ready clients live under [`sdk/`](sdk/):
+
+- [`@mediarouter/media-platform`](sdk/typescript/) — strict TypeScript with ESM and CommonJS builds, typed resources, retries, uploads, downloads, polling, and MCP helpers.
+- [`media-platform`](sdk/python/) — typed Python resource client with stdlib-only runtime dependencies, streamed downloads, multipart uploads, retries, polling, and typed exceptions.
+
+Both clients use the existing `/openapi.json` as their discovery source and keep FFmpeg, Whisper, yt-dlp, and all business logic on the backend. Run the SDK workflow for type checks, mocked transport tests, Python tests, and package builds before publishing a semantic-versioned release.
+
 ## Model Context Protocol (MCP)
 
 MCP is an open protocol that lets an AI client discover and call typed tools, read application resources, and use reusable prompts. This project uses the official Python MCP SDK and exposes the same processing implementation through both interfaces:
@@ -128,7 +259,7 @@ MCP tools ───────────┘                         ├─ FF
                                                └─ faster-whisper
 ```
 
-No FFmpeg command, download implementation, transcription implementation, input parser, logger, or cleanup implementation is duplicated in `app/mcp`. The MCP registry creates a UUID for each tool call and reuses the existing request workspace, validation, structured logging, publication, and expiry cleanup behavior. MCP has no authentication, as requested; use a private Space or an external gateway if access control or quotas are required.
+No FFmpeg command, download implementation, transcription implementation, input parser, logger, cleanup implementation, or authentication implementation is duplicated in `app/mcp`. Mounted and standalone Streamable HTTP pass through the same API-key middleware as REST. The transport-neutral registry also enforces MCP scopes, limits, and audit logging for stdio calls.
 
 ### MCP transports
 
@@ -142,7 +273,7 @@ The trailing slash on `/mcp/` is recommended. Streamable HTTP is stateless and r
 For a local stdio client, run from the project directory:
 
 ```bash
-python -m app.mcp.server
+MCP_STDIO_API_KEY="$MEDIAROUTER_API_KEY" python -m app.mcp.server
 ```
 
 Structured logs are sent to stderr in stdio mode so JSON-RPC messages on stdout are never corrupted. A standalone MCP-only HTTP process is also available for development:
@@ -165,6 +296,7 @@ Claude Desktop and other clients that accept the conventional `mcpServers` JSON
       "args": ["-m", "app.mcp.server"],
       "cwd": "/absolute/path/media-api",
       "env": {
+        "MCP_STDIO_API_KEY": "mp_live_REPLACE_WITH_KEY",
         "TEMP_DIR": "/absolute/path/media-api/temp",
         "OUTPUT_DIR": "/absolute/path/media-api/outputs",
         "WHISPER_MODEL": "small",
@@ -181,7 +313,10 @@ Clients that support remote Streamable HTTP can use:
 {
   "mcpServers": {
     "enterprise-media": {
-      "url": "https://OWNER-SPACE.hf.space/mcp/"
+      "url": "https://OWNER-SPACE.hf.space/mcp/",
+      "headers": {
+        "Authorization": "Bearer mp_live_REPLACE_WITH_KEY"
+      }
     }
   }
 }
@@ -199,7 +334,7 @@ Common client locations and connection choices are:
 | Cline | Open **MCP Servers → Configure** and add the `mcpServers` JSON entry. |
 | Windsurf | Add the same entry in Windsurf MCP settings (`mcp_config.json`). |
 
-Client configuration keys can vary between releases; select **Streamable HTTP**, not legacy SSE, when a client asks for the transport. No authorization header is needed.
+Client configuration keys can vary between releases; select **Streamable HTTP**, not legacy SSE, and configure an `Authorization: Bearer <api_key>` header. A client without custom-header support cannot connect to a protected remote MCP endpoint; use its stdio mode with `MCP_STDIO_API_KEY` instead.
 
 For a Docker-based stdio client, override the image command and keep stdin open:
 
@@ -207,6 +342,7 @@ For a Docker-based stdio client, override the image command and keep stdin open:
 docker run --rm -i \
   -v "$PWD/temp:/app/temp" \
   -v "$PWD/outputs:/app/outputs" \
+  -e MCP_STDIO_API_KEY="$MEDIAROUTER_API_KEY" \
   media-api python -m app.mcp.server
 ```
 
@@ -612,14 +748,15 @@ Keep old YAML documents when introducing a new version so existing automations r
 List and inspect:
 
 ```bash
-curl http://localhost:7860/v1/templates
-curl http://localhost:7860/v1/templates/instagram_reel@1
+curl -H "Authorization: Bearer $MEDIAROUTER_API_KEY" http://localhost:7860/v1/templates
+curl -H "Authorization: Bearer $MEDIAROUTER_API_KEY" http://localhost:7860/v1/templates/instagram_reel@1
 ```
 
 Run with a direct or yt-dlp-supported URL:
 
 ```bash
 curl -X POST http://localhost:7860/v1/templates/run \
+  -H "Authorization: Bearer $MEDIAROUTER_API_KEY" \
   -H 'Content-Type: application/json' \
   -d '{
     "template":"instagram_reel@latest",
@@ -632,6 +769,7 @@ Run with multipart media; `parameters` is a JSON form field:
 
 ```bash
 curl -X POST http://localhost:7860/v1/templates/run \
+  -H "Authorization: Bearer $MEDIAROUTER_API_KEY" \
   -F 'file=@input.mp4' \
   -F 'template=youtube_shorts@1' \
   -F 'parameters={"crf":23,"max_duration":60}'
@@ -717,6 +855,7 @@ All media-processing and probe endpoints use the same resolver. Operation parame
 
 ```bash
 curl -X POST http://localhost:7860/v1/video/compress \
+  -H "Authorization: Bearer $MEDIAROUTER_API_KEY" \
   -F 'file=@input.mp4' \
   -F 'crf=28' \
   -F 'preset=veryfast'
@@ -726,6 +865,7 @@ Multiple-input operations accept repeated or differently named file fields; ever
 
 ```bash
 curl -X POST http://localhost:7860/v1/video/watermark \
+  -H "Authorization: Bearer $MEDIAROUTER_API_KEY" \
   -F 'video=@input.mp4' \
   -F 'watermark=@logo.png' \
   -F 'position=bottom-right' \
@@ -736,6 +876,7 @@ curl -X POST http://localhost:7860/v1/video/watermark \
 
 ```bash
 curl -X POST http://localhost:7860/v1/video/resize \
+  -H "Authorization: Bearer $MEDIAROUTER_API_KEY" \
   -H 'Content-Type: application/json' \
   -d '{"url":"https://cdn.example.com/video.mp4","width":1280,"height":720,"fit":"contain"}'
 ```
@@ -784,6 +925,7 @@ The resolver recognizes n8n properties named `binary.data`, `binary.file`, `bina
 
 ```bash
 curl -X POST 'http://localhost:7860/v1/probe' \
+  -H "Authorization: Bearer $MEDIAROUTER_API_KEY" \
   -H 'Content-Type: application/octet-stream' \
   -H 'X-Filename: input.mp4' \
   --data-binary '@input.mp4'
@@ -793,11 +935,13 @@ Raw and multipart uploads are written to disk in 1 MiB chunks. Downloads are als
 
 ## Endpoints
 
-All processing routes use `POST`. Health and output downloads use `GET`.
+All protected examples below assume `MEDIAROUTER_API_KEY` is set and include `Authorization: Bearer $MEDIAROUTER_API_KEY`. Processing routes use `POST`. Health and output downloads use `GET`.
 
 | Group | Endpoints |
 |---|---|
-| System | `/health`, `/v1/probe`, `/v1/media/{request_id}/{filename}` |
+| Public | `/`, `/health`, `/version`, `/docs`, `/openapi.json`, `/redoc` |
+| Authentication | `/v1/auth/context`, `/v1/api-keys/*`, `/v1/audit-logs` |
+| System | `/v1/probe`, `/v1/media/{request_id}/{filename}` |
 | Video basics | `/v1/video/compress`, `resize`, `crop`, `trim`, `rotate`, `reverse`, `convert`, `merge`, `concat` |
 | Video composition | `/v1/video/overlay`, `watermark`, `replace-audio`, `subtitles/burn`, `subtitles/soft` |
 | Video outputs | `/v1/video/frames`, `gif`, `thumbnail`, `remove-audio`, `mute` |
@@ -815,7 +959,9 @@ Common video parameters include `format`, `width`, `height`, `fit`, `crf`, `pres
 `POST /v1/probe` returns normalized duration, resolution, FPS, bitrate, primary codec, video/audio/subtitle stream summaries, rotation, container, creation date, size, and tags. The same probe data is included under `metadata.inputs` for processing responses.
 
 ```bash
-curl -X POST http://localhost:7860/v1/probe -F 'file=@input.mp4'
+curl -X POST http://localhost:7860/v1/probe \
+  -H "Authorization: Bearer $MEDIAROUTER_API_KEY" \
+  -F 'file=@input.mp4'
 ```
 
 ### yt-dlp
@@ -824,12 +970,14 @@ curl -X POST http://localhost:7860/v1/probe -F 'file=@input.mp4'
 
 ```bash
 curl -X POST http://localhost:7860/v1/ytdlp/download \
+  -H "Authorization: Bearer $MEDIAROUTER_API_KEY" \
   -H 'Content-Type: application/json' \
   -d '{"url":"https://www.youtube.com/watch?v=VIDEO_ID","mode":"audio","audio_format":"mp3"}'
 ```
 
 ```bash
 curl -X POST http://localhost:7860/v1/ytdlp/download \
+  -H "Authorization: Bearer $MEDIAROUTER_API_KEY" \
   -H 'Content-Type: application/json' \
   -d '{"url":"https://vimeo.com/VIDEO_ID","mode":"metadata"}'
 ```
@@ -840,6 +988,7 @@ Models: `tiny`, `base`, `small`, `medium`, and `large-v3`. The default is `small
 
 ```bash
 curl -X POST http://localhost:7860/v1/whisper/transcribe \
+  -H "Authorization: Bearer $MEDIAROUTER_API_KEY" \
   -F 'file=@meeting.mp3' \
   -F 'model=small' \
   -F 'task=transcribe' \
@@ -849,6 +998,7 @@ curl -X POST http://localhost:7860/v1/whisper/transcribe \
 
 ```bash
 curl -X POST http://localhost:7860/v1/whisper/subtitles \
+  -H "Authorization: Bearer $MEDIAROUTER_API_KEY" \
   -F 'file=@interview.mp4' \
   -F 'task=translate' \
   -F 'output_format=vtt'
@@ -894,6 +1044,7 @@ In an **HTTP Request** node:
 
 - Method: `POST`
 - URL: `https://<space>.hf.space/v1/video/compress`
+- Authentication: reusable Header Auth credential, `Authorization = Bearer <api_key>`
 - Send Body: on
 - Body Content Type: `Form-Data`
 - Add a **n8n Binary File** parameter named `file`, selecting the incoming binary property (usually `data`)
@@ -953,10 +1104,24 @@ For large files, prefer n8n's multipart binary option or raw binary body; JSON B
 | `BASE_URL` | empty | Optional public origin for absolute download URLs |
 | `FFMPEG_BINARY` | `ffmpeg` | FFmpeg executable name/path |
 | `FFPROBE_BINARY` | `ffprobe` | FFprobe executable name/path |
+| `AUTH_ENABLED` | `true` | Fail-closed API-key enforcement; disable only for isolated development/tests |
+| `DATABASE_URL` | `sqlite+aiosqlite:///./data/mediarouter.db` | Hash-only keys, audit records, and rate aggregates |
+| `AUTH_ROLE_SCOPES` | `{}` | JSON custom role-to-scope mappings |
+| `AUTH_BOOTSTRAP_KEY_HASH` | empty | SHA-256 hash for first-start administrator |
+| `AUTH_BOOTSTRAP_KEY_PREFIX` | empty | Safe display prefix matching the bootstrap key |
+| `AUTH_BOOTSTRAP_KEY_NAME` | `Bootstrap Administrator` | Bootstrap record label |
+| `AUTH_BOOTSTRAP_ENVIRONMENT` | `live` | `live` or `test`; must match the prefix |
+| `AUTH_LAST_USED_UPDATE_SECONDS` | `60` | Throttle for database last-used writes |
+| `AUTH_DEFAULT_REQUESTS_PER_MINUTE` | `100` | Default per-key request window |
+| `AUTH_DEFAULT_CONCURRENT_JOBS` | `10` | Default processing request concurrency |
+| `AUTH_DEFAULT_UPLOADS_PER_HOUR` | `20` | Default upload request window |
+| `AUTH_DEFAULT_PROCESSING_BYTES_PER_DAY` | `107374182400` | Default daily uploaded processing bytes |
+| `AUTH_TRUST_PROXY_HEADERS` | `true` | Use first `X-Forwarded-For` address for audits behind HF/Vercel |
+| `MCP_STDIO_API_KEY` | empty | Existing API key required by authenticated standalone stdio MCP |
 
 ## Logging and safety
 
-Logs are one JSON object per line. Request completion and operation records include UUID, operation, input/output bytes, wall time, CPU percentage, RSS memory, status, and errors. FFmpeg records include the exact argument array, bounded stdout/stderr, exit status, and duration.
+Logs are one JSON object per line. Request completion and operation records include UUID, API key ID/name (never the secret), operation, input/output bytes, wall time, CPU percentage, RSS memory, status, and errors. FFmpeg records include the exact argument array, bounded stdout/stderr, exit status, and duration. Structured audit rows preserve authenticated request metadata independently of application logs.
 
 Safety controls include filename normalization, path containment checks, extension/MIME/size validation, duration and pixel limits, HTTP scheme and redirect validation, SSRF address filtering, bounded concurrency, bounded output capture, and subprocess arrays with `shell=False` semantics. User filenames never select output paths.
 
@@ -969,7 +1134,7 @@ ruff check app tests main.py
 black --check app tests main.py
 ```
 
-The suite covers health, Base64/n8n resolution, streamed URL download, real FFprobe metadata, a real FFmpeg conversion, lazy mocked Whisper transcription, MCP registration, YAML loading and versioning, parameter substitution, template execution, cleanup behavior, and safe error envelopes. Binary integration tests skip only when the respective system executable is absent.
+The suite covers health, Base64/n8n resolution, streamed URL download, real FFprobe metadata, a real FFmpeg conversion, lazy mocked Whisper transcription, MCP registration and authorization, key entropy/hash-only storage, invalid/expired/revoked keys, scopes, rotation grace, disable/enable, request and concurrent limits, audit persistence, middleware contracts, YAML loading and versioning, parameter substitution, template execution, cleanup behavior, and safe error envelopes. Binary integration tests skip only when the respective system executable is absent.
 
 ## Extending the API
 
@@ -991,5 +1156,7 @@ This contract keeps new operations independent of multipart, URLs, Base64, n8n,
 - **Upload receives 413/422:** increase `MAX_UPLOAD_SIZE` only after checking Space disk and RAM. Prefer multipart/raw streaming over Base64.
 - **ImageMagick policy error:** media operations use FFmpeg for image transforms; ImageMagick is installed for extension use but is not required by the built-in image routes.
 - **No audio after merging videos:** if any input lacks an audio stream, the merge deliberately emits video-only output instead of failing the whole request. Add silent audio before merging if a continuous audio track is required.
-- **MCP client cannot connect:** use the trailing-slash URL `/mcp/`, select Streamable HTTP rather than legacy SSE, and verify `/health` first. Stdio clients must launch the command with the project directory as their working directory.
+- **Every protected request returns 401 after first deployment:** the database has no administrator. Generate a bootstrap key, configure its matching hash/prefix/environment secrets, and restart once. Ensure persistent storage contains the expected database.
+- **Hugging Face restarted and keys disappeared:** the default relative SQLite file was on ephemeral storage. Attach persistent storage and use `sqlite+aiosqlite:////data/mediarouter.db`, then create/rotate keys again.
+- **MCP client cannot connect:** use the trailing-slash URL `/mcp/`, select Streamable HTTP rather than legacy SSE, include the bearer header, and verify `/health` first. Stdio clients must set `MCP_STDIO_API_KEY` and launch with the project directory as their working directory.
 - **MCP output URL is relative:** set `BASE_URL` to the public Space origin. A managed `output_file` can also be passed directly into a later MCP call as `temp_path` before cleanup expires.

app/__init__.py

@@ -1,3 +1,3 @@
-"""Enterprise Media Processing API package."""
+"""MediaRouter media automation API package."""
 
 __version__ = "1.0.0"

app/api/api_keys.py

@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, HTTPException, Query, Request, Response, status
+
+from app.security.errors import APIKeyConflictError, APIKeyNotFoundError
+from app.security.schemas import (
+    APIKeyCreate,
+    APIKeyCreated,
+    APIKeyList,
+    APIKeyPatch,
+    APIKeyRotate,
+    APIKeyView,
+    AuthContextView,
+    AuditLogView,
+)
+from app.security.scopes import ALL_SCOPES
+
+router = APIRouter(prefix="/v1", tags=["authentication"])
+
+
+def _view(record: object) -> APIKeyView:
+    return APIKeyView.model_validate(record)
+
+
+def _not_found() -> HTTPException:
+    return HTTPException(status_code=404, detail="API key was not found")
+
+
+@router.get("/auth/context", response_model=AuthContextView)
+async def current_auth_context(request: Request) -> AuthContextView:
+    """Validate a key and return only its safe, non-secret authorization context."""
+    context = request.state.auth
+    return AuthContextView(
+        id=context.api_key_id,
+        name=context.key_name,
+        key_prefix=context.key_prefix,
+        environment=context.environment,
+        role=context.role,
+        scopes=sorted(context.scopes),
+        expires_at=context.expires_at,
+    )
+
+
+@router.get("/api-keys/capabilities")
+async def api_key_capabilities(request: Request) -> dict[str, object]:
+    container = request.app.state.container
+    return {
+        "scopes": sorted(ALL_SCOPES),
+        "roles": {
+            role: sorted(scopes)
+            for role, scopes in container.api_keys.roles.items()
+        },
+        "defaults": {
+            "requests_per_minute": container.settings.auth_default_requests_per_minute,
+            "concurrent_jobs": container.settings.auth_default_concurrent_jobs,
+            "uploads_per_hour": container.settings.auth_default_uploads_per_hour,
+            "processing_bytes_per_day": (
+                container.settings.auth_default_processing_bytes_per_day
+            ),
+        },
+    }
+
+
+@router.get("/api-keys", response_model=APIKeyList)
+async def list_api_keys(
+    request: Request,
+    offset: int = Query(default=0, ge=0),
+    limit: int = Query(default=100, ge=1, le=500),
+) -> APIKeyList:
+    records, total = await request.app.state.container.api_keys.list(
+        offset=offset, limit=limit
+    )
+    return APIKeyList(items=[_view(record) for record in records], total=total)
+
+
+@router.post(
+    "/api-keys", response_model=APIKeyCreated, status_code=status.HTTP_201_CREATED
+)
+async def create_api_key(request: Request, payload: APIKeyCreate) -> APIKeyCreated:
+    context = request.state.auth
+    try:
+        record, secret = await request.app.state.container.api_keys.create(
+            payload, created_by=context.api_key_id
+        )
+    except APIKeyConflictError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    return APIKeyCreated(**_view(record).model_dump(), api_key=secret)
+
+
+@router.get("/api-keys/{key_id}", response_model=APIKeyView)
+async def get_api_key(request: Request, key_id: str) -> APIKeyView:
+    try:
+        return _view(await request.app.state.container.api_keys.get(key_id))
+    except APIKeyNotFoundError as exc:
+        raise _not_found() from exc
+
+
+@router.patch("/api-keys/{key_id}", response_model=APIKeyView)
+async def patch_api_key(
+    request: Request, key_id: str, payload: APIKeyPatch
+) -> APIKeyView:
+    try:
+        return _view(await request.app.state.container.api_keys.patch(key_id, payload))
+    except APIKeyNotFoundError as exc:
+        raise _not_found() from exc
+    except APIKeyConflictError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+
+
+@router.delete("/api-keys/{key_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def revoke_api_key(request: Request, key_id: str) -> Response:
+    try:
+        await request.app.state.container.api_keys.set_status(key_id, "revoked")
+    except APIKeyNotFoundError as exc:
+        raise _not_found() from exc
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+
+@router.post(
+    "/api-keys/{key_id}/rotate",
+    response_model=APIKeyCreated,
+    status_code=status.HTTP_201_CREATED,
+)
+async def rotate_api_key(
+    request: Request, key_id: str, payload: APIKeyRotate
+) -> APIKeyCreated:
+    try:
+        record, secret = await request.app.state.container.api_keys.rotate(
+            key_id,
+            payload.grace_period_seconds,
+            created_by=request.state.auth.api_key_id,
+        )
+    except APIKeyNotFoundError as exc:
+        raise _not_found() from exc
+    except APIKeyConflictError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    return APIKeyCreated(**_view(record).model_dump(), api_key=secret)
+
+
+@router.post("/api-keys/{key_id}/disable", response_model=APIKeyView)
+async def disable_api_key(request: Request, key_id: str) -> APIKeyView:
+    try:
+        return _view(
+            await request.app.state.container.api_keys.set_status(key_id, "disabled")
+        )
+    except APIKeyNotFoundError as exc:
+        raise _not_found() from exc
+    except APIKeyConflictError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+
+
+@router.post("/api-keys/{key_id}/enable", response_model=APIKeyView)
+async def enable_api_key(request: Request, key_id: str) -> APIKeyView:
+    try:
+        return _view(
+            await request.app.state.container.api_keys.set_status(key_id, "active")
+        )
+    except APIKeyNotFoundError as exc:
+        raise _not_found() from exc
+    except APIKeyConflictError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+
+
+@router.get("/audit-logs", response_model=list[AuditLogView])
+async def list_audit_logs(
+    request: Request,
+    offset: int = Query(default=0, ge=0),
+    limit: int = Query(default=100, ge=1, le=500),
+) -> list[AuditLogView]:
+    records = await request.app.state.container.audit.list(offset=offset, limit=limit)
+    return [AuditLogView.model_validate(record) for record in records]

app/container.py

@@ -12,6 +12,10 @@ from app.services.media_service import MediaProcessor
 from app.services.validator import MediaValidator
 from app.services.whisper_service import WhisperService
 from app.services.ytdlp_service import YTDLPService
+from app.security.audit import AuditService
+from app.security.database import SecurityDatabase
+from app.security.rate_limit import APIKeyRateLimiter
+from app.security.service import APIKeyService
 from app.templates.executor import OperationExecutor, TemplateExecutor
 from app.templates.loader import TemplateLoader
 from app.templates.registry import TemplateRegistry
@@ -32,9 +36,17 @@ class Container:
     processor: MediaProcessor
     template_registry: TemplateRegistry
     template_executor: TemplateExecutor
+    security_database: SecurityDatabase
+    api_keys: APIKeyService
+    rate_limiter: APIKeyRateLimiter
+    audit: AuditService
 
 
 def build_container(settings: Settings) -> Container:
+    security_database = SecurityDatabase(settings.database_url)
+    api_keys = APIKeyService(security_database, settings)
+    rate_limiter = APIKeyRateLimiter(security_database)
+    audit = AuditService(security_database)
     cleanup = CleanupService(settings)
     validator = MediaValidator(settings)
     downloader = Downloader(settings, validator)
@@ -53,16 +65,20 @@ def build_container(settings: Settings) -> Container:
     )
     template_executor = TemplateExecutor(template_registry, operation_executor, processor)
     return Container(
-        settings,
-        cleanup,
-        validator,
-        downloader,
-        ytdlp,
-        ffmpeg,
-        ffprobe,
-        whisper,
-        resolver,
-        processor,
-        template_registry,
-        template_executor,
+        settings=settings,
+        cleanup=cleanup,
+        validator=validator,
+        downloader=downloader,
+        ytdlp=ytdlp,
+        ffmpeg=ffmpeg,
+        ffprobe=ffprobe,
+        whisper=whisper,
+        resolver=resolver,
+        processor=processor,
+        template_registry=template_registry,
+        template_executor=template_executor,
+        security_database=security_database,
+        api_keys=api_keys,
+        rate_limiter=rate_limiter,
+        audit=audit,
     )

app/core/config.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from functools import lru_cache
 from pathlib import Path
 
-from pydantic import Field, field_validator
+from pydantic import Field, SecretStr, field_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 DEFAULT_TEMPLATE_DIR = Path(__file__).resolve().parents[1] / "templates" / "categories"
@@ -16,7 +16,7 @@ class Settings(BaseSettings):
         env_file=".env", env_file_encoding="utf-8", case_sensitive=False, extra="ignore"
     )
 
-    app_name: str = "Enterprise Media Processing API"
+    app_name: str = "MediaRouter"
     app_version: str = "1.0.0"
     host: str = "0.0.0.0"
     port: int = 7860
@@ -36,6 +36,22 @@ class Settings(BaseSettings):
     base_url: str = ""
     ffmpeg_binary: str = "ffmpeg"
     ffprobe_binary: str = "ffprobe"
+    auth_enabled: bool = True
+    database_url: str = "sqlite+aiosqlite:///./data/mediarouter.db"
+    auth_role_scopes: dict[str, list[str]] = Field(default_factory=dict)
+    auth_bootstrap_key_hash: str = ""
+    auth_bootstrap_key_prefix: str = ""
+    auth_bootstrap_key_name: str = "Bootstrap Administrator"
+    auth_bootstrap_environment: str = "live"
+    auth_last_used_update_seconds: int = Field(default=60, ge=0, le=3600)
+    auth_default_requests_per_minute: int = Field(default=100, ge=1, le=1_000_000)
+    auth_default_concurrent_jobs: int = Field(default=10, ge=1, le=10_000)
+    auth_default_uploads_per_hour: int = Field(default=20, ge=1, le=1_000_000)
+    auth_default_processing_bytes_per_day: int = Field(
+        default=107_374_182_400, ge=1_048_576
+    )
+    auth_trust_proxy_headers: bool = True
+    mcp_stdio_api_key: SecretStr | None = None
 
     @field_validator("whisper_model")
     @classmethod
@@ -57,6 +73,23 @@ class Settings(BaseSettings):
     def ensure_directories(self) -> None:
         self.temp_dir.mkdir(parents=True, exist_ok=True)
         self.output_dir.mkdir(parents=True, exist_ok=True)
+        sqlite_prefixes = ("sqlite+aiosqlite:///", "sqlite:///")
+        for prefix in sqlite_prefixes:
+            if self.database_url.startswith(prefix):
+                database_path = self.database_url.removeprefix(prefix)
+                if database_path and database_path != ":memory:":
+                    Path(database_path).expanduser().resolve().parent.mkdir(
+                        parents=True, exist_ok=True
+                    )
+                break
+
+    @field_validator("auth_bootstrap_environment")
+    @classmethod
+    def validate_auth_environment(cls, value: str) -> str:
+        normalized = value.strip().lower()
+        if normalized not in {"live", "test"}:
+            raise ValueError("AUTH_BOOTSTRAP_ENVIRONMENT must be live or test")
+        return normalized
 
 
 @lru_cache

app/mcp/registry.py

@@ -20,6 +20,8 @@ from app.core.logger import get_logger, request_id_context
 from app.core.response import SuccessResponse
 from app.models.media import ResolvedRequest
 from app.operations.common import AUDIO_FORMATS, IMAGE_FORMATS, VIDEO_FORMATS
+from app.security.context import AuthContext, auth_context, http_auth_applied
+from app.security.errors import RateLimitError
 from app.services.media_service import Operation
 
 logger = get_logger(__name__)
@@ -232,7 +234,13 @@ class MCPRegistry:
 
         return await self._execute("run_template", action)
 
-    async def run_metadata_tool(self, tool_name: str, action: MetadataAction) -> dict[str, Any]:
+    async def run_metadata_tool(
+        self,
+        tool_name: str,
+        action: MetadataAction,
+        *,
+        required_scope: str = "mcp:execute",
+    ) -> dict[str, Any]:
         """Run a non-media utility with the same logging and error contract."""
 
         async def wrapped(request_id: str) -> SuccessResponse:
@@ -244,7 +252,7 @@ class MCPRegistry:
                 metadata=metadata,
             )
 
-        return await self._execute(tool_name, wrapped)
+        return await self._execute(tool_name, wrapped, required_scope=required_scope)
 
     async def health_data(self) -> dict[str, Any]:
         """Return dependency health without loading the Whisper model."""
@@ -347,7 +355,9 @@ class MCPRegistry:
 
     async def safe_resource(self, name: str, action: MetadataAction) -> dict[str, Any]:
         """Return resource data without exposing raw exceptions."""
-        response = await self.run_metadata_tool(f"resource.{name}", action)
+        response = await self.run_metadata_tool(
+            f"resource.{name}", action, required_scope="mcp:read"
+        )
         if response["success"]:
             return {"success": True, **response["metadata"]}
         return {
@@ -374,13 +384,64 @@ class MCPRegistry:
             ytdlp_options=ytdlp_options,
         )
 
-    async def _execute(self, tool_name: str, action: Action) -> dict[str, Any]:
+    async def _execute(
+        self,
+        tool_name: str,
+        action: Action,
+        *,
+        required_scope: str = "mcp:execute",
+    ) -> dict[str, Any]:
         request_id = str(uuid4())
-        token = request_id_context.set(request_id)
         started = time.monotonic()
+        context = auth_context.get()
+        via_http = http_auth_applied.get()
+        if self.container.settings.auth_enabled and context is None:
+            return self._tool_failure(
+                request_id,
+                started,
+                "UNAUTHORIZED",
+                "Invalid or expired API key.",
+                None,
+            )
+        lease = None
+        if context is not None and not via_http:
+            try:
+                lease = await self.container.rate_limiter.acquire(
+                    context,
+                    is_job=required_scope == "mcp:execute",
+                    is_upload=False,
+                    uploaded_bytes=0,
+                )
+            except RateLimitError as exc:
+                await self._audit_mcp(context, request_id, tool_name, 429, started)
+                return self._tool_failure(
+                    request_id,
+                    started,
+                    "RATE_LIMIT_EXCEEDED",
+                    "Retry later.",
+                    {"retry_after": exc.retry_after},
+                )
+        if context is not None and not context.allows(required_scope):
+            if lease is not None:
+                await lease.release()
+            result = self._tool_failure(
+                request_id,
+                started,
+                "FORBIDDEN",
+                "Missing required scope.",
+                None,
+            )
+            if not via_http:
+                await self._audit_mcp(context, request_id, tool_name, 403, started)
+            return result
+        token = request_id_context.set(request_id)
         cpu_started = time.process_time()
+        response_code = 500
         try:
+            if context is not None and not via_http:
+                await self.container.api_keys.mark_used(context)
             response = await action(request_id)
+            response_code = 200
             result = self._tool_success(response)
             logger.info(
                 "MCP tool completed",
@@ -394,6 +455,7 @@ class MCPRegistry:
             )
             return result
         except MediaAPIError as exc:
+            response_code = exc.status_code
             logger.warning(
                 "MCP tool failed",
                 extra={
@@ -424,6 +486,42 @@ class MCPRegistry:
                 )
             finally:
                 request_id_context.reset(token)
+                if lease is not None:
+                    await lease.release()
+                if context is not None and not via_http:
+                    await self._audit_mcp(
+                        context, request_id, tool_name, response_code, started
+                    )
+
+    async def _audit_mcp(
+        self,
+        context: AuthContext,
+        request_id: str,
+        tool_name: str,
+        response_code: int,
+        started: float,
+    ) -> None:
+        data = {
+            "request_id": request_id,
+            "api_key_id": context.api_key_id,
+            "key_name": context.key_name,
+            "ip_address": None,
+            "user_agent": "mcp-stdio",
+            "endpoint": f"mcp://tools/{tool_name}",
+            "http_method": "CALL",
+            "response_code": response_code,
+            "processing_time_ms": max(0, round((time.monotonic() - started) * 1000)),
+            "bytes_uploaded": 0,
+            "bytes_downloaded": 0,
+        }
+        try:
+            await self.container.audit.record(**data)
+        except Exception:
+            logger.exception(
+                "MCP authentication audit persistence failed",
+                extra={"api_key_id": context.api_key_id, "tool": tool_name},
+            )
+        logger.info("authenticated MCP call", extra=data)
 
     def _tool_success(self, response: SuccessResponse) -> dict[str, Any]:
         output_file: str | None = None

app/mcp/server.py

@@ -5,7 +5,10 @@ import asyncio
 import sys
 from typing import Any, Literal, cast
 
+import uvicorn
 from mcp.server.fastmcp import FastMCP
+from starlette.applications import Starlette
+from starlette.routing import Mount
 
 from app.container import Container, build_container
 from app.core.config import get_settings
@@ -21,6 +24,8 @@ from app.mcp.tools.templates import register_template_tools
 from app.mcp.tools.video import register_video_tools
 from app.mcp.tools.whisper import register_whisper_tools
 from app.mcp.tools.ytdlp import register_ytdlp_tools
+from app.security.context import auth_context
+from app.security.middleware import APIKeyAuthenticationMiddleware
 from app.workers.cleanup_worker import CleanupWorker
 
 MCPLogLevel = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
@@ -30,7 +35,7 @@ def create_mcp_server(container: Container) -> FastMCP[Any]:
     """Create a fully registered MCP server over an existing service container."""
     settings = container.settings
     server: FastMCP[Any] = FastMCP(
-        name="Enterprise Media Processing API",
+        name=settings.app_name,
         instructions=(
             "Use the registered tools for safe media processing. All media inputs accept URL, Base64, "
             "n8n binary objects, or managed temp_path values. Read media://operations for capabilities."
@@ -57,20 +62,61 @@ def create_mcp_server(container: Container) -> FastMCP[Any]:
 
 
 async def run_server(transport: Literal["stdio", "streamable-http"] = "stdio") -> None:
-    """Run MCP as a standalone stdio or Streamable HTTP server."""
+    """Run standalone MCP with the same database, keys, scopes, and limits as REST."""
     configure_logging(stream=sys.stderr if transport == "stdio" else None)
     settings = get_settings()
     container = build_container(settings)
     worker = CleanupWorker(container.cleanup, settings.cleanup_interval_seconds)
     server = create_mcp_server(container)
+    await container.security_database.initialize()
+    await container.api_keys.ensure_bootstrap_admin()
     await worker.start()
     try:
         if transport == "stdio":
-            await server.run_stdio_async()
+            context_token = None
+            if settings.auth_enabled:
+                configured_key = (
+                    settings.mcp_stdio_api_key.get_secret_value()
+                    if settings.mcp_stdio_api_key is not None
+                    else ""
+                )
+                if not configured_key:
+                    raise RuntimeError(
+                        "MCP_STDIO_API_KEY is required when AUTH_ENABLED=true"
+                    )
+                context = await container.api_keys.authenticate(configured_key)
+                if not context.allows("mcp:read"):
+                    raise RuntimeError("MCP_STDIO_API_KEY is missing the mcp:read scope")
+                await container.api_keys.mark_used(context)
+                context_token = auth_context.set(context)
+            try:
+                await server.run_stdio_async()
+            finally:
+                if context_token is not None:
+                    auth_context.reset(context_token)
         else:
-            await server.run_streamable_http_async()
+            mcp_application = server.streamable_http_app()
+            application = Starlette(routes=[Mount("/mcp", app=mcp_application)])
+            application.state.container = container
+            application.state.mcp_server = server
+            application.add_middleware(
+                APIKeyAuthenticationMiddleware,
+                settings=settings,
+                api_keys=container.api_keys,
+                rate_limiter=container.rate_limiter,
+                audit=container.audit,
+            )
+            config = uvicorn.Config(
+                application,
+                host=settings.host,
+                port=settings.port,
+                log_level=settings.log_level.lower(),
+            )
+            async with server.session_manager.run():
+                await uvicorn.Server(config).serve()
     finally:
         await worker.stop()
+        await container.security_database.close()
 
 
 def main() -> None:

app/security/__init__.py

@@ -0,0 +1,6 @@
+"""Stateless API-key authentication, authorization, rate limiting, and auditing."""
+
+from app.security.context import AuthContext
+from app.security.scopes import ALL_SCOPES, DEFAULT_ROLE_SCOPES
+
+__all__ = ["ALL_SCOPES", "DEFAULT_ROLE_SCOPES", "AuthContext"]

app/security/audit.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from sqlalchemy import select
+
+from app.security.database import SecurityDatabase
+from app.security.models import AuditLog
+
+
+class AuditService:
+    def __init__(self, database: SecurityDatabase) -> None:
+        self.database = database
+
+    async def record(
+        self,
+        *,
+        request_id: str,
+        api_key_id: str,
+        key_name: str,
+        ip_address: str | None,
+        user_agent: str | None,
+        endpoint: str,
+        http_method: str,
+        response_code: int,
+        processing_time_ms: int,
+        bytes_uploaded: int,
+        bytes_downloaded: int,
+    ) -> None:
+        async with self.database.session() as session:
+            session.add(
+                AuditLog(
+                    request_id=request_id,
+                    api_key_id=api_key_id,
+                    key_name=key_name,
+                    ip_address=ip_address,
+                    user_agent=user_agent,
+                    endpoint=endpoint,
+                    http_method=http_method,
+                    response_code=response_code,
+                    processing_time_ms=processing_time_ms,
+                    bytes_uploaded=bytes_uploaded,
+                    bytes_downloaded=bytes_downloaded,
+                )
+            )
+            await session.commit()
+
+    async def list(self, *, offset: int = 0, limit: int = 100) -> list[AuditLog]:
+        async with self.database.session() as session:
+            return list(
+                (
+                    await session.scalars(
+                        select(AuditLog)
+                        .order_by(AuditLog.created_at.desc())
+                        .offset(offset)
+                        .limit(limit)
+                    )
+                ).all()
+            )

app/security/cli.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+
+from app.core.config import get_settings
+from app.security.database import SecurityDatabase
+from app.security.schemas import APIKeyCreate, APIKeyView
+from app.security.service import APIKeyService
+
+
+async def _create(arguments: argparse.Namespace) -> None:
+    settings = get_settings()
+    database = SecurityDatabase(settings.database_url)
+    service = APIKeyService(database, settings)
+    await database.initialize()
+    try:
+        record, secret = await service.create(
+            APIKeyCreate(
+                name=arguments.name,
+                environment=arguments.environment,
+                role=arguments.role,
+                expires_in_seconds=arguments.expires_in,
+            ),
+            created_by="security-cli",
+        )
+        payload = APIKeyView.model_validate(record).model_dump(mode="json")
+        payload["api_key"] = secret
+        print(json.dumps(payload, indent=2))
+    finally:
+        await database.close()
+
+
+def _bootstrap(arguments: argparse.Namespace) -> None:
+    material = APIKeyService.generate_material(arguments.environment)
+    print("Store the API key in your password manager. It will not be shown again.\n")
+    print(f"API_KEY={material.api_key}")
+    print(f"AUTH_BOOTSTRAP_KEY_HASH={material.key_hash}")
+    print(f"AUTH_BOOTSTRAP_KEY_PREFIX={material.key_prefix}")
+    print(f"AUTH_BOOTSTRAP_ENVIRONMENT={material.environment}")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="MediaRouter API-key administration")
+    commands = parser.add_subparsers(dest="command", required=True)
+    bootstrap = commands.add_parser(
+        "generate-bootstrap", help="Generate a key and hash-only bootstrap settings"
+    )
+    bootstrap.add_argument("--environment", choices=("live", "test"), default="live")
+    create = commands.add_parser("create", help="Create a key directly in the configured database")
+    create.add_argument("--name", required=True)
+    create.add_argument("--environment", choices=("live", "test"), default="live")
+    create.add_argument("--role", default="admin")
+    create.add_argument("--expires-in", type=int, default=None)
+    arguments = parser.parse_args()
+    if arguments.command == "generate-bootstrap":
+        _bootstrap(arguments)
+    else:
+        asyncio.run(_create(arguments))
+
+
+if __name__ == "__main__":
+    main()

app/security/context.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import contextvars
+from dataclasses import dataclass
+from datetime import datetime
+
+
+@dataclass(frozen=True, slots=True)
+class AuthContext:
+    api_key_id: str
+    key_name: str
+    key_prefix: str
+    environment: str
+    role: str | None
+    scopes: frozenset[str]
+    requests_per_minute: int
+    concurrent_jobs: int
+    uploads_per_hour: int
+    processing_bytes_per_day: int
+    expires_at: datetime | None
+
+    def allows(self, required_scope: str) -> bool:
+        return "admin" in self.scopes or required_scope in self.scopes
+
+
+auth_context: contextvars.ContextVar[AuthContext | None] = contextvars.ContextVar(
+    "auth_context", default=None
+)
+
+# HTTP authentication already applies request and concurrency limits before the
+# MCP ASGI application runs. Standalone stdio MCP calls use this marker to apply
+# the same limiter exactly once inside the transport-neutral registry.
+http_auth_applied: contextvars.ContextVar[bool] = contextvars.ContextVar(
+    "http_auth_applied", default=False
+)

app/security/database.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+from sqlalchemy import event
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
+
+from app.security.models import Base
+
+
+class SecurityDatabase:
+    """Owns the authentication database engine and short-lived async sessions."""
+
+    def __init__(self, database_url: str) -> None:
+        self.engine: AsyncEngine = create_async_engine(
+            database_url,
+            pool_pre_ping=True,
+        )
+        if database_url.startswith("sqlite"):
+            event.listen(self.engine.sync_engine, "connect", self._configure_sqlite)
+        self.session_factory = async_sessionmaker(
+            self.engine, expire_on_commit=False, class_=AsyncSession
+        )
+
+    @staticmethod
+    def _configure_sqlite(dbapi_connection: object, _record: object) -> None:
+        cursor = dbapi_connection.cursor()  # type: ignore[attr-defined]
+        cursor.execute("PRAGMA foreign_keys=ON")
+        cursor.execute("PRAGMA journal_mode=WAL")
+        cursor.execute("PRAGMA busy_timeout=5000")
+        cursor.close()
+
+    async def initialize(self) -> None:
+        async with self.engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+
+    async def close(self) -> None:
+        await self.engine.dispose()
+
+    @asynccontextmanager
+    async def session(self) -> AsyncIterator[AsyncSession]:
+        async with self.session_factory() as session:
+            yield session

app/security/errors.py

@@ -0,0 +1,24 @@
+class SecurityError(Exception):
+    """Base class for deliberately generic authentication failures."""
+
+
+class UnauthorizedError(SecurityError):
+    pass
+
+
+class ForbiddenError(SecurityError):
+    pass
+
+
+class RateLimitError(SecurityError):
+    def __init__(self, retry_after: int) -> None:
+        super().__init__("Rate limit exceeded")
+        self.retry_after = max(1, retry_after)
+
+
+class APIKeyNotFoundError(SecurityError):
+    pass
+
+
+class APIKeyConflictError(SecurityError):
+    pass

app/security/middleware.py

@@ -0,0 +1,193 @@
+from __future__ import annotations
+
+import time
+from typing import Any
+from uuid import uuid4
+
+from fastapi import Request
+from fastapi.responses import JSONResponse
+from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
+from starlette.responses import Response
+
+from app.core.config import Settings
+from app.core.logger import get_logger
+from app.security.audit import AuditService
+from app.security.context import AuthContext, auth_context, http_auth_applied
+from app.security.errors import ForbiddenError, RateLimitError, UnauthorizedError
+from app.security.policy import ScopePolicy
+from app.security.rate_limit import APIKeyRateLimiter, RateLimitLease
+from app.security.service import APIKeyService
+
+logger = get_logger(__name__)
+
+
+class APIKeyAuthenticationMiddleware(BaseHTTPMiddleware):
+    """Authenticates, authorizes, rate-limits, and audits protected HTTP requests."""
+
+    def __init__(
+        self,
+        app: Any,
+        *,
+        settings: Settings,
+        api_keys: APIKeyService,
+        rate_limiter: APIKeyRateLimiter,
+        audit: AuditService,
+    ) -> None:
+        super().__init__(app)
+        self.settings = settings
+        self.api_keys = api_keys
+        self.rate_limiter = rate_limiter
+        self.audit = audit
+        self.policy = ScopePolicy()
+
+    async def dispatch(
+        self, request: Request, call_next: RequestResponseEndpoint
+    ) -> Response:
+        if not getattr(request.state, "request_id", None):
+            request.state.request_id = str(uuid4())
+        if not self.settings.auth_enabled or self.policy.is_public(request):
+            return await call_next(request)
+        started = time.monotonic()
+        context: AuthContext | None = None
+        lease: RateLimitLease | None = None
+        context_token = None
+        http_auth_token = None
+        response_code = 500
+        bytes_uploaded = self._content_length(request.headers.get("content-length"))
+        bytes_downloaded = 0
+        try:
+            api_key = self._bearer_token(request.headers.get("authorization"))
+            context = await self.api_keys.authenticate(api_key)
+            request.state.auth = context
+            context_token = auth_context.set(context)
+            required_scope = await self.policy.required_scope(request)
+            lease = await self.rate_limiter.acquire(
+                context,
+                is_job=self.policy.is_job(required_scope),
+                is_upload=self.policy.is_upload(request, required_scope),
+                uploaded_bytes=bytes_uploaded,
+            )
+            self.api_keys.authorize(context, required_scope)
+            await self.api_keys.mark_used(context)
+            http_auth_token = http_auth_applied.set(True)
+            response = await call_next(request)
+            response_code = response.status_code
+            bytes_downloaded = self._content_length(response.headers.get("content-length"))
+            response.headers.setdefault("X-Request-ID", request.state.request_id)
+            return response
+        except UnauthorizedError:
+            response_code = 401
+            return self._error(
+                401,
+                "Unauthorized",
+                "Invalid or expired API key.",
+                request,
+                {"WWW-Authenticate": "Bearer"},
+            )
+        except ForbiddenError:
+            response_code = 403
+            response = self._error(
+                403, "Forbidden", "Missing required scope.", request
+            )
+            bytes_downloaded = len(response.body)
+            return response
+        except RateLimitError as exc:
+            response_code = 429
+            response = self._error(
+                429,
+                "Rate limit exceeded",
+                "Retry later.",
+                request,
+                {"Retry-After": str(exc.retry_after)},
+            )
+            bytes_downloaded = len(response.body)
+            return response
+        finally:
+            if lease is not None:
+                await lease.release()
+            if http_auth_token is not None:
+                http_auth_applied.reset(http_auth_token)
+            if context_token is not None:
+                auth_context.reset(context_token)
+            if context is not None:
+                elapsed_ms = max(0, round((time.monotonic() - started) * 1000))
+                await self._audit_request(
+                    request,
+                    context,
+                    response_code,
+                    elapsed_ms,
+                    bytes_uploaded,
+                    bytes_downloaded,
+                )
+
+    @staticmethod
+    def _bearer_token(header: str | None) -> str:
+        if not header:
+            raise UnauthorizedError
+        parts = header.strip().split()
+        if len(parts) != 2 or parts[0].casefold() != "bearer" or not parts[1]:
+            raise UnauthorizedError
+        return parts[1]
+
+    @staticmethod
+    def _content_length(value: str | None) -> int:
+        try:
+            return max(0, int(value or 0))
+        except ValueError:
+            return 0
+
+    @staticmethod
+    def _error(
+        status_code: int,
+        error: str,
+        message: str,
+        request: Request,
+        headers: dict[str, str] | None = None,
+    ) -> JSONResponse:
+        response_headers = dict(headers or {})
+        request_id = getattr(request.state, "request_id", None)
+        if request_id:
+            response_headers["X-Request-ID"] = request_id
+        return JSONResponse(
+            {"error": error, "message": message},
+            status_code=status_code,
+            headers=response_headers,
+        )
+
+    def _client_ip(self, request: Request) -> str | None:
+        if self.settings.auth_trust_proxy_headers:
+            forwarded = request.headers.get("x-forwarded-for")
+            if forwarded:
+                return forwarded.split(",", 1)[0].strip()[:64]
+        return request.client.host[:64] if request.client else None
+
+    async def _audit_request(
+        self,
+        request: Request,
+        context: AuthContext,
+        response_code: int,
+        elapsed_ms: int,
+        bytes_uploaded: int,
+        bytes_downloaded: int,
+    ) -> None:
+        data = {
+            "request_id": getattr(request.state, "request_id", "-"),
+            "api_key_id": context.api_key_id,
+            "key_name": context.key_name,
+            "ip_address": self._client_ip(request),
+            "user_agent": request.headers.get("user-agent", "")[:512] or None,
+            "endpoint": request.url.path,
+            "http_method": request.method,
+            "response_code": response_code,
+            "processing_time_ms": elapsed_ms,
+            "bytes_uploaded": bytes_uploaded,
+            "bytes_downloaded": bytes_downloaded,
+        }
+        try:
+            await self.audit.record(**data)
+        except Exception:
+            logger.exception(
+                "authentication audit persistence failed",
+                extra={"api_key_id": context.api_key_id},
+            )
+        logger.info("authenticated request", extra=data)

app/security/migrations/0001_api_key_security.sql

@@ -0,0 +1,58 @@
+-- MediaRouter security schema migration 0001 (SQLite).
+-- Production startup applies the equivalent SQLAlchemy metadata transactionally.
+CREATE TABLE IF NOT EXISTS api_keys (
+  id VARCHAR(36) PRIMARY KEY,
+  name VARCHAR(120) NOT NULL,
+  key_prefix VARCHAR(40) NOT NULL,
+  key_hash VARCHAR(64) NOT NULL UNIQUE,
+  environment VARCHAR(16) NOT NULL,
+  status VARCHAR(16) NOT NULL,
+  role VARCHAR(64),
+  scopes JSON NOT NULL,
+  created_at DATETIME NOT NULL,
+  last_used_at DATETIME,
+  expires_at DATETIME,
+  grace_expires_at DATETIME,
+  created_by VARCHAR(120),
+  notes TEXT,
+  rotated_from_id VARCHAR(36) REFERENCES api_keys(id) ON DELETE SET NULL,
+  requests_per_minute INTEGER NOT NULL,
+  concurrent_jobs INTEGER NOT NULL,
+  uploads_per_hour INTEGER NOT NULL,
+  processing_bytes_per_day BIGINT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS ix_api_keys_key_prefix ON api_keys(key_prefix);
+CREATE UNIQUE INDEX IF NOT EXISTS ix_api_keys_key_hash ON api_keys(key_hash);
+CREATE INDEX IF NOT EXISTS ix_api_keys_status ON api_keys(status);
+CREATE INDEX IF NOT EXISTS ix_api_keys_expires_at ON api_keys(expires_at);
+
+CREATE TABLE IF NOT EXISTS audit_logs (
+  id VARCHAR(36) PRIMARY KEY,
+  request_id VARCHAR(36) NOT NULL,
+  api_key_id VARCHAR(36) REFERENCES api_keys(id) ON DELETE SET NULL,
+  key_name VARCHAR(120),
+  ip_address VARCHAR(64),
+  user_agent VARCHAR(512),
+  endpoint VARCHAR(1024) NOT NULL,
+  http_method VARCHAR(16) NOT NULL,
+  response_code INTEGER NOT NULL,
+  processing_time_ms INTEGER NOT NULL,
+  bytes_uploaded BIGINT NOT NULL DEFAULT 0,
+  bytes_downloaded BIGINT NOT NULL DEFAULT 0,
+  created_at DATETIME NOT NULL
+);
+CREATE INDEX IF NOT EXISTS ix_audit_logs_api_key_id ON audit_logs(api_key_id);
+CREATE INDEX IF NOT EXISTS ix_audit_logs_created_at ON audit_logs(created_at);
+CREATE INDEX IF NOT EXISTS ix_audit_logs_request_id ON audit_logs(request_id);
+
+CREATE TABLE IF NOT EXISTS rate_limits (
+  api_key_id VARCHAR(36) NOT NULL REFERENCES api_keys(id) ON DELETE CASCADE,
+  bucket_type VARCHAR(32) NOT NULL,
+  bucket_start DATETIME NOT NULL,
+  count BIGINT NOT NULL DEFAULT 0,
+  units BIGINT NOT NULL DEFAULT 0,
+  updated_at DATETIME NOT NULL,
+  PRIMARY KEY(api_key_id, bucket_type, bucket_start)
+);
+CREATE INDEX IF NOT EXISTS ix_rate_limits_api_key_id ON rate_limits(api_key_id);
+CREATE INDEX IF NOT EXISTS ix_rate_limits_bucket_start ON rate_limits(bucket_start);

app/security/models.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from uuid import uuid4
+
+from sqlalchemy import BigInteger, DateTime, ForeignKey, Index, Integer, JSON, String, Text
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+
+def utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class APIKey(Base):
+    __tablename__ = "api_keys"
+    __table_args__ = (
+        Index("ix_api_keys_key_prefix", "key_prefix"),
+        Index("ix_api_keys_key_hash", "key_hash", unique=True),
+        Index("ix_api_keys_status", "status"),
+        Index("ix_api_keys_expires_at", "expires_at"),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid4()))
+    name: Mapped[str] = mapped_column(String(120), nullable=False)
+    key_prefix: Mapped[str] = mapped_column(String(40), nullable=False)
+    key_hash: Mapped[str] = mapped_column(String(64), nullable=False)
+    environment: Mapped[str] = mapped_column(String(16), nullable=False)
+    status: Mapped[str] = mapped_column(String(16), nullable=False, default="active")
+    role: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    scopes: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, default=utcnow
+    )
+    last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    grace_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    created_by: Mapped[str | None] = mapped_column(String(120))
+    notes: Mapped[str | None] = mapped_column(Text)
+    rotated_from_id: Mapped[str | None] = mapped_column(
+        String(36), ForeignKey("api_keys.id", ondelete="SET NULL")
+    )
+    requests_per_minute: Mapped[int] = mapped_column(Integer, nullable=False, default=100)
+    concurrent_jobs: Mapped[int] = mapped_column(Integer, nullable=False, default=10)
+    uploads_per_hour: Mapped[int] = mapped_column(Integer, nullable=False, default=20)
+    processing_bytes_per_day: Mapped[int] = mapped_column(
+        BigInteger, nullable=False, default=107_374_182_400
+    )
+
+
+class AuditLog(Base):
+    __tablename__ = "audit_logs"
+    __table_args__ = (
+        Index("ix_audit_logs_api_key_id", "api_key_id"),
+        Index("ix_audit_logs_created_at", "created_at"),
+        Index("ix_audit_logs_request_id", "request_id"),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid4()))
+    request_id: Mapped[str] = mapped_column(String(36), nullable=False)
+    api_key_id: Mapped[str | None] = mapped_column(
+        String(36), ForeignKey("api_keys.id", ondelete="SET NULL")
+    )
+    key_name: Mapped[str | None] = mapped_column(String(120))
+    ip_address: Mapped[str | None] = mapped_column(String(64))
+    user_agent: Mapped[str | None] = mapped_column(String(512))
+    endpoint: Mapped[str] = mapped_column(String(1024), nullable=False)
+    http_method: Mapped[str] = mapped_column(String(16), nullable=False)
+    response_code: Mapped[int] = mapped_column(Integer, nullable=False)
+    processing_time_ms: Mapped[int] = mapped_column(Integer, nullable=False)
+    bytes_uploaded: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    bytes_downloaded: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, default=utcnow
+    )
+
+
+class RateLimit(Base):
+    __tablename__ = "rate_limits"
+    __table_args__ = (
+        Index("ix_rate_limits_api_key_id", "api_key_id"),
+        Index("ix_rate_limits_bucket_start", "bucket_start"),
+    )
+
+    api_key_id: Mapped[str] = mapped_column(
+        String(36), ForeignKey("api_keys.id", ondelete="CASCADE"), primary_key=True
+    )
+    bucket_type: Mapped[str] = mapped_column(String(32), primary_key=True)
+    bucket_start: Mapped[datetime] = mapped_column(DateTime(timezone=True), primary_key=True)
+    count: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    units: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, default=utcnow, onupdate=utcnow
+    )

app/security/policy.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import json
+
+from fastapi import Request
+
+PUBLIC_GET_PATHS = frozenset(
+    {"/", "/health", "/version", "/docs", "/docs/oauth2-redirect", "/openapi.json", "/redoc"}
+)
+
+
+class ScopePolicy:
+    """Maps existing and future API route families to stable authorization scopes."""
+
+    @staticmethod
+    def is_public(request: Request) -> bool:
+        return request.method == "GET" and request.url.path in PUBLIC_GET_PATHS
+
+    async def required_scope(self, request: Request) -> str | None:
+        path = request.url.path
+        method = request.method
+        if path == "/v1/auth/context":
+            return None
+        if path.startswith("/mcp"):
+            return await self._mcp_scope(request)
+        if path.startswith("/v1/api-keys") or path.startswith("/v1/audit-logs"):
+            return "admin"
+        if path == "/v1/templates" or path == "/v1/templates/categories" or (
+            path.startswith("/v1/templates/") and path != "/v1/templates/run"
+        ):
+            return "templates:read" if method == "GET" else "admin"
+        if path == "/v1/templates/run":
+            return "templates:run"
+        if path.startswith("/v1/jobs"):
+            if method == "GET":
+                return "jobs:read"
+            if method == "DELETE" or path.endswith("/cancel"):
+                return "jobs:cancel"
+            return "jobs:create"
+        if path.startswith("/v1/assets"):
+            if method == "GET":
+                return "assets:read"
+            if method == "DELETE":
+                return "assets:delete"
+            return "assets:write"
+        if path.startswith("/v1/media/") and method == "GET":
+            return "operations:read"
+        if path.startswith(
+            ("/v1/video", "/v1/audio", "/v1/image", "/v1/whisper", "/v1/ytdlp")
+        ) or path == "/v1/probe":
+            return "operations:execute"
+        if method == "GET":
+            return "system:read"
+        return "admin"
+
+    @staticmethod
+    async def _mcp_scope(request: Request) -> str:
+        if request.method != "POST":
+            return "mcp:read"
+        try:
+            payload = json.loads(await request.body())
+        except (json.JSONDecodeError, UnicodeDecodeError):
+            return "mcp:read"
+        messages = payload if isinstance(payload, list) else [payload]
+        methods = {
+            str(message.get("method", ""))
+            for message in messages
+            if isinstance(message, dict)
+        }
+        return "mcp:execute" if "tools/call" in methods else "mcp:read"
+
+    @staticmethod
+    def is_job(required_scope: str | None) -> bool:
+        return required_scope in {
+            "templates:run",
+            "operations:execute",
+            "jobs:create",
+            "mcp:execute",
+        }
+
+    @staticmethod
+    def is_upload(request: Request, required_scope: str | None) -> bool:
+        return request.method in {"POST", "PUT", "PATCH"} and required_scope in {
+            "templates:run",
+            "operations:execute",
+            "jobs:create",
+            "assets:write",
+            "mcp:execute",
+        }

app/security/rate_limit.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+import asyncio
+import math
+import time
+from collections import defaultdict, deque
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
+
+from app.security.context import AuthContext
+from app.security.database import SecurityDatabase
+from app.security.errors import RateLimitError
+from app.security.models import RateLimit, utcnow
+
+
+@dataclass(slots=True)
+class RateLimitLease:
+    limiter: APIKeyRateLimiter
+    api_key_id: str
+    concurrent: bool
+
+    async def release(self) -> None:
+        if self.concurrent:
+            await self.limiter.release_job(self.api_key_id)
+
+
+class APIKeyRateLimiter:
+    """Low-latency per-key windows with durable aggregate counters for auditing."""
+
+    def __init__(self, database: SecurityDatabase) -> None:
+        self.database = database
+        self._lock = asyncio.Lock()
+        self._requests: dict[str, deque[float]] = defaultdict(deque)
+        self._uploads: dict[str, deque[float]] = defaultdict(deque)
+        self._concurrent: dict[str, int] = defaultdict(int)
+        self._daily_bytes: dict[tuple[str, str], int] = defaultdict(int)
+
+    async def acquire(
+        self,
+        context: AuthContext,
+        *,
+        is_job: bool,
+        is_upload: bool,
+        uploaded_bytes: int,
+    ) -> RateLimitLease:
+        now = time.time()
+        today = datetime.now(timezone.utc).date().isoformat()
+        retry_after = 0
+        async with self._lock:
+            requests = self._requests[context.api_key_id]
+            self._prune(requests, now - 60)
+            if len(requests) >= context.requests_per_minute:
+                retry_after = max(1, math.ceil(requests[0] + 60 - now))
+            uploads = self._uploads[context.api_key_id]
+            self._prune(uploads, now - 3600)
+            if not retry_after and is_upload and len(uploads) >= context.uploads_per_hour:
+                retry_after = max(1, math.ceil(uploads[0] + 3600 - now))
+            daily_key = (context.api_key_id, today)
+            daily_total = self._daily_bytes[daily_key]
+            if (
+                not retry_after
+                and uploaded_bytes
+                and daily_total + uploaded_bytes > context.processing_bytes_per_day
+            ):
+                tomorrow = datetime.now(timezone.utc).replace(
+                    hour=0, minute=0, second=0, microsecond=0
+                ) + timedelta(days=1)
+                retry_after = max(1, int((tomorrow - datetime.now(timezone.utc)).total_seconds()))
+            if (
+                not retry_after
+                and is_job
+                and self._concurrent[context.api_key_id] >= context.concurrent_jobs
+            ):
+                retry_after = 1
+            if retry_after:
+                raise RateLimitError(retry_after)
+            requests.append(now)
+            if is_upload:
+                uploads.append(now)
+            if uploaded_bytes:
+                self._daily_bytes[daily_key] += uploaded_bytes
+            if is_job:
+                self._concurrent[context.api_key_id] += 1
+        await self._record(context.api_key_id, "requests_minute", 1, 0, 60)
+        if is_upload:
+            await self._record(context.api_key_id, "uploads_hour", 1, 0, 3600)
+        if uploaded_bytes:
+            await self._record(
+                context.api_key_id, "processing_bytes_day", 0, uploaded_bytes, 86_400
+            )
+        return RateLimitLease(self, context.api_key_id, is_job)
+
+    async def release_job(self, api_key_id: str) -> None:
+        async with self._lock:
+            self._concurrent[api_key_id] = max(0, self._concurrent[api_key_id] - 1)
+
+    @staticmethod
+    def _prune(values: deque[float], cutoff: float) -> None:
+        while values and values[0] <= cutoff:
+            values.popleft()
+
+    async def _record(
+        self, api_key_id: str, bucket_type: str, count: int, units: int, seconds: int
+    ) -> None:
+        now = datetime.now(timezone.utc)
+        epoch = int(now.timestamp())
+        bucket_start = datetime.fromtimestamp(epoch - (epoch % seconds), timezone.utc)
+        async with self.database.session() as session:
+            statement = sqlite_insert(RateLimit).values(
+                api_key_id=api_key_id,
+                bucket_type=bucket_type,
+                bucket_start=bucket_start,
+                count=count,
+                units=units,
+                updated_at=utcnow(),
+            )
+            statement = statement.on_conflict_do_update(
+                index_elements=[
+                    RateLimit.api_key_id,
+                    RateLimit.bucket_type,
+                    RateLimit.bucket_start,
+                ],
+                set_={
+                    "count": RateLimit.count + statement.excluded.count,
+                    "units": RateLimit.units + statement.excluded.units,
+                    "updated_at": utcnow(),
+                },
+            )
+            await session.execute(statement)
+            await session.commit()

app/security/schemas.py

@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from app.security.scopes import ALL_SCOPES
+
+Environment = Literal["live", "test"]
+
+
+class APIKeyCreate(BaseModel):
+    name: str = Field(min_length=1, max_length=120)
+    environment: Environment = "live"
+    role: str | None = Field(default="viewer", min_length=1, max_length=64)
+    scopes: list[str] = Field(default_factory=list)
+    expires_at: datetime | None = None
+    expires_in_seconds: int | None = Field(default=None, ge=60, le=31_536_000)
+    notes: str | None = Field(default=None, max_length=4000)
+    requests_per_minute: int | None = Field(default=None, ge=1, le=1_000_000)
+    concurrent_jobs: int | None = Field(default=None, ge=1, le=10_000)
+    uploads_per_hour: int | None = Field(default=None, ge=1, le=1_000_000)
+    processing_bytes_per_day: int | None = Field(default=None, ge=1_048_576)
+
+    @field_validator("name")
+    @classmethod
+    def validate_name(cls, value: str) -> str:
+        normalized = value.strip()
+        if not normalized:
+            raise ValueError("API key name cannot be blank")
+        return normalized
+
+    @field_validator("scopes")
+    @classmethod
+    def validate_scopes(cls, values: list[str]) -> list[str]:
+        normalized = list(dict.fromkeys(value.strip().lower() for value in values))
+        unknown = set(normalized) - ALL_SCOPES
+        if unknown:
+            raise ValueError(f"Unsupported scopes: {sorted(unknown)}")
+        return normalized
+
+    @model_validator(mode="after")
+    def validate_expiration(self) -> APIKeyCreate:
+        if self.expires_at is not None and self.expires_in_seconds is not None:
+            raise ValueError("Use either expires_at or expires_in_seconds, not both")
+        return self
+
+
+class APIKeyPatch(BaseModel):
+    name: str | None = Field(default=None, min_length=1, max_length=120)
+    role: str | None = Field(default=None, min_length=1, max_length=64)
+    scopes: list[str] | None = None
+    expires_at: datetime | None = None
+    clear_expiration: bool = False
+    notes: str | None = Field(default=None, max_length=4000)
+    requests_per_minute: int | None = Field(default=None, ge=1, le=1_000_000)
+    concurrent_jobs: int | None = Field(default=None, ge=1, le=10_000)
+    uploads_per_hour: int | None = Field(default=None, ge=1, le=1_000_000)
+    processing_bytes_per_day: int | None = Field(default=None, ge=1_048_576)
+
+    @field_validator("name")
+    @classmethod
+    def validate_name(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        normalized = value.strip()
+        if not normalized:
+            raise ValueError("API key name cannot be blank")
+        return normalized
+
+    @field_validator("scopes")
+    @classmethod
+    def validate_scopes(cls, values: list[str] | None) -> list[str] | None:
+        if values is None:
+            return None
+        normalized = list(dict.fromkeys(value.strip().lower() for value in values))
+        unknown = set(normalized) - ALL_SCOPES
+        if unknown:
+            raise ValueError(f"Unsupported scopes: {sorted(unknown)}")
+        return normalized
+
+    @model_validator(mode="after")
+    def validate_expiration(self) -> APIKeyPatch:
+        if self.clear_expiration and "expires_at" in self.model_fields_set:
+            raise ValueError("Use either expires_at or clear_expiration, not both")
+        return self
+
+
+class APIKeyRotate(BaseModel):
+    grace_period_seconds: int = Field(default=0, ge=0, le=86_400)
+
+
+class APIKeyView(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: str
+    name: str
+    key_prefix: str
+    environment: str
+    status: str
+    role: str | None
+    scopes: list[str]
+    created_at: datetime
+    last_used_at: datetime | None
+    expires_at: datetime | None
+    grace_expires_at: datetime | None
+    created_by: str | None
+    notes: str | None
+    rotated_from_id: str | None
+    requests_per_minute: int
+    concurrent_jobs: int
+    uploads_per_hour: int
+    processing_bytes_per_day: int
+
+
+class APIKeyCreated(APIKeyView):
+    api_key: str
+
+
+class APIKeyList(BaseModel):
+    items: list[APIKeyView]
+    total: int
+
+
+class AuthContextView(BaseModel):
+    id: str
+    name: str
+    key_prefix: str
+    environment: str
+    role: str | None
+    scopes: list[str]
+    expires_at: datetime | None
+
+
+class AuditLogView(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: str
+    request_id: str
+    api_key_id: str | None
+    key_name: str | None
+    ip_address: str | None
+    user_agent: str | None
+    endpoint: str
+    http_method: str
+    response_code: int
+    processing_time_ms: int
+    bytes_uploaded: int
+    bytes_downloaded: int
+    created_at: datetime

app/security/scopes.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+
+ALL_SCOPES = frozenset(
+    {
+        "templates:read",
+        "templates:run",
+        "operations:read",
+        "operations:execute",
+        "jobs:read",
+        "jobs:create",
+        "jobs:cancel",
+        "assets:read",
+        "assets:write",
+        "assets:delete",
+        "mcp:read",
+        "mcp:execute",
+        "system:read",
+        "admin",
+    }
+)
+
+DEFAULT_ROLE_SCOPES: dict[str, frozenset[str]] = {
+    "admin": frozenset({"admin"}),
+    "developer": frozenset(
+        {
+            "templates:read",
+            "templates:run",
+            "operations:read",
+            "operations:execute",
+            "jobs:read",
+            "jobs:create",
+            "jobs:cancel",
+            "assets:read",
+            "assets:write",
+            "mcp:read",
+            "mcp:execute",
+            "system:read",
+        }
+    ),
+    "operator": frozenset(
+        {
+            "templates:read",
+            "templates:run",
+            "operations:read",
+            "operations:execute",
+            "jobs:read",
+            "jobs:create",
+            "jobs:cancel",
+            "assets:read",
+            "mcp:read",
+            "mcp:execute",
+            "system:read",
+        }
+    ),
+    "viewer": frozenset(
+        {
+            "templates:read",
+            "operations:read",
+            "jobs:read",
+            "assets:read",
+            "mcp:read",
+            "system:read",
+        }
+    ),
+}
+
+
+def configured_roles(
+    overrides: Mapping[str, Sequence[str]] | None = None,
+) -> dict[str, frozenset[str]]:
+    roles = dict(DEFAULT_ROLE_SCOPES)
+    for role, scopes in (overrides or {}).items():
+        normalized_role = role.strip().lower()
+        normalized_scopes = frozenset(scope.strip().lower() for scope in scopes)
+        unknown = normalized_scopes - ALL_SCOPES
+        if unknown:
+            raise ValueError(f"Role '{normalized_role}' has unsupported scopes: {sorted(unknown)}")
+        roles[normalized_role] = normalized_scopes
+    return roles
+
+
+def effective_scopes(
+    role: str | None,
+    explicit_scopes: Sequence[str],
+    overrides: Mapping[str, Sequence[str]] | None = None,
+) -> frozenset[str]:
+    scopes = set(scope.strip().lower() for scope in explicit_scopes)
+    if role:
+        scopes.update(configured_roles(overrides).get(role.strip().lower(), ()))
+    if "admin" in scopes:
+        return ALL_SCOPES
+    return frozenset(scopes)

app/security/service.py

@@ -0,0 +1,375 @@
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import hmac
+import re
+import secrets
+import time
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import func, select, update
+
+from app.core.config import Settings
+from app.security.context import AuthContext
+from app.security.database import SecurityDatabase
+from app.security.errors import (
+    APIKeyConflictError,
+    APIKeyNotFoundError,
+    ForbiddenError,
+    UnauthorizedError,
+)
+from app.security.models import APIKey
+from app.security.schemas import APIKeyCreate, APIKeyPatch
+from app.security.scopes import configured_roles, effective_scopes
+
+KEY_PATTERN = re.compile(r"^mp_([a-z][a-z0-9]{1,15})_([A-Za-z0-9_-]{43,})$")
+HASH_PATTERN = re.compile(r"^[a-f0-9]{64}$")
+
+
+def utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def aware(value: datetime | None) -> datetime | None:
+    if value is None:
+        return None
+    return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value.astimezone(timezone.utc)
+
+
+@dataclass(frozen=True, slots=True)
+class KeyMaterial:
+    api_key: str
+    key_prefix: str
+    key_hash: str
+    environment: str
+
+
+class APIKeyService:
+    """Creates and validates opaque API keys without retaining plaintext secrets."""
+
+    def __init__(self, database: SecurityDatabase, settings: Settings) -> None:
+        self.database = database
+        self.settings = settings
+        self.roles = configured_roles(settings.auth_role_scopes)
+        self._last_used_cache: dict[str, float] = {}
+        self._last_used_lock = asyncio.Lock()
+
+    @staticmethod
+    def generate_material(environment: str) -> KeyMaterial:
+        normalized = environment.strip().lower()
+        if normalized not in {"live", "test"}:
+            raise ValueError("API key environment must be live or test")
+        secret = secrets.token_urlsafe(32)
+        api_key = f"mp_{normalized}_{secret}"
+        return KeyMaterial(
+            api_key=api_key,
+            key_prefix=f"mp_{normalized}_{secret[:8]}",
+            key_hash=hashlib.sha256(api_key.encode("utf-8")).hexdigest(),
+            environment=normalized,
+        )
+
+    @staticmethod
+    def hash_key(api_key: str) -> str:
+        return hashlib.sha256(api_key.encode("utf-8")).hexdigest()
+
+    @staticmethod
+    def parse_key(api_key: str) -> tuple[str, str]:
+        match = KEY_PATTERN.fullmatch(api_key)
+        if not match:
+            raise UnauthorizedError
+        environment, secret = match.groups()
+        return environment, f"mp_{environment}_{secret[:8]}"
+
+    async def ensure_bootstrap_admin(self) -> None:
+        key_hash = self.settings.auth_bootstrap_key_hash.strip().lower()
+        key_prefix = self.settings.auth_bootstrap_key_prefix.strip()
+        if not key_hash and not key_prefix:
+            return
+        if not HASH_PATTERN.fullmatch(key_hash) or not re.fullmatch(
+            r"mp_[a-z][a-z0-9]{1,15}_[A-Za-z0-9_-]{8}", key_prefix
+        ):
+            raise ValueError("Bootstrap key hash or prefix is malformed")
+        prefix_environment = key_prefix.split("_", 2)[1]
+        if prefix_environment != self.settings.auth_bootstrap_environment:
+            raise ValueError("Bootstrap key prefix and environment do not match")
+        async with self.database.session() as session:
+            count = await session.scalar(select(func.count()).select_from(APIKey))
+            if count:
+                return
+            session.add(
+                APIKey(
+                    name=self.settings.auth_bootstrap_key_name,
+                    key_prefix=key_prefix,
+                    key_hash=key_hash,
+                    environment=self.settings.auth_bootstrap_environment,
+                    status="active",
+                    role="admin",
+                    scopes=["admin"],
+                    created_by="bootstrap",
+                    notes="Hash-only bootstrap administrator",
+                    requests_per_minute=self.settings.auth_default_requests_per_minute,
+                    concurrent_jobs=self.settings.auth_default_concurrent_jobs,
+                    uploads_per_hour=self.settings.auth_default_uploads_per_hour,
+                    processing_bytes_per_day=(
+                        self.settings.auth_default_processing_bytes_per_day
+                    ),
+                )
+            )
+            await session.commit()
+
+    async def create(
+        self, payload: APIKeyCreate, *, created_by: str | None
+    ) -> tuple[APIKey, str]:
+        role = payload.role.strip().lower() if payload.role else None
+        if role and role not in self.roles:
+            raise APIKeyConflictError(f"Unknown role '{role}'")
+        material = self.generate_material(payload.environment)
+        expires_at = aware(payload.expires_at)
+        if payload.expires_in_seconds is not None:
+            expires_at = utcnow() + timedelta(seconds=payload.expires_in_seconds)
+        record = APIKey(
+            name=payload.name.strip(),
+            key_prefix=material.key_prefix,
+            key_hash=material.key_hash,
+            environment=material.environment,
+            status="active",
+            role=role,
+            scopes=payload.scopes,
+            expires_at=expires_at,
+            created_by=created_by,
+            notes=payload.notes,
+            requests_per_minute=(
+                payload.requests_per_minute
+                or self.settings.auth_default_requests_per_minute
+            ),
+            concurrent_jobs=(
+                payload.concurrent_jobs or self.settings.auth_default_concurrent_jobs
+            ),
+            uploads_per_hour=(
+                payload.uploads_per_hour or self.settings.auth_default_uploads_per_hour
+            ),
+            processing_bytes_per_day=(
+                payload.processing_bytes_per_day
+                or self.settings.auth_default_processing_bytes_per_day
+            ),
+        )
+        async with self.database.session() as session:
+            session.add(record)
+            await session.commit()
+            await session.refresh(record)
+        return record, material.api_key
+
+    async def authenticate(self, api_key: str) -> AuthContext:
+        environment, key_prefix = self.parse_key(api_key)
+        supplied_hash = self.hash_key(api_key)
+        async with self.database.session() as session:
+            candidates = list(
+                (
+                    await session.scalars(
+                        select(APIKey).where(APIKey.key_prefix == key_prefix)
+                    )
+                ).all()
+            )
+        record: APIKey | None = None
+        for candidate in candidates:
+            if hmac.compare_digest(candidate.key_hash, supplied_hash):
+                record = candidate
+        now = utcnow()
+        if record is None or record.environment != environment:
+            raise UnauthorizedError
+        expires_at = aware(record.expires_at)
+        grace_expires_at = aware(record.grace_expires_at)
+        if record.status == "rotating":
+            if grace_expires_at is None or grace_expires_at <= now:
+                await self._finalize_rotation(record.id)
+                raise UnauthorizedError
+        elif record.status != "active":
+            raise UnauthorizedError
+        if expires_at is not None and expires_at <= now:
+            raise UnauthorizedError
+        scopes = effective_scopes(record.role, record.scopes or [], self.settings.auth_role_scopes)
+        return AuthContext(
+            api_key_id=record.id,
+            key_name=record.name,
+            key_prefix=record.key_prefix,
+            environment=record.environment,
+            role=record.role,
+            scopes=scopes,
+            requests_per_minute=record.requests_per_minute,
+            concurrent_jobs=record.concurrent_jobs,
+            uploads_per_hour=record.uploads_per_hour,
+            processing_bytes_per_day=record.processing_bytes_per_day,
+            expires_at=expires_at,
+        )
+
+    @staticmethod
+    def authorize(context: AuthContext, required_scope: str | None) -> None:
+        if required_scope is not None and not context.allows(required_scope):
+            raise ForbiddenError
+
+    async def mark_used(self, context: AuthContext) -> None:
+        await self._touch_last_used(context.api_key_id)
+
+    async def list(self, *, offset: int = 0, limit: int = 100) -> tuple[list[APIKey], int]:
+        await self._finalize_expired_rotations()
+        async with self.database.session() as session:
+            total = int(await session.scalar(select(func.count()).select_from(APIKey)) or 0)
+            records = list(
+                (
+                    await session.scalars(
+                        select(APIKey)
+                        .order_by(APIKey.created_at.desc())
+                        .offset(offset)
+                        .limit(limit)
+                    )
+                ).all()
+            )
+        return records, total
+
+    async def get(self, key_id: str) -> APIKey:
+        await self._finalize_expired_rotations()
+        async with self.database.session() as session:
+            record = await session.get(APIKey, key_id)
+            if record is None:
+                raise APIKeyNotFoundError
+            return record
+
+    async def patch(self, key_id: str, payload: APIKeyPatch) -> APIKey:
+        async with self.database.session() as session:
+            record = await session.get(APIKey, key_id)
+            if record is None:
+                raise APIKeyNotFoundError
+            fields = payload.model_fields_set
+            if payload.name is not None:
+                record.name = payload.name.strip()
+            if "role" in fields:
+                role = payload.role.strip().lower() if payload.role else None
+                if role and role not in self.roles:
+                    raise APIKeyConflictError(f"Unknown role '{role}'")
+                record.role = role
+            if payload.scopes is not None:
+                record.scopes = payload.scopes
+            if payload.clear_expiration:
+                record.expires_at = None
+            elif "expires_at" in fields:
+                record.expires_at = aware(payload.expires_at)
+            if "notes" in fields:
+                record.notes = payload.notes
+            for field in (
+                "requests_per_minute",
+                "concurrent_jobs",
+                "uploads_per_hour",
+                "processing_bytes_per_day",
+            ):
+                value = getattr(payload, field)
+                if value is not None:
+                    setattr(record, field, value)
+            await session.commit()
+            await session.refresh(record)
+            return record
+
+    async def set_status(self, key_id: str, status: str) -> APIKey:
+        if status not in {"active", "disabled", "revoked"}:
+            raise APIKeyConflictError("Unsupported API key status transition")
+        async with self.database.session() as session:
+            record = await session.get(APIKey, key_id)
+            if record is None:
+                raise APIKeyNotFoundError
+            expires_at = aware(record.expires_at)
+            if record.status == "revoked" and status != "revoked":
+                raise APIKeyConflictError("Revoked API keys cannot be changed")
+            if status == "active" and record.status != "disabled":
+                raise APIKeyConflictError("Only disabled API keys can be enabled")
+            if (
+                status == "active"
+                and expires_at is not None
+                and expires_at <= utcnow()
+            ):
+                raise APIKeyConflictError("Expired API keys cannot be enabled")
+            if status == "disabled" and record.status != "active":
+                raise APIKeyConflictError("Only active API keys can be disabled")
+            record.status = status
+            if status != "rotating":
+                record.grace_expires_at = None
+            await session.commit()
+            await session.refresh(record)
+            return record
+
+    async def rotate(
+        self, key_id: str, grace_period_seconds: int, *, created_by: str | None
+    ) -> tuple[APIKey, str]:
+        async with self.database.session() as session:
+            old = await session.get(APIKey, key_id)
+            if old is None:
+                raise APIKeyNotFoundError
+            if old.status != "active":
+                raise APIKeyConflictError("Only active API keys can be rotated")
+            old_expires_at = aware(old.expires_at)
+            if old_expires_at is not None and old_expires_at <= utcnow():
+                raise APIKeyConflictError("Expired API keys cannot be rotated")
+            material = self.generate_material(old.environment)
+            replacement = APIKey(
+                name=old.name,
+                key_prefix=material.key_prefix,
+                key_hash=material.key_hash,
+                environment=old.environment,
+                status="active",
+                role=old.role,
+                scopes=list(old.scopes or []),
+                expires_at=old.expires_at,
+                created_by=created_by,
+                notes=old.notes,
+                rotated_from_id=old.id,
+                requests_per_minute=old.requests_per_minute,
+                concurrent_jobs=old.concurrent_jobs,
+                uploads_per_hour=old.uploads_per_hour,
+                processing_bytes_per_day=old.processing_bytes_per_day,
+            )
+            session.add(replacement)
+            old.status = "rotating" if grace_period_seconds else "revoked"
+            old.grace_expires_at = (
+                utcnow() + timedelta(seconds=grace_period_seconds)
+                if grace_period_seconds
+                else None
+            )
+            await session.commit()
+            await session.refresh(replacement)
+            return replacement, material.api_key
+
+    async def _touch_last_used(self, key_id: str) -> None:
+        interval = self.settings.auth_last_used_update_seconds
+        current = time.monotonic()
+        async with self._last_used_lock:
+            previous = self._last_used_cache.get(key_id)
+            if previous is not None and current - previous < interval:
+                return
+            self._last_used_cache[key_id] = current
+        async with self.database.session() as session:
+            record = await session.get(APIKey, key_id)
+            if record is not None:
+                record.last_used_at = utcnow()
+                await session.commit()
+
+    async def _finalize_rotation(self, key_id: str) -> None:
+        async with self.database.session() as session:
+            await session.execute(
+                update(APIKey)
+                .where(APIKey.id == key_id, APIKey.status == "rotating")
+                .values(status="revoked", grace_expires_at=None)
+            )
+            await session.commit()
+
+    async def _finalize_expired_rotations(self) -> None:
+        async with self.database.session() as session:
+            await session.execute(
+                update(APIKey)
+                .where(
+                    APIKey.status == "rotating",
+                    APIKey.grace_expires_at.is_not(None),
+                    APIKey.grace_expires_at <= utcnow(),
+                )
+                .values(status="revoked", grace_expires_at=None)
+            )
+            await session.commit()

main.py

@@ -11,13 +11,14 @@ from fastapi.responses import JSONResponse, ORJSONResponse
 from starlette.middleware.base import RequestResponseEndpoint
 from starlette.responses import Response
 
-from app.api import audio, health, image, media, probe, templates, video, whisper, ytdlp
+from app.api import api_keys, audio, health, image, media, probe, templates, video, whisper, ytdlp
 from app.container import build_container
 from app.core.config import Settings, get_settings
 from app.core.exceptions import MediaAPIError
 from app.core.logger import configure_logging, get_logger, request_id_context
 from app.core.response import ErrorBody, ErrorResponse
 from app.mcp.server import create_mcp_server
+from app.security.middleware import APIKeyAuthenticationMiddleware
 from app.workers.cleanup_worker import CleanupWorker
 
 configure_logging()
@@ -46,6 +47,8 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     async def lifespan(application: FastAPI) -> AsyncIterator[None]:
         application.state.container = container
         application.state.mcp_server = mcp_server
+        await container.security_database.initialize()
+        await container.api_keys.ensure_bootstrap_admin()
         async with mcp_server.session_manager.run():
             await cleanup_worker.start()
             logger.info(
@@ -56,6 +59,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
                 yield
             finally:
                 await cleanup_worker.stop()
+                await container.security_database.close()
                 logger.info("media API stopped")
 
     application = FastAPI(
@@ -74,6 +78,15 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     application.state.container = container
     application.state.mcp_server = mcp_server
 
+    # Added before request_context so the request-ID/logging middleware remains outermost.
+    application.add_middleware(
+        APIKeyAuthenticationMiddleware,
+        settings=active_settings,
+        api_keys=container.api_keys,
+        rate_limiter=container.rate_limiter,
+        audit=container.audit,
+    )
+
     @application.middleware("http")
     async def request_context(request: Request, call_next: RequestResponseEndpoint) -> Response:
         request_id = str(uuid4())
@@ -152,6 +165,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
         return DefaultJSONResponse(body.model_dump(), status_code=500)
 
     application.include_router(health.router)
+    application.include_router(api_keys.router)
     application.include_router(media.router)
     application.include_router(video.router)
     application.include_router(audio.router)
@@ -161,6 +175,42 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     application.include_router(whisper.router)
     application.include_router(templates.router)
     application.mount("/mcp", mcp_http_app, name="mcp")
+
+    @application.get("/", tags=["public"], include_in_schema=False)
+    async def root() -> dict[str, str]:
+        return {
+            "name": active_settings.app_name,
+            "version": active_settings.app_version,
+            "status": "healthy",
+            "docs": "/docs",
+        }
+
+    @application.get("/version", tags=["public"], include_in_schema=False)
+    async def version() -> dict[str, str]:
+        return {"version": active_settings.app_version}
+
+    original_openapi = application.openapi
+
+    def secure_openapi() -> dict[str, object]:
+        if application.openapi_schema:
+            return application.openapi_schema  # type: ignore[return-value]
+        schema = original_openapi()
+        components = schema.setdefault("components", {})
+        components.setdefault("securitySchemes", {})["APIKeyBearer"] = {
+            "type": "http",
+            "scheme": "bearer",
+            "bearerFormat": "mp_live_…",
+            "description": "MediaRouter API key",
+        }
+        schema["security"] = [{"APIKeyBearer": []}]
+        for path in ("/health",):
+            for operation in schema.get("paths", {}).get(path, {}).values():
+                if isinstance(operation, dict):
+                    operation["security"] = []
+        application.openapi_schema = schema
+        return schema
+
+    application.openapi = secure_openapi  # type: ignore[method-assign]
     return application
 
 

requirements.txt

@@ -12,3 +12,5 @@ yt-dlp==2026.7.4
 faster-whisper==1.2.1
 mcp==1.28.1
 PyYAML==6.0.3
+SQLAlchemy==2.0.43
+aiosqlite==0.21.0

tests/conftest.py

@@ -19,4 +19,6 @@ def settings(tmp_path: Path) -> Settings:
         whisper_model="tiny",
         max_workers=1,
         allow_private_urls=True,
+        auth_enabled=False,
+        database_url=f"sqlite+aiosqlite:///{tmp_path / 'security.db'}",
     )

tests/test_authentication.py

@@ -0,0 +1,307 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from app.container import build_container
+from app.core.config import Settings
+from app.mcp.registry import MCPRegistry
+from app.security.context import auth_context
+from app.security.errors import APIKeyConflictError, ForbiddenError, RateLimitError, UnauthorizedError
+from app.security.models import APIKey, AuditLog
+from app.security.schemas import APIKeyCreate
+from app.security.service import APIKeyService
+from main import create_app
+
+
+def security_settings(tmp_path: Path, **overrides: object) -> Settings:
+    values: dict[str, object] = {
+        "_env_file": None,
+        "temp_dir": tmp_path / "temp",
+        "output_dir": tmp_path / "outputs",
+        "database_url": f"sqlite+aiosqlite:///{tmp_path / 'security.db'}",
+        "auth_enabled": True,
+        "auth_last_used_update_seconds": 0,
+        "cleanup_interval_seconds": 3600,
+        "whisper_model": "tiny",
+        "max_workers": 1,
+    }
+    values.update(overrides)
+    return Settings(**values)
+
+
+@pytest.fixture
+async def security_container(tmp_path: Path):
+    container = build_container(security_settings(tmp_path))
+    await container.security_database.initialize()
+    try:
+        yield container
+    finally:
+        await container.security_database.close()
+
+
+async def create_key(container, **overrides: object) -> tuple[APIKey, str]:
+    values: dict[str, object] = {
+        "name": "Automation",
+        "environment": "test",
+        "role": None,
+        "scopes": ["templates:read"],
+    }
+    values.update(overrides)
+    return await container.api_keys.create(APIKeyCreate(**values), created_by="tests")
+
+
+async def test_key_generation_has_256_bits_and_database_never_stores_secret(
+    security_container,
+) -> None:
+    record, secret = await create_key(security_container)
+
+    environment, encoded_secret = secret.split("_", 2)[1:]
+    raw_secret = base64.urlsafe_b64decode(encoded_secret + "=")
+    assert environment == "test"
+    assert len(raw_secret) == 32
+    assert record.key_prefix == f"mp_test_{encoded_secret[:8]}"
+    assert record.key_hash == hashlib.sha256(secret.encode()).hexdigest()
+
+    async with security_container.security_database.session() as session:
+        stored = await session.get(APIKey, record.id)
+        assert stored is not None
+        assert secret not in vars(stored).values()
+        assert not hasattr(stored, "api_key")
+
+
+async def test_authentication_rejects_invalid_expired_disabled_and_revoked_keys(
+    security_container,
+) -> None:
+    active, active_secret = await create_key(security_container)
+    assert (await security_container.api_keys.authenticate(active_secret)).api_key_id == active.id
+
+    replacement = "A" if active_secret[-1] != "A" else "B"
+    with pytest.raises(UnauthorizedError):
+        await security_container.api_keys.authenticate(active_secret[:-1] + replacement)
+
+    _, expired_secret = await create_key(
+        security_container,
+        name="Expired",
+        expires_at=datetime.now(timezone.utc) - timedelta(seconds=1),
+    )
+    with pytest.raises(UnauthorizedError):
+        await security_container.api_keys.authenticate(expired_secret)
+
+    await security_container.api_keys.set_status(active.id, "disabled")
+    with pytest.raises(UnauthorizedError):
+        await security_container.api_keys.authenticate(active_secret)
+    await security_container.api_keys.set_status(active.id, "active")
+    assert (await security_container.api_keys.authenticate(active_secret)).api_key_id == active.id
+
+    await security_container.api_keys.set_status(active.id, "revoked")
+    with pytest.raises(UnauthorizedError):
+        await security_container.api_keys.authenticate(active_secret)
+    with pytest.raises(APIKeyConflictError):
+        await security_container.api_keys.set_status(active.id, "disabled")
+    with pytest.raises(APIKeyConflictError):
+        await security_container.api_keys.set_status(active.id, "active")
+
+
+async def test_scope_enforcement_and_rotation_grace_period(security_container) -> None:
+    old, old_secret = await create_key(security_container)
+    context = await security_container.api_keys.authenticate(old_secret)
+    security_container.api_keys.authorize(context, "templates:read")
+    with pytest.raises(ForbiddenError):
+        security_container.api_keys.authorize(context, "operations:execute")
+
+    replacement, replacement_secret = await security_container.api_keys.rotate(
+        old.id, 60, created_by="tests"
+    )
+    assert replacement.rotated_from_id == old.id
+    assert (await security_container.api_keys.authenticate(old_secret)).api_key_id == old.id
+    assert (
+        await security_container.api_keys.authenticate(replacement_secret)
+    ).api_key_id == replacement.id
+    with pytest.raises(APIKeyConflictError):
+        await security_container.api_keys.set_status(old.id, "disabled")
+
+    async with security_container.security_database.session() as session:
+        rotating = await session.get(APIKey, old.id)
+        assert rotating is not None
+        rotating.grace_expires_at = datetime.now(timezone.utc) - timedelta(seconds=1)
+        await session.commit()
+    with pytest.raises(UnauthorizedError):
+        await security_container.api_keys.authenticate(old_secret)
+    assert (await security_container.api_keys.get(old.id)).status == "revoked"
+
+
+async def test_per_key_request_and_concurrent_job_limits(security_container) -> None:
+    _, request_secret = await create_key(
+        security_container, name="RPM", requests_per_minute=1
+    )
+    request_context = await security_container.api_keys.authenticate(request_secret)
+    lease = await security_container.rate_limiter.acquire(
+        request_context, is_job=False, is_upload=False, uploaded_bytes=0
+    )
+    await lease.release()
+    with pytest.raises(RateLimitError) as rate_error:
+        await security_container.rate_limiter.acquire(
+            request_context, is_job=False, is_upload=False, uploaded_bytes=0
+        )
+    assert rate_error.value.retry_after >= 1
+
+    _, job_secret = await create_key(
+        security_container, name="Concurrency", concurrent_jobs=1
+    )
+    job_context = await security_container.api_keys.authenticate(job_secret)
+    running = await security_container.rate_limiter.acquire(
+        job_context, is_job=True, is_upload=False, uploaded_bytes=0
+    )
+    with pytest.raises(RateLimitError):
+        await security_container.rate_limiter.acquire(
+            job_context, is_job=True, is_upload=False, uploaded_bytes=0
+        )
+    await running.release()
+    next_job = await security_container.rate_limiter.acquire(
+        job_context, is_job=True, is_upload=False, uploaded_bytes=0
+    )
+    await next_job.release()
+
+
+async def test_stdio_mcp_uses_shared_context_scopes_rate_limits_and_audit(
+    security_container,
+) -> None:
+    _, secret = await create_key(
+        security_container, name="MCP Reader", scopes=["mcp:read"]
+    )
+    context = await security_container.api_keys.authenticate(secret)
+    registry = MCPRegistry(security_container)
+    unauthorized = await registry.run_metadata_tool("system_info", registry.system_info_data)
+    token = auth_context.set(context)
+    try:
+        resource = await registry.safe_resource("version", registry.version_data)
+        forbidden = await registry.run_metadata_tool("system_info", registry.system_info_data)
+    finally:
+        auth_context.reset(token)
+
+    assert unauthorized["success"] is False
+    assert unauthorized["error"]["code"] == "UNAUTHORIZED"
+    assert resource["success"] is True
+    assert forbidden["success"] is False
+    assert forbidden["error"]["code"] == "FORBIDDEN"
+    async with security_container.security_database.session() as session:
+        logs = list((await session.scalars(select(AuditLog))).all())
+    assert {log.endpoint for log in logs} >= {
+        "mcp://tools/resource.version",
+        "mcp://tools/system_info",
+    }
+
+
+def test_http_middleware_public_and_authentication_contracts(tmp_path: Path) -> None:
+    material = APIKeyService.generate_material("test")
+    settings = security_settings(
+        tmp_path,
+        auth_bootstrap_key_hash=material.key_hash,
+        auth_bootstrap_key_prefix=material.key_prefix,
+        auth_bootstrap_environment="test",
+        auth_default_requests_per_minute=1000,
+    )
+    application = create_app(settings)
+    authorization = {"Authorization": f"Bearer {material.api_key}"}
+
+    with TestClient(application) as client:
+        for path in ("/", "/health", "/version", "/docs", "/openapi.json", "/redoc"):
+            assert client.get(path).status_code == 200
+
+        missing = client.get("/v1/auth/context")
+        malformed = client.get(
+            "/v1/auth/context", headers={"Authorization": "Basic not-a-mediarouter-key"}
+        )
+        invalid = client.get(
+            "/v1/auth/context", headers={"Authorization": "Bearer mp_test_invalid"}
+        )
+        for response in (missing, malformed, invalid):
+            assert response.status_code == 401
+            assert response.json() == {
+                "error": "Unauthorized",
+                "message": "Invalid or expired API key.",
+            }
+            assert response.headers["www-authenticate"] == "Bearer"
+
+        mcp_missing = client.post("/mcp/", json={"jsonrpc": "2.0", "id": 1})
+        assert mcp_missing.status_code == 401
+
+        identity = client.get("/v1/auth/context", headers=authorization)
+        assert identity.status_code == 200
+        assert identity.json()["key_prefix"] == material.key_prefix
+        assert "admin" in identity.json()["scopes"]
+
+        created = client.post(
+            "/v1/api-keys",
+            headers=authorization,
+            json={
+                "name": "Template Reader",
+                "environment": "test",
+                "role": None,
+                "scopes": ["templates:read"],
+            },
+        )
+        assert created.status_code == 201
+        limited_authorization = {
+            "Authorization": f"Bearer {created.json()['api_key']}"
+        }
+        assert client.get("/v1/auth/context", headers=limited_authorization).status_code == 200
+        forbidden = client.get("/v1/health", headers=limited_authorization)
+        assert forbidden.status_code == 403
+        assert forbidden.json() == {
+            "error": "Forbidden",
+            "message": "Missing required scope.",
+        }
+
+        mcp_forbidden = client.post(
+            "/mcp/",
+            headers=limited_authorization,
+            json={
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "tools/call",
+                "params": {"name": "health", "arguments": {}},
+            },
+        )
+        assert mcp_forbidden.status_code == 403
+
+        audit_logs = client.get("/v1/audit-logs", headers=authorization)
+        assert audit_logs.status_code == 200
+        entries = audit_logs.json()
+        assert any(
+            entry["endpoint"] == "/v1/auth/context"
+            and entry["api_key_id"] == identity.json()["id"]
+            and entry["response_code"] == 200
+            for entry in entries
+        )
+
+
+def test_http_rate_limit_returns_retry_after(tmp_path: Path) -> None:
+    material = APIKeyService.generate_material("test")
+    application = create_app(
+        security_settings(
+            tmp_path,
+            auth_bootstrap_key_hash=material.key_hash,
+            auth_bootstrap_key_prefix=material.key_prefix,
+            auth_bootstrap_environment="test",
+            auth_default_requests_per_minute=1,
+        )
+    )
+    headers = {"Authorization": f"Bearer {material.api_key}"}
+    with TestClient(application) as client:
+        assert client.get("/v1/auth/context", headers=headers).status_code == 200
+        limited = client.get("/v1/auth/context", headers=headers)
+
+    assert limited.status_code == 429
+    assert limited.json() == {
+        "error": "Rate limit exceeded",
+        "message": "Retry later.",
+    }
+    assert int(limited.headers["retry-after"]) >= 1