Harden local viewer and git source inspection

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(

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

src/codealmanac/server/app.py

@@ -5,15 +5,25 @@
 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
 
 
 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,

src/codealmanac/server/assets/viewer/jobs.js

@@ -760,13 +760,27 @@ function inlineMarkdown(text) {
   html = html.replace(/`([^`]+)`/g, (_, code) => `<code>${code}</code>`);
   html = html.replace(/\*\*([^*]+)\*\*/g, "<strong>$1</strong>");
   html = html.replace(/(^|[^*])\*([^*\n]+)\*/g, "$1<em>$2</em>");
-  html = html.replace(
-    /\[([^\]]+)\]\(([^)\s]+)\)/g,
-    (_, label, url) => `<a href="${escapeAttr(url)}" rel="noreferrer">${label}</a>`,
-  );
+  html = html.replace(/\[([^\]]+)\]\(([^)\s]+)\)/g, (match, label, url) => {
+    const href = safeHref(url);
+    if (href === null) return match;
+    return `<a href="${escapeAttr(href)}" rel="noreferrer">${label}</a>`;
+  });
   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,

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

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,15 +119,15 @@ 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})
     other_page = client.get(
         "/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,14 +209,77 @@ 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")
 
     assert response.status_code == 422
     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)

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()