From 54c3260e4acad2cbb17fc20bac68a9954697f7df Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 05:16:39 +0000 Subject: [PATCH 01/16] Fix critical injection paths from untrusted documents Documents ingested by OpenFOIA are untrusted: a hostile agency can craft a FOIA response that breaks out of the context it is later rendered into. - graph_template: escape <, >, &, U+2028/9 as \uXXXX before embedding the graph JSON in an inline `` terminates the script element and everything after it is + parsed as HTML — arbitrary JS execution in the reader's browser. + + These characters never appear in JSON *structure*, only inside string + literals, so replacing them with their ``\\uXXXX`` escapes preserves the + decoded value exactly while making breakout impossible. U+2028/U+2029 are + valid in JSON but are line terminators in JavaScript, so they go too. + """ + return ( + graph_json.replace("&", "\\u0026") + .replace("<", "\\u003c") + .replace(">", "\\u003e") + .replace("
", "\\u2028") + .replace("
", "\\u2029") + ) + + def render(graph_json: str, output_path: Path) -> None: """Render the graph template with embedded data and write to file.""" - html = _TEMPLATE.replace("__GRAPH_DATA__", graph_json) + html = _TEMPLATE.replace("__GRAPH_DATA__", escape_json_for_script(graph_json)) output_path.write_text(html) diff --git a/openfoia/security.py b/openfoia/security.py index f718587..713b067 100644 --- a/openfoia/security.py +++ b/openfoia/security.py @@ -279,9 +279,20 @@ def _can_open_db(db_path: Path, password: str) -> bool: """ try: import pysqlcipher3.dbapi2 as sqlcipher + except ImportError as exc: + # Encryption support missing entirely — a configuration error, not a + # wrong password. Swallowing this would make the duress password + # silently never match, so the decoy would never open under coercion. + raise RuntimeError( + "Cannot verify the database password: pysqlcipher3 is not installed. " + "Install encryption support: openfoia install-extras encryption" + ) from exc + from .db import sqlcipher_key_pragma + + try: conn = sqlcipher.connect(str(db_path)) - conn.execute(f"PRAGMA key='{password}'") + conn.execute(sqlcipher_key_pragma(password)) conn.execute("PRAGMA cipher_compatibility = 4") conn.execute("SELECT count(*) FROM sqlite_master") conn.close() @@ -350,16 +361,25 @@ def seed_decoy_db(db_path: Path, password: str | None = None) -> None: try: import pysqlcipher3.dbapi2 as sqlcipher + from .db import sqlcipher_key_pragma + def _creator(): conn = sqlcipher.connect(str(db_path)) - conn.execute(f"PRAGMA key='{password}'") + conn.execute(sqlcipher_key_pragma(password)) conn.execute("PRAGMA cipher_compatibility = 4") conn.execute("PRAGMA cipher_memory_security = ON") return conn engine = create_engine("sqlite+pysqlite:///", creator=_creator, echo=False) - except ImportError: - engine = create_engine(f"sqlite:///{db_path}", echo=False) + except ImportError as exc: + # Fail closed. Falling back to plain SQLite here would write an + # UNENCRYPTED decoy database while telling the user duress mode + # is configured — the opposite of the promised guarantee. + raise RuntimeError( + "Cannot create the decoy profile: pysqlcipher3 is not installed. " + "A plaintext decoy would contradict the duress-mode guarantee. " + "Install encryption support: openfoia install-extras encryption" + ) from exc else: engine = create_engine(f"sqlite:///{db_path}", echo=False) diff --git a/tests/test_security_agent.py b/tests/test_security_agent.py new file mode 100644 index 0000000..2048ec1 --- /dev/null +++ b/tests/test_security_agent.py @@ -0,0 +1,116 @@ +"""Security regression tests: the LLM agent tool surface. + +Threat model: the agent reads UNTRUSTED document text (extract_entities feeds +document content straight to the model). A prompt-injected document must not +be able to make the agent read arbitrary files off the journalist's disk. +""" + +from __future__ import annotations + +import asyncio + +import pytest + + +@pytest.fixture +def data_dir(tmp_path, monkeypatch): + d = tmp_path / "openfoia-data" + d.mkdir() + monkeypatch.setenv("OPENFOIA_DATA_DIR", str(d)) + return d + + +def _agent(): + from openfoia.agent import OpenFOIAAgent + + return OpenFOIAAgent(db_session=None, config={}) + + +def _run(coro): + return asyncio.run(coro) + + +def test_process_document_rejects_path_outside_data_dir(data_dir, tmp_path): + """The classic prompt-injection target: read the user's SSH key.""" + secret = tmp_path / "id_rsa" + secret.write_text("-----BEGIN OPENSSH PRIVATE KEY-----\n") + + result = _run(_agent().execute_tool("process_document", {"document_path": str(secret)})) + + assert "error" in result + assert "outside" in result["error"].lower() or "not allowed" in result["error"].lower() + # Must not leak the resolved absolute path of the probed file back to the LLM. + assert "BEGIN OPENSSH" not in str(result) + + +def test_process_document_rejects_traversal_escape(data_dir, tmp_path): + """../ escapes out of the data dir must be caught after resolution.""" + secret = tmp_path / "outside.txt" + secret.write_text("sensitive") + + sneaky = str(data_dir / ".." / "outside.txt") + result = _run(_agent().execute_tool("process_document", {"document_path": sneaky})) + + assert "error" in result + assert "outside" in result["error"].lower() or "not allowed" in result["error"].lower() + + +def test_process_document_rejects_symlink_escape(data_dir, tmp_path): + """A symlink inside the data dir pointing out of it must not be followed.""" + secret = tmp_path / "secret.txt" + secret.write_text("sensitive") + link = data_dir / "innocent.pdf" + try: + link.symlink_to(secret) + except (OSError, NotImplementedError): + pytest.skip("symlinks not supported on this platform") + + result = _run(_agent().execute_tool("process_document", {"document_path": str(link)})) + + assert "error" in result + assert "outside" in result["error"].lower() or "not allowed" in result["error"].lower() + + +def test_process_document_reports_missing_file_without_echoing_path(data_dir): + """Non-existent paths inside the data dir are a normal error.""" + missing = data_dir / "nope.pdf" + result = _run(_agent().execute_tool("process_document", {"document_path": str(missing)})) + + assert "error" in result + assert "not found" in result["error"].lower() + + +def test_agent_tool_errors_are_not_raw_exception_text(data_dir): + """Raw exception strings leak absolute paths / DB internals into the LLM.""" + agent = OpenFOIAAgentWithBrokenDB() + result = _run(agent.execute_tool("list_requests", {})) + + assert "error" in result + assert "Traceback" not in result["error"] + assert "/home/" not in result["error"] + + +class OpenFOIAAgentWithBrokenDB: + """Minimal harness: a db session whose query() blows up with a pathy message.""" + + def __new__(cls): + from openfoia.agent import OpenFOIAAgent + + class _BoomDB: + def query(self, *a, **k): + raise RuntimeError("no such table: /home/journalist/.openfoia/data.db is locked") + + return OpenFOIAAgent(db_session=_BoomDB(), config={}) + + +def test_system_prompt_hardens_against_prompt_injection(): + """The agent must be told document content is data, never instructions.""" + from openfoia import agent as agent_mod + + prompt = getattr(agent_mod, "AGENT_SYSTEM_PROMPT", "") + lowered = prompt.lower() + + assert "never" in lowered or "do not" in lowered + assert "instruction" in lowered + # Must explicitly frame document content as untrusted data. + assert "untrusted" in lowered or "not instructions" in lowered diff --git a/tests/test_security_crypto.py b/tests/test_security_crypto.py new file mode 100644 index 0000000..2d556cf --- /dev/null +++ b/tests/test_security_crypto.py @@ -0,0 +1,106 @@ +"""Security regression tests: SQLCipher key handling. + +The passphrase was interpolated into `PRAGMA key='{password}'` with an +f-string. A passphrase containing an apostrophe silently truncated the +effective key (everything after `'--` became a SQL comment), so a long +passphrase could be reduced to a couple of characters -- while still +appearing to work, because create and unlock truncated identically. +""" + +from __future__ import annotations + +import sqlite3 + +import pytest + + +NASTY_PASSPHRASES = [ + "it's a secret", + "abc'--rest-of-my-very-long-passphrase", + "quote'quote'quote", + "trailing'", + "'leading", + "back\\slash", + 'double"quote', + "semi;colon", + "unicode-éè-passphrase", + "normal-passphrase-no-quotes", +] + + +def _key_literal(password: str) -> str: + from openfoia.db import sqlcipher_key_literal + + return sqlcipher_key_literal(password) + + +@pytest.mark.parametrize("password", NASTY_PASSPHRASES) +def test_key_literal_round_trips_through_sqlite_parser(password): + """The escaped literal must decode back to the EXACT passphrase. + + This is the real regression: we hand the literal to SQLite's own parser + and require the full passphrase back. Truncation shows up immediately. + """ + literal = _key_literal(password) + + conn = sqlite3.connect(":memory:") + try: + (value,) = conn.execute(f"SELECT {literal}").fetchone() + finally: + conn.close() + + assert value == password + + +def test_key_literal_doubles_single_quotes(): + assert _key_literal("it's") == "'it''s'" + + +def test_key_literal_is_not_truncated_by_comment_injection(): + """`abc'--tail` must NOT reduce to `abc`.""" + password = "abc'--tail" + literal = _key_literal(password) + + conn = sqlite3.connect(":memory:") + try: + (value,) = conn.execute(f"SELECT {literal}").fetchone() + finally: + conn.close() + + assert value != "abc" + assert value == password + + +@pytest.mark.parametrize("password", NASTY_PASSPHRASES) +def test_key_pragma_statement_is_single_statement(password): + """The built PRAGMA must not be splittable into extra statements.""" + from openfoia.db import sqlcipher_key_pragma + + stmt = sqlcipher_key_pragma(password) + + assert stmt.startswith("PRAGMA key = '") + assert stmt.endswith("'") + # sqlite3.complete_statement only returns True for one finished statement. + assert sqlite3.complete_statement(stmt + ";") + + +def test_no_passphrase_interpolated_inside_a_quoted_pragma_literal(): + """No `PRAGMA key='{password}'` may remain anywhere in the package. + + The dangerous shape is interpolation *inside* the quotes, which is what + lets an apostrophe close the literal early. The helper's own + `PRAGMA key = {literal}` (quotes supplied by the escaper) is fine. + """ + import re + from pathlib import Path + + import openfoia + + pkg = Path(openfoia.__file__).parent + offenders = [] + for path in pkg.rglob("*.py"): + for lineno, line in enumerate(path.read_text().splitlines(), 1): + if re.search(r"""PRAGMA\s+key\s*=?\s*['"]\{""", line): + offenders.append(f"{path.relative_to(pkg)}:{lineno}") + + assert offenders == [], f"f-string PRAGMA key interpolation remains: {offenders}" diff --git a/tests/test_security_injection.py b/tests/test_security_injection.py new file mode 100644 index 0000000..33a17c2 --- /dev/null +++ b/tests/test_security_injection.py @@ -0,0 +1,159 @@ +"""Security regression tests: injection from untrusted document content. + +Threat model: documents ingested by OpenFOIA are UNTRUSTED. A hostile agency +can return a FOIA response whose text is crafted to break out of the context +it is later rendered into. These tests pin the escaping/sandboxing that stops +untrusted content from becoming executable code. +""" + +from __future__ import annotations + +import json + +import pytest + + +# --------------------------------------------------------------------------- +# Entity graph HTML — untrusted document text is embedded in an inline " +) + + +def _render_graph(tmp_path, graph_data): + from openfoia.graph_template import render + + out = tmp_path / "graph.html" + render(json.dumps(graph_data), out) + return out.read_text() + + +def test_graph_render_escapes_script_close_in_document_text(tmp_path): + """A inside document text must not terminate the data block.""" + html = _render_graph( + tmp_path, + {"nodes": [], "links": [], "documents": [{"id": "d1", "text": XSS_PAYLOAD}]}, + ) + + # The literal breakout sequence must not survive into the page. + assert "" in html.lower() # the template's own closing tags remain + assert html.lower().count("", "", html, flags=re.DOTALL | re.IGNORECASE) + html = re.sub(r"]*>.*?", "", html, flags=re.DOTALL | re.IGNORECASE) + # Drop remote subresource elements. + html = re.sub( + r"<(?:img|iframe|frame|embed|object|video|audio|source|track)\b[^>]*/?>", + "", + html, + flags=re.IGNORECASE, + ) + # Drop elements that fetch (stylesheets, preloads, prefetch...). + html = re.sub(r"]*>", "", html, flags=re.IGNORECASE) + # Drop inline event handlers (onload=, onerror=, ...). + html = re.sub(r"\son[a-z]+\s*=\s*(?:\"[^\"]*\"|'[^']*'|[^\s>]+)", "", html, flags=re.IGNORECASE) + # Neutralise any remaining external url() references in inline styles. + html = re.sub(r"url\(\s*['\"]?https?://[^)]*\)", "url(about:blank)", html, flags=re.IGNORECASE) + return html + + def _extract_content(html: str) -> tuple[str, str]: """Extract main text content and title from HTML. @@ -307,9 +333,13 @@ async def archive_url( dest_dir = storage / doc_id[:2] / doc_id[2:4] dest_dir.mkdir(parents=True, exist_ok=True) - # Save raw HTML + # Save the SANITIZED HTML, never the raw page. Persisting raw HTML kept + # analytics scripts and tracking pixels live in the archive: opening the + # saved page later would fire those beacons and tell the tracker (and the + # journalist's ISP) that the page was re-read, and from where. + safe_html = _sanitize_for_archive(result.raw_html) html_path = dest_dir / f"{doc_id}.html" - html_path.write_text(result.raw_html, encoding="utf-8") + html_path.write_text(safe_html, encoding="utf-8") # Save extracted text text_path = dest_dir / f"{doc_id}.txt" diff --git a/openfoia/records/base.py b/openfoia/records/base.py index c8aa9b3..f62f8f0 100644 --- a/openfoia/records/base.py +++ b/openfoia/records/base.py @@ -2,9 +2,88 @@ from __future__ import annotations +import ipaddress +import posixpath +import re +import socket from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Any +from urllib.parse import unquote, urlparse + + +def validate_download_url(url: str) -> str: + """Return *url* if it is safe to fetch, else raise ValueError. + + File URLs in third-party API responses (MuckRock ``ffile``, + DocumentCloud ``asset_url``) are attacker-influenced: a compromised or + MITM'd upstream can point them at ``file:///etc/passwd`` or cloud + metadata endpoints, and the body is written straight to disk. + """ + parsed = urlparse(url) + + if parsed.scheme != "https": + raise ValueError(f"Refusing to download over {parsed.scheme or 'missing'} scheme: {url!r}") + + host = parsed.hostname + if not host: + raise ValueError(f"Refusing to download from a URL with no host: {url!r}") + + # Block obvious internal targets without doing a network lookup. + if host in ("localhost", "localhost.localdomain") or host.endswith(".localhost"): + raise ValueError(f"Refusing to download from a loopback host: {url!r}") + + try: + ip = ipaddress.ip_address(host) + except ValueError: + ip = None + if ip is not None and ( + ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast + ): + raise ValueError(f"Refusing to download from a non-public address: {url!r}") + + return url + + +def resolves_to_public_address(host: str) -> bool: + """Best-effort DNS check that *host* is not an internal address. + + Separate from :func:`validate_download_url` so callers can opt in; DNS + resolution is itself a network call. + """ + try: + infos = socket.getaddrinfo(host, None) + except OSError: + return False + + for info in infos: + addr = info[4][0] + try: + ip = ipaddress.ip_address(addr) + except ValueError: + return False + if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: + return False + return True + + +def safe_download_filename(url: str, default: str = "download") -> str: + """Derive a safe basename from *url*. + + ``url.split("/")[-1]`` accepted percent-encoded traversal, empty names and + dotfiles straight into the output directory. + """ + path = unquote(urlparse(url).path or "") + name = posixpath.basename(posixpath.normpath(path)) + + # normpath can still yield traversal markers for pathological input. + name = name.replace("\\", "/").split("/")[-1] + name = name.strip().lstrip(".") + name = re.sub(r"[^A-Za-z0-9._-]", "_", name) + + if not name or name in (".", ".."): + return default + return name[:255] @dataclass diff --git a/openfoia/records/documentcloud.py b/openfoia/records/documentcloud.py index d327163..31bfbd9 100644 --- a/openfoia/records/documentcloud.py +++ b/openfoia/records/documentcloud.py @@ -21,7 +21,7 @@ import httpx -from .base import RecordAdapter, RecordEntity, SearchResult +from .base import RecordAdapter, RecordEntity, SearchResult, validate_download_url logger = logging.getLogger(__name__) @@ -213,6 +213,9 @@ async def fetch(self, identifier: str, **kwargs: Any) -> RecordEntity | None: if text_url: txt_url = f"{text_url}documents/{doc_id}/{slug}.txt" try: + # asset_url comes from the API response, so it is + # attacker-influenced if the upstream is compromised. + validate_download_url(txt_url) txt_resp = await client.get(txt_url, timeout=15) if txt_resp.status_code == 200: full_text = txt_resp.text diff --git a/openfoia/records/muckrock.py b/openfoia/records/muckrock.py index a439852..48ccd45 100644 --- a/openfoia/records/muckrock.py +++ b/openfoia/records/muckrock.py @@ -15,7 +15,16 @@ import httpx -from .base import RecordAdapter, RecordEntity, SearchResult +from .base import ( + RecordAdapter, + RecordEntity, + SearchResult, + safe_download_filename, + validate_download_url, +) + +#: Cap on a single downloaded response file (100 MiB), matching ingest. +MAX_DOWNLOAD_BYTES = 100 * 1024 * 1024 logger = logging.getLogger(__name__) @@ -358,18 +367,25 @@ async def download_files( if not files: return [] - async with httpx.AsyncClient(timeout=60, follow_redirects=True) as client: + # follow_redirects is off: a redirect is a second, unvalidated URL and + # would bypass the scheme/host checks below. + async with httpx.AsyncClient(timeout=60, follow_redirects=False) as client: for f in files: url = f.get("url") if not url: continue try: + validate_download_url(url) resp = await client.get(url) resp.raise_for_status() - # Extract filename from URL - filename = url.split("/")[-1] + if len(resp.content) > MAX_DOWNLOAD_BYTES: + raise ValueError( + f"Refusing file over {MAX_DOWNLOAD_BYTES} bytes: {len(resp.content)}" + ) + + filename = safe_download_filename(url) dest = output_path / filename dest.write_bytes(resp.content) diff --git a/openfoia/security.py b/openfoia/security.py index 713b067..5bc0100 100644 --- a/openfoia/security.py +++ b/openfoia/security.py @@ -14,6 +14,7 @@ from __future__ import annotations import os +import shutil import tempfile from pathlib import Path @@ -221,6 +222,13 @@ def print_ssd_warning() -> None: _PROFILE_SLOTS = ["profile_0.db", "profile_1.db"] +def _has_sqlcipher() -> bool: + """Return True if SQLCipher support is importable.""" + from .db import has_sqlcipher + + return has_sqlcipher() + + def get_profile_paths() -> list[Path]: """Return paths for both profile slots.""" from .db import get_data_dir @@ -229,14 +237,43 @@ def get_profile_paths() -> list[Path]: return [data_dir / name for name in _PROFILE_SLOTS] +def real_profile_path() -> Path: + """Path of the slot holding the real database.""" + return get_profile_paths()[0] + + +def duress_mode_active() -> bool: + """True once the real database has been migrated into a profile slot.""" + return real_profile_path().exists() + + def setup_duress_mode(duress_password: str) -> Path: """Create and seed a decoy profile encrypted with the duress password. Returns the path to the decoy database. No password hash is stored anywhere — the password is verified by attempting to open the DB. + + The real database is moved into slot 0 at the same time. Leaving it as + ``data.db`` next to a file named ``profile_1.db`` would tell any examiner + both that duress mode is configured and which file is the decoy — the + opposite of the "opaque filenames" the design promises. """ from .db import get_data_dir + if not _has_sqlcipher(): + # A plaintext decoy contradicts the guarantee. Fail closed. + raise RuntimeError( + "Duress mode requires database encryption, but pysqlcipher3 is not " + "installed. A plaintext decoy would provide no protection. " + "Install encryption support: openfoia install-extras encryption" + ) + + # Migrate the real database into slot 0 so both slots look alike. + legacy_path = get_data_dir() / "data.db" + real_path = real_profile_path() + if legacy_path.exists() and not real_path.exists(): + shutil.move(str(legacy_path), str(real_path)) + # Use the second slot for the decoy decoy_path = get_data_dir() / _PROFILE_SLOTS[1] diff --git a/openfoia/server.py b/openfoia/server.py index 2d99094..9eefa14 100644 --- a/openfoia/server.py +++ b/openfoia/server.py @@ -14,6 +14,8 @@ from fastapi import FastAPI, Request, HTTPException, Depends, Query, UploadFile, File from fastapi.responses import HTMLResponse from fastapi.middleware.cors import CORSMiddleware +from fastapi.staticfiles import StaticFiles +from starlette.middleware.trustedhost import TrustedHostMiddleware from pydantic import BaseModel from sqlalchemy import func @@ -27,6 +29,13 @@ class CreateRequestBody(BaseModel): expedited: bool = False +def _default_data_dir() -> Path: + """Resolve the data dir the same way the rest of the toolkit does.""" + from .db import get_data_dir + + return get_data_dir() + + def create_app(token: str, data_dir: Path | None = None) -> FastAPI: """Create the FastAPI application with token authentication.""" @@ -40,8 +49,21 @@ def create_app(token: str, data_dir: Path | None = None) -> FastAPI: # Store token and data directory in app state app.state.auth_token = token - app.state.data_dir = data_dir or Path.home() / ".openfoia" - app.state.data_dir.mkdir(parents=True, exist_ok=True) + from .db import _ensure_private_dir + + app.state.data_dir = _ensure_private_dir(Path(data_dir) if data_dir else _default_data_dir()) + + # Serve the vendored stylesheet from disk — no CDN, works air-gapped. + static_dir = Path(__file__).parent / "static" + if static_dir.is_dir(): + app.mount("/static", StaticFiles(directory=str(static_dir)), name="static") + + # Reject requests that did not arrive addressed to loopback (DNS rebinding). + # Starlette compares the Host header with the port stripped. + app.add_middleware( + TrustedHostMiddleware, + allowed_hosts=["127.0.0.1", "localhost", "::1", "[::1]"], + ) # CORS - only allow localhost app.add_middleware( @@ -53,6 +75,32 @@ def create_app(token: str, data_dir: Path | None = None) -> FastAPI: allow_headers=["*"], ) + @app.middleware("http") + async def security_headers(request: Request, call_next): + """Structurally forbid the page from talking to anything off-machine. + + Defence in depth behind the escaping fixes: even if untrusted document + text did reach the DOM as markup, `connect-src 'self'` blocks the + exfiltration step, and `default-src 'self'` blocks remote subresources. + """ + response = await call_next(request) + response.headers["Content-Security-Policy"] = ( + "default-src 'self'; " + "script-src 'self' 'unsafe-inline'; " + "style-src 'self' 'unsafe-inline'; " + "img-src 'self' data:; " + "connect-src 'self'; " + "form-action 'self'; " + "frame-ancestors 'none'; " + "base-uri 'none'" + ) + # Keep the ?token= out of outbound Referer headers and shared caches. + response.headers["Referrer-Policy"] = "no-referrer" + response.headers["Cache-Control"] = "no-store" + response.headers["X-Content-Type-Options"] = "nosniff" + response.headers["X-Frame-Options"] = "DENY" + return response + # Token verification dependency async def verify_token( request: Request, @@ -60,7 +108,8 @@ async def verify_token( ): # Check query param first, then cookie auth_token = token or request.cookies.get("openfoia_token") - if auth_token != app.state.auth_token: + # Constant-time: a plain != leaks the token prefix through timing. + if not secrets.compare_digest(auth_token or "", app.state.auth_token): raise HTTPException(status_code=401, detail="Invalid or missing token") return auth_token @@ -544,10 +593,10 @@ def get_index_html() -> str: OpenFOIA - - + +