Add VedAstro secret and renderer audit tools

This commit is contained in:
732642856
2026-07-17 00:23:31 +08:00
parent 92e44737dd
commit 13db4aaf7c
4 changed files with 178 additions and 0 deletions
+64
View File
@@ -0,0 +1,64 @@
"""Configure VedAstro local secret without echoing it.
Run manually from a trusted terminal. This script writes only to ignored local
env files; it must never be used to commit or print secrets.
"""
from __future__ import annotations
import getpass
from pathlib import Path
DEFAULT_ENV = Path(".env.local")
DEFAULT_ENDPOINT = "https://api.vedastro.org/api"
def update_env_text(text: str, updates: dict[str, str]) -> str:
lines = text.splitlines()
seen: set[str] = set()
output: list[str] = []
for line in lines:
stripped = line.strip()
if not stripped or stripped.startswith("#") or "=" not in line:
output.append(line)
continue
key, _value = line.split("=", 1)
key = key.strip()
if key in updates:
output.append(f"{key}={updates[key]}")
seen.add(key)
else:
output.append(line)
for key, value in updates.items():
if key not in seen:
output.append(f"{key}={value}")
return "\n".join(output).rstrip() + "\n"
def write_env(path: Path, updates: dict[str, str]) -> None:
existing = path.read_text(encoding="utf-8") if path.exists() else ""
path.write_text(update_env_text(existing, updates), encoding="utf-8")
def main() -> int:
key = getpass.getpass("VedAstro API key (hidden): ").strip()
if not key:
print("No key entered; nothing changed.")
return 1
endpoint = input(f"VedAstro endpoint [{DEFAULT_ENDPOINT}]: ").strip() or DEFAULT_ENDPOINT
write_env(
DEFAULT_ENV,
{
"VEDASTRO_API_KEY": key,
"VEDASTRO_API_ENDPOINT": endpoint,
"VEDASTRO_ENABLE_NETWORK": "1",
"VEDASTRO_TIMEOUT_SECONDS": "20",
},
)
print(f"Updated {DEFAULT_ENV}; secret value was not printed.")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+87
View File
@@ -0,0 +1,87 @@
#!/usr/bin/env python3
"""Run an isolated Chromium proof that report rendering cannot fetch external resources."""
from __future__ import annotations
import json
import argparse
import tempfile
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
try:
from report_builder import is_allowed_report_resource_url
except ImportError:
from scripts.report_builder import is_allowed_report_resource_url
class _ProbeHandler(BaseHTTPRequestHandler):
requests = 0
def do_GET(self) -> None: # noqa: N802
type(self).requests += 1
self.send_response(200)
self.end_headers()
def log_message(self, _format: str, *_args: Any) -> None:
return
def run_poc() -> dict[str, Any]:
try:
from playwright.sync_api import sync_playwright
except ImportError:
return {"scope": "report_renderer_isolation_poc", "status": "blocked", "reason": "playwright_python_missing"}
server = ThreadingHTTPServer(("127.0.0.1", 0), _ProbeHandler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
secret = root / "secret.txt"
secret.write_text("must-not-load", encoding="utf-8")
html = root / "report.html"
remote_url = f"http://127.0.0.1:{server.server_port}/probe"
html.write_text(
f'<img src="{remote_url}"><img src="{secret.as_uri()}"><p>report</p>',
encoding="utf-8",
)
report_url = html.as_uri()
blocked: list[str] = []
try:
with sync_playwright() as playwright:
browser = playwright.chromium.launch(headless=True)
context = browser.new_context(java_script_enabled=False)
page = context.new_page()
page.route(
"**/*",
lambda route: route.continue_()
if is_allowed_report_resource_url(route.request.url, report_url=report_url)
else (blocked.append(route.request.url), route.abort())[1],
)
page.goto(report_url, wait_until="networkidle")
context.close()
browser.close()
except Exception as exc: # Browser binary/startup is an environment boundary.
return {"scope": "report_renderer_isolation_poc", "status": "blocked", "reason": f"chromium_unavailable:{type(exc).__name__}"}
return {
"scope": "report_renderer_isolation_poc",
"status": "pass" if _ProbeHandler.requests == 0 and len(blocked) >= 2 else "fail",
"http_probe_requests": _ProbeHandler.requests,
"blocked_resource_count": len(blocked),
"blocked_schemes": sorted({url.split(":", 1)[0] for url in blocked}),
}
finally:
server.shutdown()
server.server_close()
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--strict", action="store_true", help="Return nonzero unless the isolation probe passes.")
args = parser.parse_args()
result = run_poc()
print(json.dumps(result, ensure_ascii=False, sort_keys=True))
raise SystemExit(0 if not args.strict or result["status"] == "pass" else 1)
+18
View File
@@ -0,0 +1,18 @@
from scripts.configure_vedastro_secret import update_env_text
def test_update_env_text_adds_and_replaces_vedastro_settings():
text = "VEDASTRO_API_ENDPOINT=https://old.example/api\nOTHER=value\n"
updated = update_env_text(
text,
{
"VEDASTRO_API_KEY": "sample-secret",
"VEDASTRO_API_ENDPOINT": "https://api.vedastro.org/api",
"VEDASTRO_ENABLE_NETWORK": "1",
},
)
assert "VEDASTRO_API_ENDPOINT=https://api.vedastro.org/api" in updated
assert "VEDASTRO_API_KEY=sample-secret" in updated
assert "VEDASTRO_ENABLE_NETWORK=1" in updated
assert "OTHER=value" in updated
assert "https://old.example/api" not in updated
@@ -0,0 +1,9 @@
from scripts.report_renderer_isolation_poc import run_poc
def test_report_renderer_isolation_poc_never_claims_pass_without_browser() -> None:
result = run_poc()
assert result["status"] in {"pass", "fail", "blocked"}
if result["status"] == "pass":
assert result["http_probe_requests"] == 0
assert result["blocked_resource_count"] >= 2