Add API security and async job contracts

2026-07-16 23:40:01 +08:00
parent a7fdfcb033
commit e013fed79c
10 changed files with 573 additions and 17 deletions
@@ -0,0 +1,105 @@
+import json
+import time
+from pathlib import Path
+
+import pytest
+
+from scripts import jyotish_api_server as api
+
+
+def test_evidence_packet_view_exposes_only_auditable_result_sections():
+    packet = api.build_evidence_packet_view({
+        "job_id": "job_1",
+        "status": "completed",
+        "result": {
+            "fallback_reason": "VedAstro official snapshot blocked: timeout",
+            "machine_evidence_packet": {"status": "draft", "metadata": {"capture_id": "x"}},
+            "technique_audit": [{"technique": "D9", "status": "used"}],
+            "ai_prompt_pack": {"prompt_zh": "internal prompt"},
+        },
+    })
+
+    assert packet["job_id"] == "job_1"
+    assert packet["execution_status"]["official_evidence_status"] == "official_blocked"
+    assert packet["machine_evidence_packet"]["metadata"]["capture_id"] == "x"
+    assert "ai_prompt_pack" not in packet
+
+
+def test_async_job_route_extracts_id_before_loading(monkeypatch, tmp_path):
+    record = {"job_id": "chart_abc", "status": "completed", "result": {}}
+    monkeypatch.setattr(api, "_load_async_job_record", lambda *args, **kwargs: record)
+
+    handler = api.JyotishAPIHandler.__new__(api.JyotishAPIHandler)
+    handler.path = "/api/chart/jobs/chart_abc"
+    handler.headers = {"Origin": ""}
+    handler._enforce_request_security = lambda: None
+    captured = {}
+    handler._json = lambda data, status=200: captured.update(data=data, status=status)
+    handler._error_json = lambda message, status=500, error_code="ERR_INTERNAL": captured.update(error=error_code, status=status)
+    handler._job_access_token = lambda: "token"
+
+    handler.do_GET()
+
+    assert captured["status"] == 200
+    assert captured["data"]["job_id"] == "chart_abc"
+
+
+def test_evidence_packet_page_is_present_and_does_not_embed_birth_data():
+    page = Path(api.REPO_ROOT) / "web" / "evidence_packet.html"
+    source = page.read_text(encoding="utf-8")
+
+    assert "Evidence Packet" in source
+    assert "access token" in source
+    assert "birth" not in source.lower()
+
+
+def test_rectification_page_uses_choice_questionnaire_contract():
+    page = Path(api.REPO_ROOT) / "web" / "rectification.html"
+    source = page.read_text(encoding="utf-8")
+
+    assert "/api/rectification/questionnaire" in source
+    assert "/api/rectification/answers" in source
+    assert "候选簇排序" in source
+
+
+def test_home_page_keeps_location_confirmation_local():
+    page = Path(api.REPO_ROOT) / "web" / "index.html"
+    source = page.read_text(encoding="utf-8")
+
+    assert "/api/location/resolve" in source
+    assert "第三方地理服务" in source
+
+
+def test_startup_cleanup_removes_only_expired_job_records(monkeypatch, tmp_path):
+    expired = tmp_path / "expired.json"
+    active = tmp_path / "active.json"
+    expired.write_text(json.dumps({"expires_at_unix": time.time() - 1}), encoding="utf-8")
+    active.write_text(json.dumps({"expires_at_unix": time.time() + 60}), encoding="utf-8")
+    monkeypatch.setattr(api, "_async_job_dir", lambda scope: tmp_path)
+
+    result = api.prune_expired_async_jobs()
+
+    assert result["removed"] == 1
+    assert not expired.exists()
+    assert active.exists()
+
+
+def test_rate_limit_is_configurable_and_rejects_over_budget(monkeypatch):
+    api._RATE_LIMIT_BUCKETS.clear()
+    monkeypatch.setenv("JYOTISH_API_RATE_LIMIT_PER_MINUTE", "1")
+
+    api.enforce_rate_limit("test-client", now=0)
+    with pytest.raises(api.RateLimited):
+        api.enforce_rate_limit("test-client", now=1)
+
+
+def test_sqlite_async_job_backend_preserves_token_and_ttl(monkeypatch, tmp_path):
+    monkeypatch.setenv("JYOTISH_ASYNC_JOB_BACKEND", "sqlite")
+    monkeypatch.setattr(api, "_sqlite_job_db_path", lambda: tmp_path / "jobs.sqlite3")
+    record = {"job_id": "job_sqlite", "access_token_hash": api._access_token_hash("secret"), "expires_at_unix": time.time() + 60}
+
+    api._write_async_job_record("test_scope", "job_sqlite", record)
+
+    assert api._load_async_job_record("test_scope", "job_sqlite", access_token="secret")["job_id"] == "job_sqlite"
+    with pytest.raises(api.JobAccessDenied):
+        api._load_async_job_record("test_scope", "job_sqlite", access_token="wrong")
@@ -0,0 +1,156 @@
+from __future__ import annotations
+
+import hashlib
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / "scripts"
+if str(SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS))
+
+import jyotish_api_server as api  # noqa: E402
+import report_builder  # noqa: E402
+
+
+class _Headers(dict):
+    def get(self, key, default=None):
+        return super().get(key, default)
+
+
+class _Server:
+    allowed_origins = {"http://localhost:3456"}
+    server_address = ("127.0.0.1", 5200)
+
+
+def _handler(headers: dict[str, str]):
+    handler = api.JyotishAPIHandler.__new__(api.JyotishAPIHandler)
+    handler.headers = _Headers(headers)
+    handler.server = _Server()
+    return handler
+
+
+def test_untrusted_origin_is_rejected_before_post_side_effects() -> None:
+    handler = _handler(
+        {
+            "Origin": "https://evil.example",
+            "Host": "127.0.0.1:5200",
+            "Content-Type": "application/json",
+        }
+    )
+    with pytest.raises(api.Forbidden, match="Origin"):
+        handler._enforce_request_security(require_json=True)
+
+
+def test_post_requires_json_content_type() -> None:
+    handler = _handler(
+        {
+            "Origin": "http://localhost:3456",
+            "Host": "127.0.0.1:5200",
+            "Content-Type": "text/plain",
+        }
+    )
+    with pytest.raises(api.UnsupportedMediaType):
+        handler._enforce_request_security(require_json=True)
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://example.com/image.png",
+        "http://127.0.0.1:8080/private",
+        "file:///etc/passwd",
+        "ftp://example.com/file",
+    ],
+)
+def test_report_renderer_blocks_external_and_local_resources(url: str) -> None:
+    assert report_builder.is_allowed_report_resource_url(
+        url,
+        report_url="file:///tmp/report.html",
+    ) is False
+
+
+def test_report_renderer_allows_only_document_and_embedded_resources() -> None:
+    assert report_builder.is_allowed_report_resource_url(
+        "file:///tmp/report.html",
+        report_url="file:///tmp/report.html",
+    ) is True
+    assert report_builder.is_allowed_report_resource_url(
+        "data:image/png;base64,AA==",
+        report_url="file:///tmp/report.html",
+    ) is True
+
+
+def test_async_job_identity_is_random_and_capability_protected(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    monkeypatch.setattr(api, "_async_job_dir", lambda _scope: tmp_path)
+    first = api._new_async_job_identity("chart")
+    second = api._new_async_job_identity("chart")
+    assert first["job_id"] != second["job_id"]
+    assert len(first["job_id"].split("_", 1)[1]) >= 32
+    assert first["access_token"] != second["access_token"]
+
+    record = {
+        "job_id": first["job_id"],
+        "status": "queued",
+        "access_token_hash": hashlib.sha256(first["access_token"].encode()).hexdigest(),
+        "expires_at_unix": time.time() + 60,
+    }
+    api._write_async_job_record("chart", first["job_id"], record)
+    assert api._load_async_job_record(
+        "chart", first["job_id"], access_token=first["access_token"]
+    )["status"] == "queued"
+    with pytest.raises(api.JobAccessDenied):
+        api._load_async_job_record("chart", first["job_id"], access_token="wrong")
+
+
+def test_expired_async_job_is_deleted(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    monkeypatch.setattr(api, "_async_job_dir", lambda _scope: tmp_path)
+    identity = api._new_async_job_identity("chart")
+    api._write_async_job_record(
+        "chart",
+        identity["job_id"],
+        {
+            "job_id": identity["job_id"],
+            "access_token_hash": hashlib.sha256(identity["access_token"].encode()).hexdigest(),
+            "expires_at_unix": time.time() - 1,
+        },
+    )
+    assert api._load_async_job_record(
+        "chart", identity["job_id"], access_token=identity["access_token"]
+    ) is None
+    assert not (tmp_path / f"{identity['job_id']}.json").exists()
+
+
+def test_authenticated_frontend_does_not_use_local_storage_api_base() -> None:
+    auth_source = (ROOT / "jyotish-app" / "auth.js").read_text(encoding="utf-8")
+    chat_source = (ROOT / "jyotish-app" / "ai-chat.js").read_text(encoding="utf-8")
+    assert "localStorage.getItem(API_BASE_KEY)" not in auth_source
+    assert "localStorage.getItem('jyotish_api_base')" not in chat_source
+
+
+def test_frontend_async_poll_uses_ephemeral_job_capability() -> None:
+    bridge_source = (ROOT / "jyotish-app" / "api-bridge.js").read_text(encoding="utf-8")
+    assert "pollAsyncJob(data, { base })" in bridge_source
+    assert "Authorization: `Bearer ${job.access_token}`" in bridge_source
+    assert "sessionStorage.setItem('jyotish_job" not in bridge_source
+
+
+def test_background_job_queue_rejects_when_capacity_is_full(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class _FullCapacity:
+        def acquire(self, blocking=False):
+            return False
+
+    monkeypatch.setattr(api, "_ASYNC_JOB_CAPACITY", _FullCapacity())
+    with pytest.raises(api.JobQueueFull):
+        api._submit_background_job(lambda: None)