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