279 lines
13 KiB
Python
279 lines
13 KiB
Python
"""Executable gates for trusted WebUI-to-HUX session context."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
import sys
|
|
import types
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SOURCE = ROOT / "dockerfiles/hermes-webui-hux-context.py"
|
|
PATCHER = ROOT / "dockerfiles/hermes-webui-hux-context-patch.py"
|
|
|
|
|
|
def _load(path: Path, name: str):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def _context_module(monkeypatch, tmp_path: Path, identity=None):
|
|
api = types.ModuleType("api")
|
|
api.__path__ = []
|
|
config = types.ModuleType("api.config")
|
|
config.STATE_DIR = tmp_path
|
|
bff = types.ModuleType("api.hux_bff")
|
|
trusted = identity or ("slot-3", "usr_" + "a" * 64, "chat")
|
|
bff._trusted_identity = lambda _handler: trusted
|
|
monkeypatch.setitem(sys.modules, "api", api)
|
|
monkeypatch.setitem(sys.modules, "api.config", config)
|
|
monkeypatch.setitem(sys.modules, "api.hux_bff", bff)
|
|
return _load(SOURCE, f"hux_context_{id(tmp_path)}")
|
|
|
|
|
|
def _session(**changes):
|
|
value = {"session_id": "webui-session-1", "project_id": "project-alpha", "profile": "default"}
|
|
value.update(changes)
|
|
return value
|
|
|
|
|
|
def test_context_is_stable_opaque_and_scoped_to_trusted_server_identity(monkeypatch, tmp_path):
|
|
module = _context_module(monkeypatch, tmp_path)
|
|
first = module.build_hux_context(object(), _session())
|
|
again = module.build_hux_context(object(), _session())
|
|
assert first == again
|
|
assert first == {
|
|
"schema": "hux.webui_context.v1",
|
|
"webui_session_id": "webui-session-1",
|
|
"session_id": first["session_id"],
|
|
"conversation_id": first["conversation_id"],
|
|
"project_id": first["project_id"],
|
|
"project_source": "profile:default",
|
|
"identity": {"tenant_slot": "slot-3", "subject": "usr_" + "a" * 64,
|
|
"surface": "chat", "trust": "relay"},
|
|
}
|
|
assert first["session_id"].startswith("ses_") and len(first["session_id"]) == 36
|
|
assert first["conversation_id"].startswith("conv_") and len(first["conversation_id"]) == 37
|
|
assert first["project_id"].startswith("prj_") and len(first["project_id"]) == 36
|
|
assert module.build_hux_context(object(), _session(session_id="webui-session-2")) != first
|
|
changed_project = module.build_hux_context(object(), _session(project_id="project-beta"))
|
|
assert changed_project["session_id"] == first["session_id"]
|
|
assert changed_project["project_id"] == first["project_id"]
|
|
module._KEY_CACHE = None
|
|
assert module.build_hux_context(object(), _session()) == first
|
|
key = tmp_path / ".hux-context-key"
|
|
assert stat.S_IMODE(key.stat().st_mode) == 0o600 and key.stat().st_size == 32
|
|
assert module.derive_hux_id(b"k" * 32, "run", "run", "slot-3",
|
|
"usr_" + "a" * 64, "turn-7") == "run_46e71223aae02a0f1f8f37f385320c5f"
|
|
assert module.derive_hux_id(b"k" * 32, "msg", "message", "slot-3",
|
|
"usr_" + "a" * 64, "12").startswith("msg_")
|
|
with pytest.raises(module.HuxContextUnavailable):
|
|
module.derive_hux_id(b"short", "run", "run", "slot-3", "usr_" + "a" * 64, "turn-7")
|
|
with pytest.raises(module.HuxContextUnavailable):
|
|
module.derive_hux_id(b"k" * 32, "conv", "run", "slot-3", "usr_" + "a" * 64, "turn-7")
|
|
|
|
|
|
def test_context_uses_server_project_source_and_never_accepts_inbound_context(monkeypatch, tmp_path):
|
|
module = _context_module(monkeypatch, tmp_path)
|
|
default = module.build_hux_context(object(), _session(project_id="browser-project", profile="research"))
|
|
assert default["project_source"] == "profile:default"
|
|
changed_session = module.build_hux_context(object(), _session(project_id="other", profile="other"))
|
|
assert changed_session["project_id"] == default["project_id"]
|
|
monkeypatch.setenv("HUX_PROJECT_SOURCE", "project:server-owned")
|
|
overridden = module.build_hux_context(object(), _session(project_id="browser-project"))
|
|
assert overridden["project_source"] == "project:server-owned"
|
|
assert overridden["project_id"] != default["project_id"]
|
|
forged = _session(hux_context={"identity": {"subject": "attacker"}})
|
|
attached = module.attach_hux_context(object(), forged)
|
|
assert attached["hux_context"]["identity"]["subject"] == "usr_" + "a" * 64
|
|
assert forged["hux_context"]["identity"]["subject"] == "attacker"
|
|
custom = tmp_path / "shared-context.key"
|
|
monkeypatch.setenv("HUX_CONTEXT_KEY_FILE", str(custom))
|
|
module._KEY_CACHE = None
|
|
assert module.build_hux_context(object(), _session())["conversation_id"] != default["conversation_id"]
|
|
assert custom.is_file() and stat.S_IMODE(custom.stat().st_mode) == 0o600
|
|
|
|
|
|
def test_context_enriches_only_authoritative_server_lifecycle_values(monkeypatch, tmp_path):
|
|
module = _context_module(monkeypatch, tmp_path)
|
|
enriched = module.build_hux_context(object(), _session(
|
|
turn_id="turn-7", active_stream_id="stream-not-a-run",
|
|
messages=[{"id": 4}, {"message_id": "server-5"}],
|
|
branch_point_message_id=3, notebook_id="nb_reviewed1",
|
|
))
|
|
assert enriched["run_id"].startswith("run_")
|
|
assert enriched["message_id"].startswith("msg_")
|
|
assert enriched["branch_point_message_id"].startswith("msg_")
|
|
assert enriched["notebook_id"] == "nb_reviewed1"
|
|
assert enriched["message_id"] != enriched["branch_point_message_id"]
|
|
|
|
partial = module.build_hux_context(object(), _session(
|
|
active_stream_id="stream-not-a-run", turn_id=True,
|
|
messages=[{"id": True}, {"message_id": "bad/value"}],
|
|
branch_point_message_id={"forged": True}, notebook_id="browser-notebook",
|
|
))
|
|
for field in ("run_id", "message_id", "branch_point_message_id", "notebook_id"):
|
|
assert field not in partial
|
|
|
|
|
|
def test_context_fails_closed_for_invalid_or_untrusted_inputs(monkeypatch, tmp_path):
|
|
module = _context_module(monkeypatch, tmp_path)
|
|
assert module.attach_hux_context(object(), None) is None
|
|
assert "hux_context" not in module.attach_hux_context(object(), {"session_id": "bad/value"})
|
|
monkeypatch.setenv("HUX_PROJECT_SOURCE", "bad/value")
|
|
assert "hux_context" not in module.attach_hux_context(object(), _session())
|
|
monkeypatch.delenv("HUX_PROJECT_SOURCE")
|
|
bff = sys.modules["api.hux_bff"]
|
|
bff._trusted_identity = lambda _handler: (_ for _ in ()).throw(PermissionError("no"))
|
|
clean = module.attach_hux_context(object(), _session(hux_context={"forged": True}))
|
|
assert "hux_context" not in clean
|
|
|
|
bff._trusted_identity = lambda _handler: ("slot-3", "usr_" + "a" * 64, "chat")
|
|
module._KEY_CACHE = None
|
|
key = tmp_path / ".hux-context-key"
|
|
key.write_bytes(b"short")
|
|
os.chmod(key, 0o600)
|
|
assert "hux_context" not in module.attach_hux_context(object(), _session())
|
|
|
|
|
|
def test_context_key_rejects_unsafe_state_and_creation_errors(monkeypatch, tmp_path):
|
|
unsafe = tmp_path / "unsafe"
|
|
unsafe.mkdir()
|
|
module = _context_module(monkeypatch, unsafe)
|
|
original_geteuid = os.geteuid
|
|
module.os.geteuid = lambda: unsafe.stat().st_uid + 1
|
|
with pytest.raises(module.HuxContextUnavailable, match="state"):
|
|
module.build_hux_context(object(), _session())
|
|
module.os.geteuid = original_geteuid
|
|
module._KEY_CACHE = None
|
|
key = unsafe / ".hux-context-key"
|
|
key.write_bytes(b"x" * 32)
|
|
os.chmod(key, 0o644)
|
|
with pytest.raises(module.HuxContextUnavailable, match="unsafe"):
|
|
module.build_hux_context(object(), _session())
|
|
|
|
|
|
def test_context_low_level_key_failures_are_closed(monkeypatch, tmp_path):
|
|
module = _context_module(monkeypatch, tmp_path)
|
|
with pytest.raises(module.HuxContextUnavailable, match="unavailable"):
|
|
module._read_key(tmp_path / "missing")
|
|
monkeypatch.setenv("HUX_CONTEXT_KEY_FILE", "relative.key")
|
|
with pytest.raises(module.HuxContextUnavailable, match="path"):
|
|
module._key_path()
|
|
monkeypatch.setenv("HUX_CONTEXT_KEY_FILE", str(tmp_path / "missing/key"))
|
|
with pytest.raises(module.HuxContextUnavailable, match="state"):
|
|
module._key_path()
|
|
parent_file = tmp_path / "parent-file"
|
|
parent_file.write_text("x", encoding="utf-8")
|
|
monkeypatch.setenv("HUX_CONTEXT_KEY_FILE", str(parent_file / "key"))
|
|
with pytest.raises(module.HuxContextUnavailable, match="state"):
|
|
module._key_path()
|
|
monkeypatch.delenv("HUX_CONTEXT_KEY_FILE")
|
|
|
|
short = tmp_path / "short-read"
|
|
short.write_bytes(b"x" * 32)
|
|
os.chmod(short, 0o600)
|
|
monkeypatch.setattr(module.os, "read", lambda _descriptor, _size: b"short")
|
|
with pytest.raises(module.HuxContextUnavailable, match="invalid"):
|
|
module._read_key(short)
|
|
monkeypatch.undo()
|
|
|
|
|
|
def test_context_rejects_incomplete_key_creation_and_every_invalid_id_field(monkeypatch, tmp_path):
|
|
module = _context_module(monkeypatch, tmp_path)
|
|
blocked = tmp_path / "blocked-key"
|
|
original_open = module.os.open
|
|
monkeypatch.setattr(module.os, "open", lambda *_args, **_kwargs: (_ for _ in ()).throw(PermissionError()))
|
|
with pytest.raises(module.HuxContextUnavailable, match="created"):
|
|
module._create_key(blocked)
|
|
monkeypatch.setattr(module.os, "open", original_open)
|
|
monkeypatch.setattr(module.os, "write", lambda _descriptor, _value: 1)
|
|
with pytest.raises(module.HuxContextUnavailable, match="incomplete"):
|
|
module._create_key(tmp_path / "partial-key")
|
|
|
|
valid = (b"k" * 32, "run", "run", "slot-3", "usr_" + "a" * 64, "turn-7")
|
|
invalid = [
|
|
(None, *valid[1:]),
|
|
(b"short", *valid[1:]),
|
|
(valid[0], "bad", *valid[2:]),
|
|
(valid[0], valid[1], "conversation", *valid[3:]),
|
|
(*valid[:3], "tenant", *valid[4:]),
|
|
(*valid[:4], "raw-user", valid[5]),
|
|
(*valid[:5], "bad/value"),
|
|
]
|
|
for values in invalid:
|
|
with pytest.raises(module.HuxContextUnavailable):
|
|
module.derive_hux_id(*values)
|
|
with pytest.raises(module.HuxContextUnavailable, match="payload"):
|
|
module.build_hux_context(object(), [])
|
|
|
|
|
|
def test_pinned_patch_installs_session_and_chat_start_response_paths_and_rejects_drift(tmp_path):
|
|
patcher = _load(PATCHER, "hux_context_patcher")
|
|
root = tmp_path / "webui"
|
|
(root / "api").mkdir(parents=True)
|
|
routes = root / "api/routes.py"
|
|
routes.write_text(
|
|
" redact = redact_session_data(raw)\n"
|
|
' return j(handler, {"session": redact_session_data(sess)})\n'
|
|
' payload = {"session": s.compact() | {"messages": s.messages}}\n'
|
|
" return j(handler, response, status=status)\n",
|
|
encoding="utf-8",
|
|
)
|
|
patcher.apply(root, SOURCE)
|
|
patched = routes.read_text(encoding="utf-8")
|
|
assert patched.count("from api.hux_context import attach_hux_context") == 4
|
|
assert "attach_hux_context(handler, redact)" in patched
|
|
assert "response = attach_hux_context(handler, response)" in patched
|
|
assert (root / "api/hux_context.py").read_text(encoding="utf-8") == SOURCE.read_text(encoding="utf-8")
|
|
with pytest.raises(SystemExit, match="already installed"):
|
|
patcher.apply(root, SOURCE)
|
|
|
|
drift = tmp_path / "drift"
|
|
(drift / "api").mkdir(parents=True)
|
|
drift_routes = drift / "api/routes.py"
|
|
drift_routes.write_text(" redact = redact_session_data(raw)\n", encoding="utf-8")
|
|
with pytest.raises(SystemExit, match="patch changed"):
|
|
patcher.apply(drift, SOURCE)
|
|
assert not (drift / "api/hux_context.py").exists()
|
|
|
|
invalid_source = tmp_path / "invalid.py"
|
|
invalid_source.write_text("# missing helpers\n", encoding="utf-8")
|
|
fresh = tmp_path / "fresh"
|
|
(fresh / "api").mkdir(parents=True)
|
|
(fresh / "api/routes.py").write_text(
|
|
" redact = redact_session_data(raw)\n"
|
|
' return j(handler, {"session": redact_session_data(sess)})\n'
|
|
' payload = {"session": s.compact() | {"messages": s.messages}}\n'
|
|
" return j(handler, response, status=status)\n", encoding="utf-8")
|
|
with pytest.raises(SystemExit, match="module is invalid"):
|
|
patcher.apply(fresh, invalid_source)
|
|
patcher.ROOT = fresh
|
|
patcher.MODULE_SOURCE = SOURCE
|
|
patcher.apply()
|
|
|
|
|
|
def test_context_delivery_is_wired_into_image_ci_and_smoke_contract():
|
|
dockerfile = (ROOT / "dockerfiles/Dockerfile.hermes-webui").read_text(encoding="utf-8")
|
|
jenkins = (ROOT / "ci/Jenkinsfile.hermes-webui-image").read_text(encoding="utf-8")
|
|
assert "hermes-webui-hux-context-patch.py" in dockerfile
|
|
assert "hermes-webui-hux-context.py" in dockerfile
|
|
assert "api/hux_context.py" in dockerfile
|
|
assert "COPY dockerfiles/hermes-hux-foundation/hux /opt/hermes-hux/hux" in dockerfile
|
|
assert "from hux.server import build_router" in dockerfile
|
|
assert "PYTHONPATH=/opt/hermes-hux" in dockerfile
|
|
assert "test_hermes_webui_hux_context.py" in jenkins
|
|
assert len(SOURCE.read_text(encoding="utf-8").splitlines()) <= 500
|
|
assert len(PATCHER.read_text(encoding="utf-8").splitlines()) <= 500
|
|
lowered = SOURCE.read_text(encoding="utf-8").lower()
|
|
for forbidden in ("query", "request body", "localstorage", "cookie", "location.href"):
|
|
assert forbidden not in lowered
|