71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
"""The engine must not accept anonymous traffic when INTERNAL_TOKEN is set.
|
|
|
|
Background (2026-09-15): lib/engine.ts on the edge has always SENT
|
|
`X-Internal-Token`, but the engine never verified it — a deployed engine would
|
|
have accepted anyone who could reach the port. The gate is fail-closed:
|
|
|
|
* INTERNAL_TOKEN set -> every route except /health requires the EXACT
|
|
token; wrong/missing = 401.
|
|
* INTERNAL_TOKEN unset -> gate inert (local dev); production compose always
|
|
sets it.
|
|
|
|
`/health` stays public on purpose: it is the liveness probe that
|
|
`lib/health-checks.ts` calls from the edge, and it leaks nothing but booleans.
|
|
|
|
The token is read at module import (the same value the running process serves
|
|
with), so each test reloads the module with the env it wants — and the
|
|
teardown reloads it one more time with the env CLEARED so later test files in
|
|
the same session never inherit a baked-in token.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import os
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture()
|
|
def engine(monkeypatch):
|
|
monkeypatch.setenv("INTERNAL_TOKEN", "s3cret")
|
|
import orchestrator.main as main
|
|
|
|
importlib.reload(main)
|
|
yield TestClient(main.app)
|
|
os.environ.pop("INTERNAL_TOKEN", None)
|
|
importlib.reload(main)
|
|
|
|
|
|
@pytest.fixture()
|
|
def engine_no_token(monkeypatch):
|
|
monkeypatch.delenv("INTERNAL_TOKEN", raising=False)
|
|
import orchestrator.main as main
|
|
|
|
importlib.reload(main)
|
|
yield TestClient(main.app)
|
|
os.environ.pop("INTERNAL_TOKEN", None)
|
|
importlib.reload(main)
|
|
|
|
|
|
def test_gate_off_when_internal_token_unset(engine_no_token):
|
|
assert engine_no_token.get("/health").status_code == 200
|
|
assert engine_no_token.get("/sources").status_code == 200
|
|
|
|
|
|
def test_health_stays_public_but_sources_require_token(engine):
|
|
# liveness probe is public by design
|
|
assert engine.get("/health").status_code == 200
|
|
# anonymous call to a data route -> 401 (was 200 before 2026-09-15)
|
|
r = engine.get("/sources")
|
|
assert r.status_code == 401
|
|
assert r.json()["error"]["code"] == "unauthorized"
|
|
|
|
|
|
def test_wrong_token_is_401_and_correct_token_passes(engine):
|
|
wrong = engine.get("/estimates?markets=SE", headers={"X-Internal-Token": "nope"})
|
|
assert wrong.status_code == 401
|
|
right = engine.get("/sources", headers={"X-Internal-Token": "s3cret"})
|
|
assert right.status_code == 200
|