harden calculation contracts and local API boundaries

tests/test_runtime_security_p0.py

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