diff --git a/src/codealmanac/cli/dispatch/serve.py b/src/codealmanac/cli/dispatch/serve.py index 349de3f7..86e8a977 100644 --- a/src/codealmanac/cli/dispatch/serve.py +++ b/src/codealmanac/cli/dispatch/serve.py @@ -31,7 +31,7 @@ def run_serve(app: CodeAlmanac, args: argparse.Namespace) -> int: from codealmanac.server.app import create_server_app - server_app = create_server_app(app, Path.cwd(), args.wiki) + server_app = create_server_app(app, Path.cwd(), args.wiki, host=args.host) url = f"http://{args.host}:{args.port}" server = uvicorn.Server( uvicorn.Config( diff --git a/src/codealmanac/integrations/sources/git/adapter.py b/src/codealmanac/integrations/sources/git/adapter.py index 83b1018c..a10d741e 100644 --- a/src/codealmanac/integrations/sources/git/adapter.py +++ b/src/codealmanac/integrations/sources/git/adapter.py @@ -132,4 +132,10 @@ def run_git(self, cwd: Path, args: tuple[str, ...]) -> str: def require_revision_range(ref: SourceRef) -> str: if ref.revision_range is None or ref.revision_range.strip() == "": raise ExecutionFailed(f"Git source missing revision range: {ref.identity}") + # A leading dash makes git read the range as an option, so a source like + # `--output=~/.zshrc` would turn a read-only inspection into a write. + if ref.revision_range.startswith("-"): + raise ExecutionFailed( + f"Git source revision range must not start with '-': {ref.identity}" + ) return ref.revision_range diff --git a/src/codealmanac/server/app.py b/src/codealmanac/server/app.py index 4cdf1abd..639d7509 100644 --- a/src/codealmanac/server/app.py +++ b/src/codealmanac/server/app.py @@ -5,6 +5,7 @@ from codealmanac.app import CodeAlmanac from codealmanac.server.api_routes import ServerApiContext, register_api_routes from codealmanac.server.errors import register_error_handlers +from codealmanac.server.security import DEFAULT_BIND_HOST, register_security from codealmanac.server.static_routes import register_static_routes @@ -12,8 +13,17 @@ def create_server_app( codealmanac: CodeAlmanac, cwd: Path, wiki: str | None = None, + host: str = DEFAULT_BIND_HOST, ) -> FastAPI: - server = FastAPI(title="CodeAlmanac Local Viewer") + # The viewer serves local wiki content without authentication; the + # generated docs and schema endpoints add reachable surface nobody reads. + server = FastAPI( + title="CodeAlmanac Local Viewer", + docs_url=None, + redoc_url=None, + openapi_url=None, + ) + register_security(server, host) register_error_handlers(server) register_api_routes( server, diff --git a/src/codealmanac/server/assets/viewer/jobs.js b/src/codealmanac/server/assets/viewer/jobs.js index 8c1b4363..ac9f9e54 100644 --- a/src/codealmanac/server/assets/viewer/jobs.js +++ b/src/codealmanac/server/assets/viewer/jobs.js @@ -760,13 +760,27 @@ function inlineMarkdown(text) { html = html.replace(/`([^`]+)`/g, (_, code) => `${code}`); html = html.replace(/\*\*([^*]+)\*\*/g, "$1"); html = html.replace(/(^|[^*])\*([^*\n]+)\*/g, "$1$2"); - html = html.replace( - /\[([^\]]+)\]\(([^)\s]+)\)/g, - (_, label, url) => `${label}`, - ); + html = html.replace(/\[([^\]]+)\]\(([^)\s]+)\)/g, (match, label, url) => { + const href = safeHref(url); + if (href === null) return match; + return `${label}`; + }); return html; } +const SAFE_LINK_SCHEMES = new Set(["http:", "https:", "mailto:"]); + +function safeHref(url) { + const raw = String(url).trim(); + if (raw.startsWith("#") || raw.startsWith("/")) return raw; + try { + const parsed = new URL(raw, window.location.origin); + return SAFE_LINK_SCHEMES.has(parsed.protocol) ? raw : null; + } catch { + return null; + } +} + function escapeHtml(value) { return String(value).replace( /[&<>"]/g, diff --git a/src/codealmanac/server/security.py b/src/codealmanac/server/security.py new file mode 100644 index 00000000..f3ab5903 --- /dev/null +++ b/src/codealmanac/server/security.py @@ -0,0 +1,60 @@ +from collections.abc import Awaitable, Callable + +from fastapi import FastAPI, Request +from fastapi.responses import Response +from starlette.middleware.trustedhost import TrustedHostMiddleware + +LOOPBACK_HOSTS = ("localhost", "127.0.0.1", "::1") +DEFAULT_BIND_HOST = "127.0.0.1" + +CONTENT_SECURITY_POLICY = "; ".join( + ( + "default-src 'self'", + "img-src 'self' data: https:", + "style-src 'self'", + "script-src 'self'", + "connect-src 'self'", + "object-src 'none'", + "base-uri 'none'", + "form-action 'none'", + "frame-ancestors 'none'", + ) +) + +SECURITY_HEADERS = { + "Content-Security-Policy": CONTENT_SECURITY_POLICY, + "Referrer-Policy": "no-referrer", + "X-Content-Type-Options": "nosniff", + "X-Frame-Options": "DENY", +} + + +def register_security(server: FastAPI, host: str) -> None: + """Harden the local viewer against browser-driven access from other sites. + + The viewer has no authentication because it only serves the wiki of the + repo it was started in. That makes the browser the attack surface: a page + the user visits can point a hostname it controls at 127.0.0.1 and read the + API through the user's own browser. Restricting the accepted Host header to + loopback names removes that rebinding path, and the response headers stop + wiki markdown from pulling in or leaking to third-party origins. + """ + if is_loopback(host): + server.add_middleware( + TrustedHostMiddleware, + allowed_hosts=[*LOOPBACK_HOSTS, "[::1]"], + ) + + @server.middleware("http") + async def apply_security_headers( + request: Request, + call_next: Callable[[Request], Awaitable[Response]], + ) -> Response: + response = await call_next(request) + for header, value in SECURITY_HEADERS.items(): + response.headers.setdefault(header, value) + return response + + +def is_loopback(host: str) -> bool: + return host.strip().strip("[]").lower() in LOOPBACK_HOSTS diff --git a/tests/test_server.py b/tests/test_server.py index 8a50708b..bde3a53a 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,5 +1,6 @@ from pathlib import Path +from fastapi import FastAPI from fastapi.testclient import TestClient from codealmanac.app import CodeAlmanac @@ -19,7 +20,7 @@ def test_server_serves_static_assets_and_viewer_api( viewer_repo: tuple[Path, CodeAlmanac], ): repo, app = viewer_repo - client = TestClient(create_server_app(app, repo)) + client = viewer_client(create_server_app(app, repo)) index = client.get("/") overview = client.get("/api/overview") @@ -82,7 +83,7 @@ def test_server_serves_jobs_api_with_readable_steps( ): repo, app = viewer_repo record = create_server_run(repo, app) - client = TestClient(create_server_app(app, repo)) + client = viewer_client(create_server_app(app, repo)) jobs = client.get("/api/jobs") detail = client.get(f"/api/jobs/{record.run_id}") @@ -118,7 +119,7 @@ def test_server_viewer_api_switches_between_registered_wikis( Tracks operational decisions. """, ) - client = TestClient(create_server_app(app, repo)) + client = viewer_client(create_server_app(app, repo)) overview = client.get("/api/overview") other_overview = client.get("/api/overview", params={"wiki": other.name}) @@ -126,7 +127,7 @@ def test_server_viewer_api_switches_between_registered_wikis( "/api/page/ops-note", params={"wiki": other.name}, ) - locked_client = TestClient(create_server_app(app, repo, other.name)) + locked_client = viewer_client(create_server_app(app, repo, other.name)) locked_overview = locked_client.get("/api/overview") locked_page = locked_client.get("/api/page/ops-note") @@ -154,7 +155,7 @@ def test_server_maps_product_errors_to_http_statuses( viewer_repo: tuple[Path, CodeAlmanac], ): repo, app = viewer_repo - client = TestClient(create_server_app(app, repo)) + client = viewer_client(create_server_app(app, repo)) response = client.get("/api/page/missing") @@ -166,7 +167,7 @@ def test_server_maps_request_validation_errors_to_http_statuses( viewer_repo: tuple[Path, CodeAlmanac], ): repo, app = viewer_repo - client = TestClient(create_server_app(app, repo)) + client = viewer_client(create_server_app(app, repo)) response = client.get("/api/search", params={"limit": "-1"}) @@ -178,7 +179,7 @@ def test_server_rejects_invalid_file_reference_paths( viewer_repo: tuple[Path, CodeAlmanac], ): repo, app = viewer_repo - client = TestClient(create_server_app(app, repo)) + client = viewer_client(create_server_app(app, repo)) response = client.get("/api/file", params={"path": "../secret.txt"}) @@ -190,7 +191,7 @@ def test_server_rejects_invalid_static_asset_paths( viewer_repo: tuple[Path, CodeAlmanac], ): repo, app = viewer_repo - client = TestClient(create_server_app(app, repo)) + client = viewer_client(create_server_app(app, repo)) traversal = client.get("/assets/%2E%2E/app.js") missing = client.get("/assets/viewer/missing.js") @@ -208,7 +209,7 @@ def test_server_rejects_path_shaped_job_ids( viewer_repo: tuple[Path, CodeAlmanac], ): repo, app = viewer_repo - client = TestClient(create_server_app(app, repo)) + client = viewer_client(create_server_app(app, repo)) response = client.get("/api/jobs/..secret") @@ -216,6 +217,69 @@ def test_server_rejects_path_shaped_job_ids( assert response.json()["detail"]["code"] == "validation_failed" +def test_server_sends_security_headers_and_hides_generated_api_docs( + viewer_repo: tuple[Path, CodeAlmanac], +): + repo, app = viewer_repo + client = viewer_client(create_server_app(app, repo)) + + index = client.get("/") + docs = client.get("/docs") + schema = client.get("/openapi.json") + + assert index.headers["x-content-type-options"] == "nosniff" + assert index.headers["x-frame-options"] == "DENY" + assert "frame-ancestors 'none'" in index.headers["content-security-policy"] + assert "default-src 'self'" in index.headers["content-security-policy"] + # Both fall through to the viewer shell instead of exposing generated docs. + assert "swagger" not in docs.text.lower() + assert "openapi" not in schema.text.lower() + + +def test_server_rejects_non_loopback_host_headers_when_bound_to_loopback( + viewer_repo: tuple[Path, CodeAlmanac], +): + repo, app = viewer_repo + client = viewer_client(create_server_app(app, repo)) + + rebound = client.get("/api/overview", headers={"host": "wiki.attacker.example"}) + loopback = client.get("/api/overview", headers={"host": "localhost:3927"}) + + assert rebound.status_code == 400 + assert loopback.status_code == 200 + + +def test_server_accepts_any_host_when_bound_beyond_loopback( + viewer_repo: tuple[Path, CodeAlmanac], +): + repo, app = viewer_repo + client = viewer_client(create_server_app(app, repo, host="0.0.0.0")) + + response = client.get("/api/overview", headers={"host": "workstation.local"}) + + assert response.status_code == 200 + + +def test_server_job_markdown_only_links_safe_url_schemes( + viewer_repo: tuple[Path, CodeAlmanac], +): + repo, app = viewer_repo + client = viewer_client(create_server_app(app, repo)) + + jobs_module = client.get("/assets/viewer/jobs.js") + + assert "safeHref" in jobs_module.text + assert 'SAFE_LINK_SCHEMES = new Set(["http:", "https:", "mailto:"])' in ( + jobs_module.text + ) + + +def viewer_client(server: FastAPI) -> TestClient: + # The viewer answers loopback Host headers only, matching what a browser on + # the same machine sends to the served address. + return TestClient(server, base_url="http://127.0.0.1:3927") + + def write_server_page(repo: Path, name: str, body: str) -> None: path = repo / "almanac" / name path.parent.mkdir(parents=True, exist_ok=True) diff --git a/tests/test_sources_service.py b/tests/test_sources_service.py index 9698684b..d01bc8b0 100644 --- a/tests/test_sources_service.py +++ b/tests/test_sources_service.py @@ -4,7 +4,7 @@ from pydantic import ValidationError from codealmanac.app import create_app -from codealmanac.core.errors import ValidationFailed +from codealmanac.core.errors import ExecutionFailed, ValidationFailed from codealmanac.integrations.command import CommandResult from codealmanac.integrations.sources.git import GitSourceRuntimeAdapter from codealmanac.services.sources.models import ( @@ -163,6 +163,24 @@ def test_sources_runtime_uses_git_adapter_for_diff_and_range(tmp_path: Path): assert ("git", ("status", "--short"), tmp_path) in runner.calls +def test_sources_reject_git_revision_ranges_that_read_as_git_options(tmp_path: Path): + runner = FakeGitRunner() + app = create_app(source_runtime_adapters=(GitSourceRuntimeAdapter(runner),)) + (option_range,) = app.sources.resolve( + ResolveSourcesRequest( + cwd=tmp_path, + inputs=("git:range:--output=escaped.txt",), + ) + ) + + with pytest.raises(ExecutionFailed): + app.sources.inspect_runtime( + InspectSourceRuntimeRequest(cwd=tmp_path, ref=option_range.ref) + ) + + assert runner.calls == [] + + def test_sources_reject_malformed_source_refs(tmp_path: Path): app = create_app()