"""identity.py — in-process per-user identity for the ai-os plugin. Runs INSIDE the Hermes process (plugin hooks/tools), unlike core/lib/identity.py which is a CLI script shelled out from skills. Both share the SAME convention (``user_id = _``, e.g. ``tg_123456789``) and the same ``users//`` guardrail, so a user is identified identically whether the code path is a skill subprocess or a plugin handler. Session identity source (verified against Hermes source): - The gateway exposes per-task session vars via ``gateway.session_context`` (ContextVar-backed, with an ``os.environ`` fallback for CLI/cron). We read ``HERMES_SESSION_{PLATFORM,USER_ID,CHAT_ID,USER_NAME}`` through ``get_session_env``. - In a DM the chat is per-user; in a shared group/forum thread Hermes does NOT isolate users in the session key, so ``HERMES_SESSION_USER_ID`` is the only reliable per-message identity. Resolution returns None when absent (CLI with no gateway) — callers fall back gracefully. Stdlib only. """ from __future__ import annotations import os import re import json import shutil import datetime from typing import Optional, Tuple PLATFORM_PREFIX = {"telegram": "tg", "slack": "sl", "discord": "dc", "whatsapp": "wa"} # -------------------------------------------------------------------------- # Session env access — prefer Hermes' ContextVar-aware helper, fall back to env. # -------------------------------------------------------------------------- def get_session_env(name: str, default: str = "") -> str: """Read a HERMES_SESSION_* value (ContextVar-aware in gateway, env elsewhere).""" try: from gateway.session_context import get_session_env as _g # type: ignore return _g(name, default) except Exception: return os.environ.get(name, default) or default # -------------------------------------------------------------------------- # Bundle root — where users/ and personas/ live (== terminal.cwd at runtime). # Resolution order is explicit→implicit so deploy can pin it deterministically. # -------------------------------------------------------------------------- def bundle_root() -> str: # 1) explicit override set by deploy (recommended on real instances) env_root = os.environ.get("AI_OS_BUNDLE_ROOT") if env_root and os.path.isdir(env_root): return os.path.abspath(env_root) # 2) the agent's working dir (Hermes sets terminal.cwd == bundle root) for cand in (get_session_env("HERMES_SESSION_CWD"), os.getcwd()): if cand and os.path.isdir(os.path.join(cand, "users")): return os.path.abspath(cand) # 3) last resort: cwd (users/ created lazily on first contact) return os.path.abspath(os.getcwd()) def users_dir() -> str: return os.path.join(bundle_root(), "users") def index_path() -> str: return os.path.join(users_dir(), "_index.json") # -------------------------------------------------------------------------- # Identity resolution # -------------------------------------------------------------------------- def stash_identity(platform: str = "", user_id: str = "", chat_id: str = "", user_name: str = "") -> None: """Bridge gateway-resolved identity into os.environ. Plugin slash-command handlers are dispatched BEFORE Hermes sets its session ContextVars (verified: gateway/run.py dispatches plugin commands ~L7820, while set_session_vars runs ~L8435 wrapping the agent turn). So inside a slash handler get_session_env() is empty. The pre_gateway_dispatch hook fires earliest with the full event.source, so we stash identity here for those handlers to read. During the agent turn the ContextVar is set and takes precedence in get_session_env, and _make_run_env overlays ContextVars onto subprocess env — so this global write is overridden where it matters and only "wins" outside a turn (slash commands).""" if platform: os.environ["HERMES_SESSION_PLATFORM"] = str(platform) if user_id: os.environ["HERMES_SESSION_USER_ID"] = str(user_id) if chat_id: os.environ["HERMES_SESSION_CHAT_ID"] = str(chat_id) if user_name: os.environ["HERMES_SESSION_USER_NAME"] = str(user_name) def resolve_user() -> Optional[str]: """Stable user_id from the session (None if no platform id is available).""" platform = (get_session_env("HERMES_SESSION_PLATFORM", "telegram") or "telegram").lower() uid = get_session_env("HERMES_SESSION_USER_ID") or get_session_env("HERMES_SESSION_CHAT_ID") if not uid: return None prefix = PLATFORM_PREFIX.get(platform) or (re.sub(r"[^a-z0-9]", "", platform)[:2] or "xx") safe = re.sub(r"[^A-Za-z0-9_-]", "", str(uid)) if not safe: return None return "%s_%s" % (prefix, safe) def user_path(user_id: str, *parts: str) -> str: """Absolute path inside users//, HARD-BLOCKING any escape (../ out).""" if not user_id or "/" in user_id or os.sep in user_id or user_id in ("", ".", ".."): raise ValueError("invalid user_id: %r" % (user_id,)) base = os.path.realpath(os.path.join(users_dir(), user_id)) target = os.path.realpath(os.path.join(base, *parts)) if parts else base if target != base and not target.startswith(base + os.sep): raise ValueError("path escape blocked: %r" % (parts,)) return target def is_inside_user(user_id: str, abspath: str) -> bool: """True iff abspath resolves inside users// (used by the guard).""" try: base = os.path.realpath(user_path(user_id)) except ValueError: return False rp = os.path.realpath(abspath) return rp == base or rp.startswith(base + os.sep) def is_under_users(abspath: str) -> bool: """True iff abspath resolves anywhere inside the users/ tree (any user).""" root = os.path.realpath(users_dir()) rp = os.path.realpath(abspath) return rp == root or rp.startswith(root + os.sep) def ensure_user_dir(user_id: str, template_persona: Optional[str] = None) -> Tuple[str, bool]: """Create users// from users/_template/ if missing. Returns (dir, created).""" d = user_path(user_id) created = False if not os.path.isdir(d): template = os.path.join(users_dir(), "_template") if os.path.isdir(template): shutil.copytree(template, d) else: os.makedirs(d, exist_ok=True) created = True return d, created # -------------------------------------------------------------------------- # Registry (users/_index.json) — source of truth for chat_id / org_role / nudge. # -------------------------------------------------------------------------- def load_index() -> dict: try: with open(index_path(), encoding="utf-8") as f: return json.load(f) except Exception: return {"version": 1, "users": {}} def save_index(idx: dict) -> None: os.makedirs(users_dir(), exist_ok=True) tmp = index_path() + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(idx, f, ensure_ascii=False, indent=2) os.replace(tmp, index_path()) def upsert_index(user_id: str, **fields) -> dict: idx = load_index() users = idx.setdefault("users", {}) u = users.setdefault(user_id, {}) now = datetime.date.today().isoformat() u["user_id"] = user_id u.setdefault("created_at", now) u["last_seen"] = now u.setdefault("platform", get_session_env("HERMES_SESSION_PLATFORM", "telegram")) chat_id = get_session_env("HERMES_SESSION_CHAT_ID") if chat_id: u["chat_id"] = chat_id # DM target for nudges name = get_session_env("HERMES_SESSION_USER_NAME") if name and not u.get("display_name"): u["display_name"] = name for k, v in fields.items(): if v not in (None, ""): u[k] = v save_index(idx) return u def org_role(user_id: str) -> str: """Permission level of the user: ceo(admin) | manager | staff (default staff).""" return (load_index().get("users", {}).get(user_id, {}) or {}).get("org_role", "staff")