106 lines
3.8 KiB
Python
106 lines
3.8 KiB
Python
"""Derive HUX entity ids from server-owned context and subject files."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
import re
|
|
import stat
|
|
from pathlib import Path
|
|
|
|
DOMAIN = "hux.context.id.v1"
|
|
KEY_BYTES = 32
|
|
RAW_ID = re.compile(r"^[A-Za-z0-9._:@+-]{1,200}$")
|
|
SLOT = re.compile(r"^slot-[0-9]{1,3}$")
|
|
SUBJECT = re.compile(r"^usr_[0-9a-f]{64}$")
|
|
PAIRS = frozenset({("ses", "session"), ("conv", "conversation"), ("prj", "project"), ("run", "run")})
|
|
|
|
|
|
class ContextUnavailable(ValueError):
|
|
"""Trusted runtime context cannot be constructed."""
|
|
|
|
|
|
def _read_owned_key(path: str | Path) -> bytes:
|
|
"""Read one single-link, owner-only 32-byte context key without following links."""
|
|
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
|
|
try:
|
|
descriptor = os.open(Path(path), flags)
|
|
except OSError as exc:
|
|
raise ContextUnavailable("HUX context key is unavailable") from exc
|
|
try:
|
|
info = os.fstat(descriptor)
|
|
if (
|
|
not stat.S_ISREG(info.st_mode)
|
|
or info.st_uid != os.geteuid()
|
|
or stat.S_IMODE(info.st_mode) != 0o600
|
|
or info.st_nlink != 1
|
|
or info.st_size != KEY_BYTES
|
|
):
|
|
raise ContextUnavailable("HUX context key is unsafe")
|
|
value = os.read(descriptor, KEY_BYTES + 1)
|
|
except ContextUnavailable:
|
|
raise
|
|
except OSError as exc:
|
|
raise ContextUnavailable("HUX context key is unavailable") from exc
|
|
finally:
|
|
os.close(descriptor)
|
|
if len(value) != KEY_BYTES:
|
|
raise ContextUnavailable("HUX context key is invalid")
|
|
return value
|
|
|
|
|
|
def derive_hux_id(
|
|
key: bytes,
|
|
prefix: str,
|
|
purpose: str,
|
|
slot: str,
|
|
subject: str,
|
|
raw: str,
|
|
) -> str:
|
|
"""Match WebUI's public ``hux.context.id.v1`` derivation exactly."""
|
|
if (prefix, purpose) not in PAIRS:
|
|
raise ContextUnavailable("unsupported HUX id purpose")
|
|
if not SLOT.fullmatch(slot) or not SUBJECT.fullmatch(subject):
|
|
raise ContextUnavailable("HUX identity is malformed")
|
|
if not isinstance(raw, str) or not RAW_ID.fullmatch(raw):
|
|
raise ContextUnavailable("runtime context id is malformed")
|
|
if not isinstance(key, bytes) or len(key) != KEY_BYTES:
|
|
raise ContextUnavailable("HUX context key is invalid")
|
|
message = "\0".join((DOMAIN, purpose, slot, subject, raw)).encode("utf-8")
|
|
digest = hmac.new(key, message, hashlib.sha256).hexdigest()[:32]
|
|
return f"{prefix}_{digest}"
|
|
|
|
|
|
class ContextIds:
|
|
"""Stable HUX ids for one Hermes tool call."""
|
|
|
|
def __init__(self, key_file: str | Path, slot: str, subject: str) -> None:
|
|
if not SLOT.fullmatch(slot) or not SUBJECT.fullmatch(subject):
|
|
raise ContextUnavailable("HUX identity is malformed")
|
|
self._key = _read_owned_key(key_file)
|
|
self._slot = slot
|
|
self._subject = subject
|
|
|
|
def conversation(self, raw_session_id: str) -> str:
|
|
"""Map the persisted Hermes/WebUI session to its shared conversation id."""
|
|
return derive_hux_id(
|
|
self._key, "conv", "conversation", self._slot, self._subject, raw_session_id
|
|
)
|
|
|
|
def session(self, raw_session_id: str) -> str:
|
|
"""Map the persisted Hermes/WebUI session to its shared session id."""
|
|
return derive_hux_id(
|
|
self._key, "ses", "session", self._slot, self._subject, raw_session_id
|
|
)
|
|
|
|
def project(self, project_source: str) -> str:
|
|
"""Map the server-selected project source to its shared project id."""
|
|
return derive_hux_id(
|
|
self._key, "prj", "project", self._slot, self._subject, project_source
|
|
)
|
|
|
|
def run(self, raw_turn_id: str) -> str:
|
|
"""Map one stable Hermes turn to its shared HUX run id."""
|
|
return derive_hux_id(self._key, "run", "run", self._slot, self._subject, raw_turn_id)
|