fix(mcp): real cited bibkeys in wiki_read + constant-time token compare
- C-3 (MED): wiki_read returned 'sources' = [concept_slug] (a placeholder), not the cited sources. Now parses the page's inline [BibKey #loc] citations and returns the actual set of cited bibkeys. - S-2 (LOW): _TokenAuth compared the Bearer header with != (length/equality timing side-channel). Use secrets.compare_digest for constant-time comparison. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -24,6 +24,7 @@ Transport
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import secrets
|
||||
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
from mcp.server.transport_security import TransportSecuritySettings
|
||||
@@ -121,7 +122,6 @@ def wiki_read(concept_slug: str) -> dict[str, object]:
|
||||
directory or the requested page does not exist.
|
||||
"""
|
||||
try:
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from codex.config import get_settings
|
||||
@@ -135,16 +135,12 @@ def wiki_read(concept_slug: str) -> dict[str, object]:
|
||||
|
||||
markdown = page_path.read_text(encoding="utf-8")
|
||||
|
||||
# Collect cited bibkeys from the compile state
|
||||
state_path = wiki_dir / ".compile-state.json"
|
||||
sources: list[str] = []
|
||||
if state_path.exists():
|
||||
try:
|
||||
state: dict[str, str] = json.loads(state_path.read_text(encoding="utf-8"))
|
||||
if concept_slug in state:
|
||||
sources = [concept_slug]
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
# Collect the actual cited bibkeys from the page's inline [BibKey #loc]
|
||||
# citations (audit C-3: this previously returned the concept slug itself,
|
||||
# not the cited sources).
|
||||
from codex.wiki import _parse_claims
|
||||
|
||||
sources = sorted({claim.bibkey for claim in _parse_claims(markdown)})
|
||||
|
||||
return {"markdown": markdown, "sources": sources}
|
||||
except Exception as exc: # noqa: BLE001
|
||||
@@ -337,7 +333,9 @@ def main() -> None:
|
||||
self, request: Request, call_next: RequestResponseEndpoint
|
||||
) -> Response:
|
||||
auth = request.headers.get("Authorization", "")
|
||||
if auth != f"Bearer {token}":
|
||||
# Constant-time comparison to avoid a token-length/equality timing
|
||||
# side-channel (audit S-2).
|
||||
if not secrets.compare_digest(auth, f"Bearer {token}"):
|
||||
return Response("Unauthorized", status_code=401)
|
||||
return await call_next(request)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user