"""guard.py — pre_tool_call privacy guard for the ai-os plugin. Hard-blocks any file/terminal tool call that would read or write **another user's** per-user folder (``users//``). This is the one protection the skill-level ``identity.py`` cannot give: it does not depend on the LLM choosing to route through ``identity.py path`` — it intercepts the tool call itself. Policy: - Paths OUTSIDE the ``users/`` tree (content/, skills/, the bundle, /tmp, …) are allowed — only per-user data is scoped. - Paths inside ``users/`` are allowed only when they resolve inside the CURRENT user's folder. ``users/_template`` and ``users/_index.json`` are allowed (shared scaffolding / registry the engine itself manages). - When no user identity is resolvable (e.g. plain CLI with no gateway), the guard FAILS OPEN — it only enforces when it knows who is chatting. Real product traffic is gateway (Telegram), where identity is always present. Returns the Hermes pre_tool_call block directive ``{"action":"block","message":..}`` to veto, or ``None`` to allow. """ from __future__ import annotations import os import re from typing import Any, Dict, List, Optional from . import identity as ident # Tool args whose string values are filesystem paths. _PATH_ARG_KEYS = ("path", "file_path", "file", "target_file", "notebook_path", "dst", "src") # users/ references inside a shell command. _USERS_REF_RE = re.compile(r"users[/\\]([A-Za-z0-9_.-]+)") _SHARED_SEGMENTS = {"_template", "_index.json", "_index.json.tmp"} def _candidate_paths(tool_name: str, args: Dict[str, Any]) -> List[str]: out: List[str] = [] for k in _PATH_ARG_KEYS: v = args.get(k) if isinstance(v, str) and v.strip(): out.append(v.strip()) # multi-edit style: list of {path/file_path: ...} edits = args.get("edits") or args.get("files") if isinstance(edits, list): for e in edits: if isinstance(e, dict): for k in _PATH_ARG_KEYS: v = e.get(k) if isinstance(v, str) and v.strip(): out.append(v.strip()) return out def _abspath(p: str) -> str: if not os.path.isabs(p): p = os.path.join(ident.bundle_root(), p) return os.path.realpath(p) def _violation(user_id: Optional[str], abspath: str) -> bool: """A path is a violation iff it is inside users/ but not the current user's subtree and not shared scaffolding.""" if not ident.is_under_users(abspath): return False # outside per-user data → fine # shared, engine-managed entries base = os.path.realpath(ident.users_dir()) rel = os.path.relpath(abspath, base) first = rel.split(os.sep, 1)[0] if first in _SHARED_SEGMENTS: return False if not user_id: return False # fail-open: identity unknown return not ident.is_inside_user(user_id, abspath) def check(tool_name: str = "", args: Optional[Dict[str, Any]] = None, **_: Any) -> Optional[Dict[str, str]]: """pre_tool_call handler. Returns a block directive or None.""" if not isinstance(args, dict): return None user_id = ident.resolve_user() # File tools: precise path check. for p in _candidate_paths(tool_name, args): try: if _violation(user_id, _abspath(p)): return _block(p, user_id) except Exception: continue # never crash the tool path on a guard error # terminal: best-effort scan of the command for other users' folders. cmd = args.get("command") if isinstance(cmd, str) and user_id: for seg in _USERS_REF_RE.findall(cmd): if seg in _SHARED_SEGMENTS or seg == user_id: continue return _block("users/%s (in shell command)" % seg, user_id) return None def _block(path: str, user_id: Optional[str]) -> Dict[str, str]: return { "action": "block", "message": ( "ai-os privacy guard: blocked access to another user's data (%s). " "You may only read/write the current user's folder users/%s/. " "This protects per-user data isolation." % (path, user_id or "") ), }