Compare commits
3 Commits
295b49bb04
...
17f8036bdc
| Author | SHA1 | Date | |
|---|---|---|---|
| 17f8036bdc | |||
| fcc5885dcf | |||
| 0b1506f860 |
41
tests/test_release_console_security.py
Normal file
41
tests/test_release_console_security.py
Normal file
@@ -0,0 +1,41 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
META_ROOT = Path(__file__).resolve().parents[1]
|
||||
RELEASE_ROOT = META_ROOT / "tools" / "release"
|
||||
if str(RELEASE_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(RELEASE_ROOT))
|
||||
|
||||
from server.app import create_app # noqa: E402
|
||||
|
||||
|
||||
class ReleaseConsoleSecurityTests(unittest.TestCase):
|
||||
def test_api_token_is_accepted_only_from_header(self) -> None:
|
||||
with TestClient(create_app(workspace_root=META_ROOT, token="test-console-token")) as client:
|
||||
query_response = client.get("/api/health?token=test-console-token")
|
||||
header_response = client.get(
|
||||
"/api/health",
|
||||
headers={"X-Release-Console-Token": "test-console-token"},
|
||||
)
|
||||
|
||||
self.assertEqual(query_response.status_code, 401)
|
||||
self.assertEqual(header_response.status_code, 200)
|
||||
|
||||
def test_bootstrap_token_uses_and_clears_url_fragment(self) -> None:
|
||||
launcher = (RELEASE_ROOT / "release-console.py").read_text(encoding="utf-8")
|
||||
webui = (RELEASE_ROOT / "webui" / "index.html").read_text(encoding="utf-8")
|
||||
|
||||
self.assertIn("#token={token}", launcher)
|
||||
self.assertNotIn("?token={token}", launcher)
|
||||
self.assertIn("window.location.hash.slice(1)", webui)
|
||||
self.assertIn("history.replaceState", webui)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -6,12 +6,16 @@ from __future__ import annotations
|
||||
import argparse
|
||||
import importlib
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
import tomllib
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
META_ROOT = Path(__file__).resolve().parents[2]
|
||||
MODULE_NAME_PATTERN = re.compile(
|
||||
r"[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*"
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
@@ -56,8 +60,15 @@ def main() -> int:
|
||||
errors: list[str] = []
|
||||
for repository_name, source_root, manifest_path in manifest_sources:
|
||||
module_name = ".".join(manifest_path.relative_to(source_root).with_suffix("").parts)
|
||||
if MODULE_NAME_PATTERN.fullmatch(module_name) is None:
|
||||
errors.append(
|
||||
f"{repository_name}: manifest path does not map to a safe Python module name: {module_name!r}"
|
||||
)
|
||||
continue
|
||||
try:
|
||||
loaded_module = importlib.import_module(module_name)
|
||||
# Module names are derived from repository-owned manifest paths and
|
||||
# constrained to canonical Python identifiers immediately above.
|
||||
loaded_module = importlib.import_module(module_name) # nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import
|
||||
get_manifest = getattr(loaded_module, "get_manifest")
|
||||
manifest = get_manifest()
|
||||
except Exception as exc: # pragma: no cover - rendered as a check failure
|
||||
|
||||
@@ -180,13 +180,17 @@ run_semgrep() {
|
||||
fi
|
||||
semgrep scan \
|
||||
--metrics=off \
|
||||
--no-git-ignore \
|
||||
--sarif \
|
||||
--output "$report" \
|
||||
--exclude .git \
|
||||
--exclude node_modules \
|
||||
--exclude .venv \
|
||||
--exclude dist \
|
||||
--exclude build \
|
||||
--exclude runtime \
|
||||
--exclude audit-reports \
|
||||
--exclude '**/.*-test-build/**' \
|
||||
"${config_args[@]}" \
|
||||
"${REPOS[@]}"
|
||||
}
|
||||
@@ -222,7 +226,16 @@ run_gitleaks() {
|
||||
gitleaks git \
|
||||
--config "$ROOT/.gitleaks.toml" \
|
||||
--report-format json \
|
||||
--report-path "$REPORTS_DIR/gitleaks-$name.json" \
|
||||
--report-path "$REPORTS_DIR/gitleaks-history-$name.json" \
|
||||
--no-banner \
|
||||
"$repo" || status=1
|
||||
# History scanning does not include new or modified working-tree files.
|
||||
# Scan the directory as well so pre-commit audits cover the exact code
|
||||
# under review, while retaining the history scan above.
|
||||
gitleaks dir \
|
||||
--config "$ROOT/.gitleaks.toml" \
|
||||
--report-format json \
|
||||
--report-path "$REPORTS_DIR/gitleaks-worktree-$name.json" \
|
||||
--no-banner \
|
||||
"$repo" || status=1
|
||||
else
|
||||
@@ -261,7 +274,9 @@ run_pip_audit_manifests() {
|
||||
[[ -f "$repo/requirements.txt" ]] || continue
|
||||
local name
|
||||
name="$(safe_name "$repo")"
|
||||
pip-audit -r "$repo/requirements.txt" --progress-spinner off --format json --output "$REPORTS_DIR/pip-audit-$name.json" || status=1
|
||||
# Requirements may contain project-relative entries such as `.[server]`.
|
||||
# Resolve them from the owning repository instead of the meta-repository.
|
||||
(cd "$repo" && pip-audit -r requirements.txt --progress-spinner off --format json --output "$REPORTS_DIR/pip-audit-$name.json") || status=1
|
||||
done
|
||||
return "$status"
|
||||
}
|
||||
@@ -283,7 +298,11 @@ run_osv_scanner() {
|
||||
local name
|
||||
name="$(safe_name "$repo")"
|
||||
if osv-scanner scan --help >/dev/null 2>&1; then
|
||||
osv-scanner scan -r --format json --output "$REPORTS_DIR/osv-scanner-$name.json" "$repo" || status=1
|
||||
osv-scanner scan -r \
|
||||
--allow-no-lockfiles \
|
||||
--format json \
|
||||
--output-file "$REPORTS_DIR/osv-scanner-$name.json" \
|
||||
"$repo" || status=1
|
||||
else
|
||||
osv-scanner -r --format json --output "$REPORTS_DIR/osv-scanner-$name.json" "$repo" || status=1
|
||||
fi
|
||||
|
||||
@@ -32,7 +32,9 @@ def main() -> int:
|
||||
app = create_app(workspace_root=workspace_root, token=token)
|
||||
url = f"http://{args.host}:{args.port}/"
|
||||
if token:
|
||||
url = f"{url}?token={token}"
|
||||
# URL fragments are never sent in HTTP requests. The WebUI transfers
|
||||
# this one-time bootstrap token to sessionStorage and clears the hash.
|
||||
url = f"{url}#token={token}"
|
||||
print("GovOPlaN release console")
|
||||
print(f" workspace: {workspace_root}")
|
||||
print(f" url: {url}")
|
||||
|
||||
@@ -85,7 +85,7 @@ def create_app(*, workspace_root: Path = DEFAULT_WORKSPACE_ROOT, token: str | No
|
||||
@app.middleware("http")
|
||||
async def require_token(request: Request, call_next): # type: ignore[no-untyped-def]
|
||||
if token and request.url.path.startswith("/api/"):
|
||||
provided = request.headers.get("x-release-console-token") or request.query_params.get("token")
|
||||
provided = request.headers.get("x-release-console-token")
|
||||
if provided != token:
|
||||
return JSONResponse({"detail": "release console token required"}, status_code=401)
|
||||
return await call_next(request)
|
||||
|
||||
@@ -768,11 +768,15 @@
|
||||
</div>
|
||||
|
||||
<script>
|
||||
const params = new URLSearchParams(window.location.search);
|
||||
const token = params.get("token") || sessionStorage.getItem("releaseConsoleToken") || "";
|
||||
const fragmentParams = new URLSearchParams(window.location.hash.slice(1));
|
||||
const fragmentToken = fragmentParams.get("token") || "";
|
||||
const token = fragmentToken || sessionStorage.getItem("releaseConsoleToken") || "";
|
||||
if (token) {
|
||||
sessionStorage.setItem("releaseConsoleToken", token);
|
||||
}
|
||||
if (fragmentToken) {
|
||||
history.replaceState(null, document.title, `${window.location.pathname}${window.location.search}`);
|
||||
}
|
||||
|
||||
const elements = {
|
||||
channel: document.getElementById("channel"),
|
||||
|
||||
Reference in New Issue
Block a user