"""Authenticated tool-call session records — the trust boundary for delegated execution.
The dedicated ``tools`` service runs the dangerous / ``UNSANDBOXED_EXEC`` / shell
tools, and ``tools/alter_privileges.has_privilege`` keys authority off ``user_id``.
The request envelope carries a ``user_id``, but the tools service **must not
trust it**: anything able to ``XADD sg:stream:tools`` could otherwise forge an
admin id and escalate.
Instead, the inference tier — which is processing an authenticated platform
message — writes the **real** identity to a short-lived Redis record keyed by the
call's ``trace_id``, and the tools service resolves identity (and therefore
privileges) from *that record*, treating the envelope's identity fields as an
untrusted lookup. A forged request that references an unknown ``trace_id`` is
rejected (fail-closed); one that references a real ``trace_id`` only ever gets
that (legitimate, non-attacker) user's identity. Defense-in-depth, paired with a
Redis ACL that restricts ``XADD sg:stream:tools`` to inference workers (ops; see
``scripts/systemd/stargazer-tools.service``).
"""
from __future__ import annotations
import logging
from typing import Any, Optional
import msgpack
logger = logging.getLogger("stargazer.tool_session")
SESSION_PREFIX = "sg:tools:session:"
DEFAULT_TTL = 300 # seconds; must exceed tools_exec_timeout
[docs]
async def write_session(
redis: Any,
trace_id: str,
identity: dict[str, Any],
ttl: int = DEFAULT_TTL,
) -> None:
"""Persist the authenticated identity for *trace_id* (inference side).
*identity* should carry the real ``user_id`` / ``guild_id`` / ``channel_id`` /
``platform`` (and optional ``user_aliases``) of the message being processed.
No-op when *trace_id* is empty.
"""
if not trace_id:
return
try:
await redis.set(
f"{SESSION_PREFIX}{trace_id}",
msgpack.packb(identity, use_bin_type=True),
ex=ttl,
)
except Exception:
logger.debug("failed to write tool session record", exc_info=True)
[docs]
async def read_session(redis: Any, trace_id: str) -> Optional[dict[str, Any]]:
"""Return the authenticated identity for *trace_id*, or ``None`` if absent."""
if not trace_id:
return None
try:
blob = await redis.get(f"{SESSION_PREFIX}{trace_id}")
except Exception:
logger.debug("failed to read tool session record", exc_info=True)
return None
if blob is None:
return None
try:
return msgpack.unpackb(blob, raw=False)
except Exception:
return None