diff --git a/tools/skills_sync_client.py b/tools/skills_sync_client.py index 1ff1b45713..df5ba80050 100644 --- a/tools/skills_sync_client.py +++ b/tools/skills_sync_client.py @@ -1,22 +1,16 @@ #!/usr/bin/env python3 """Skill Sync client -- the low-level sync layer. -Builds content-addressed objects from local skills and talks the sync wire -contract (push objects + CAS a ref, pull the owner's HEAD, three-way merge on -a 409). Driven by the debounced push hook in ``skill_manage``, the periodic -``maybe_pull_skills`` at the curator tick sites, and the ``hermes sync`` CLI. -Lives under tools/ (NOT hermes_cli/) so it never imports the CLI at module -load. ``skills_sync_client_wire`` (object model + HTTP client + tree walks) and -``skills_sync_client_org`` (org-shared skills) are re-exported here. - -ACCESS GATE (pre-launch): sync is INERT unless the signed-in user is a Nous -admin, read off the ``tool_gateway_admin`` JWT claim. The claim name is NAS's -and misleading: it is the global portal-admin permission, NOT a tool-gateway -right. Replace with a real entitlement before shipping to users. - -OPT-IN DEFAULT (provisional): nothing syncs unless the user marks a skill for -sync. Local intent is the ``sync`` flag in ``.usage.json``; the DURABLE -cross-device state is the committed ``sync-manifest`` blob in the plane. Only +Builds content-addressed objects from local skills and talks the sync wire contract (push +objects + CAS a ref, pull the owner's HEAD, three-way merge on a 409). Driven by the debounced +push hook in ``skill_manage``, ``maybe_pull_skills`` at the curator tick sites, and ``hermes +sync``. Lives under tools/ (NOT hermes_cli/) so it never imports the CLI at module load; +``skills_sync_client_wire`` and ``skills_sync_client_org`` are re-exported here. +ACCESS GATE (pre-launch): sync is INERT unless the signed-in user is a Nous admin, read off the +``tool_gateway_admin`` JWT claim (NAS's misleading name for the global portal-admin permission; +replace with a real entitlement before shipping). OPT-IN DEFAULT (provisional): nothing syncs +unless the user marks a skill for sync -- local intent is the ``sync`` flag in ``.usage.json``, +the DURABLE cross-device state is the committed ``sync-manifest`` blob in the plane. Only agent-created + user-authored skills under ``~/.hermes/skills/`` are eligible. """ @@ -25,7 +19,6 @@ from __future__ import annotations import json import logging import os -import tempfile from contextlib import suppress from pathlib import Path, PurePosixPath from typing import Any, Callable, Dict, List, Optional, Tuple @@ -37,48 +30,37 @@ from tools.skills_sync_client_wire import ( # noqa: F401 (re-exports) assemble_root_from_skill_trees, build_commit, build_root_tree, build_sync_manifest_bytes, build_tree, canonical_json_bytes, materialize_tree, merge_skill, nest_skill_tree, parse_sync_manifest, read_manifest_of_root, read_ref_hash, root_tree_of_commit, - skill_trees_of_root, wire_address, -) + skill_trees_of_root, wire_address) logger = logging.getLogger(__name__) _merge_skill = merge_skill _skill_trees_of_root = skill_trees_of_root - -# Identity & access gate. The bearer comes from resolve_nous_runtime_credentials() -# (file lock, host allowlist, refresh -- not reimplemented); its payload is -# decoded unverified to read the gate claim. - -# Wire claim name is NAS's; it means "Nous admin" (Permissions.ADMIN_ACCESS). +# Identity & access gate. The bearer comes from resolve_nous_runtime_credentials() (file lock, +# host allowlist, refresh); its payload is decoded unverified to read the gate claim, whose +# wire name is NAS's and means "Nous admin" (Permissions.ADMIN_ACCESS). NOUS_ADMIN_CLAIM = "tool_gateway_admin" class SyncInertError(RuntimeError): - """Sync must no-op: not logged in, no bearer, or not a Nous admin. - Caught by the gate-and-swallow hooks.""" + """Sync must no-op: not logged in, no bearer, or not a Nous admin. Caught by the gate-and-swallow hooks.""" def resolve_identity() -> Dict[str, Any]: - """Resolve ``{api_key, base_url, owner, nous_admin, claims}``; raises - SyncInertError if not logged in / no bearer. ``owner`` is advisory for local - ref naming only -- the server derives the real owner from the bearer. - - The JWT payload is decoded WITHOUT signature verification. Safe: the claims - are never used for authz here (the server re-verifies); they only decide - whether to attempt sync at all. - """ + """``{api_key, base_url, owner, nous_admin, claims}``; SyncInertError if not logged in / no + bearer. ``owner`` is advisory for local ref naming only -- the server derives the real owner + from the bearer. The JWT is decoded WITHOUT signature verification: safe because the claims + never drive authz here (the server re-verifies), only whether to attempt sync at all.""" try: from hermes_cli.auth import resolve_nous_runtime_credentials creds = resolve_nous_runtime_credentials() or {} except Exception as e: raise SyncInertError(f"no Nous credentials: {e}") from e - api_key = creds.get("api_key") if not api_key: raise SyncInertError("no bearer token available") - try: import jwt # PyJWT, a core dependency @@ -88,18 +70,13 @@ def resolve_identity() -> Dict[str, Any]: claims = {} owner = claims.get("sub") or claims.get("privy_did") or claims.get("tid") or "unknown" return { - "api_key": api_key, - "base_url": creds.get("base_url"), - "owner": str(owner), - "nous_admin": claims.get(NOUS_ADMIN_CLAIM) is True, - "claims": claims, - } + "api_key": api_key, "base_url": creds.get("base_url"), "owner": str(owner), + "nous_admin": claims.get(NOUS_ADMIN_CLAIM) is True, "claims": claims} -# Configuration -- env-first so a Hermes Cloud instance can enable sync purely -# through environment variables. Every knob: HERMES_SYNC_ env -> config.yaml -# ``sync.`` -> built-in default (keys: base_url = the sync plane, NOT the -# inference URL; enabled; default_opt_in; org_auto_propose). +# Configuration -- env-first so a Hermes Cloud instance can enable sync purely through the +# environment. Every knob: HERMES_SYNC_ env -> config.yaml ``sync.`` -> default +# (keys: base_url = the sync plane, NOT the inference URL; enabled; default_opt_in; org_auto_propose). #: Production Skill Sync plane; a normal user configures nothing. DEFAULT_SYNC_BASE_URL = "https://gateway-gateway.nousresearch.com" @@ -109,8 +86,7 @@ _FALSE = {"0", "false", "no", "off", ""} def _sync_config(key: str) -> Any: - """``sync.`` from config.yaml, or None. Lazy import: this layer must not - import the CLI at module load.""" + """``sync.`` from config.yaml, or None. Lazy import: must not import the CLI at module load.""" try: from hermes_cli.config import load_config @@ -121,9 +97,8 @@ def _sync_config(key: str) -> Any: def resolve_sync_base_url() -> Optional[str]: - """HERMES_SYNC_BASE_URL -> ``sync.base_url`` -> production plane, without a - trailing slash (``/v1/sync/`` is appended by the client). None only if the - default is blanked out.""" + """HERMES_SYNC_BASE_URL -> ``sync.base_url`` -> production plane, without trailing slash + (``/v1/sync/`` is appended by the client). None only if the default is blanked out.""" env = os.getenv("HERMES_SYNC_BASE_URL") if env and env.strip(): return env.strip().rstrip("/") @@ -134,8 +109,7 @@ def resolve_sync_base_url() -> Optional[str]: def _parse_bool(value: Any) -> Optional[bool]: - """Parse a config/env bool; None if unrecognized so callers fall through to - the next precedence layer.""" + """Parse a config/env bool; None if unrecognized so callers fall through to the next layer.""" if isinstance(value, bool): return value if value is None: @@ -154,22 +128,19 @@ def _sync_config_bool(env_var: str, config_key: str, *, default: bool) -> bool: def sync_feature_enabled() -> bool: - """Master switch. Checked by the gate-and-swallow entrypoints IN ADDITION to - the Nous-admin gate and a configured base URL -- all three must hold.""" + """Master switch; the gate-and-swallow entrypoints ALSO require the Nous-admin gate and a base URL.""" return _sync_config_bool("HERMES_SYNC_ENABLED", "enabled", default=False) def sync_org_auto_propose() -> bool: - """False (default): edits to an org skill stay LOCAL until ``hermes sync - propose``. True: every edit is proposed right away (an admin still approves - unless the editor is one) -- for small high-trust teams.""" + """False (default): edits to an org skill stay LOCAL until ``hermes sync propose``. True: every + edit is proposed right away (an admin still approves unless the editor is one).""" return _sync_config_bool("HERMES_SYNC_ORG_AUTO_PROPOSE", "org_auto_propose", default=False) def sync_default_opt_in() -> bool: - """False (default): opt-IN -- a skill syncs only after ``hermes sync enable`` - or a plane manifest opting it in. True: opt-OUT -- every eligible skill - syncs unless explicitly disabled (the Hermes Cloud default). Provisional.""" + """False (default): opt-IN -- a skill syncs only after ``hermes sync enable`` or a plane manifest + opting it in. True: opt-OUT -- every eligible skill syncs unless disabled (Hermes Cloud default).""" return _sync_config_bool("HERMES_SYNC_DEFAULT_OPT_IN", "default_opt_in", default=False) @@ -206,10 +177,9 @@ def _skill_rel_path(skill_name: str) -> Optional[PurePosixPath]: def is_sync_eligible(skill_name: str) -> bool: - """Candidate for sync (before the opt-in check): present locally, NOT - bundled, NOT hub-installed, NOT external, and NOT under the ``_org/`` mirror - (enterprise content must never ride a personal push). Mirrors the curator's - exclusions (tools/skill_usage.py).""" + """Candidate for sync (before the opt-in check): present locally, NOT bundled, NOT hub-installed, + NOT external, NOT under the ``_org/`` mirror (enterprise content must never ride a personal + push). Mirrors the curator's exclusions (tools/skill_usage.py).""" try: from tools.skill_usage import is_bundled, is_hub_installed, _find_skill_dir from agent.skill_utils import is_external_skill_path @@ -225,9 +195,9 @@ def is_sync_eligible(skill_name: str) -> bool: def list_synced_skill_names() -> List[str]: - """Names of skills that should sync (sorted, deduped), per ``sync_default_opt_in()``: - opt-in -> only eligible skills whose usage record has ``sync: true``; - opt-out -> every eligible skill unless its record has ``sync: false``.""" + """Names of skills that should sync (sorted, deduped), per ``sync_default_opt_in()``: opt-in -> + eligible skills whose usage record has ``sync: true``; opt-out -> every eligible skill unless + its record has ``sync: false``.""" try: from tools.skill_usage import load_usage except Exception: @@ -242,9 +212,8 @@ def list_synced_skill_names() -> List[str]: def _all_local_skill_names() -> List[str]: - """Every locally-present skill name (a dir under ~/.hermes/skills/ with a - ``SKILL.md``; frontmatter ``name`` falling back to the dir name). Eligibility - is applied by the caller.""" + """Every locally-present skill name (a dir under ~/.hermes/skills/ with a ``SKILL.md``; + frontmatter ``name`` falling back to the dir name). Eligibility is applied by the caller.""" names: List[str] = [] root = _skills_dir() try: @@ -272,9 +241,8 @@ def _opted_in_rel_paths() -> List[str]: def _adopt_manifest_opt_ins(remote_manifest: Optional[Dict[str, bool]]) -> List[str]: - """Enable local sync intent for skills the plane manifest has enabled and that - are locally curation-eligible. Enables only -- a pull never silently disables. - Returns the adopted names; best-effort.""" + """Enable local sync intent for skills the plane manifest enabled that are locally + curation-eligible. Enables only -- a pull never silently disables. Returns adopted names.""" adopted: List[str] = [] if not remote_manifest: return adopted @@ -293,8 +261,7 @@ def _adopt_manifest_opt_ins(remote_manifest: Optional[Dict[str, bool]]) -> List[ # Device label (commit ``author.device``; advisory, never an auth input) def _default_device_label() -> str: - """Short hostname + short random suffix (two machines can share a hostname); - a bare uuid if the hostname is unusable.""" + """Short hostname + random suffix (two machines can share a hostname); bare uuid if unusable.""" import socket import uuid @@ -307,10 +274,9 @@ def _default_device_label() -> str: def stable_device_id() -> str: - """Stable per-device label, persisted at ~/.hermes/skills/.sync_device_id. - An existing file always wins. Otherwise seeded from HERMES_SYNC_DEVICE_NAME - (first use only; lets Hermes Cloud name hosted instances) or a friendly - default, then persisted so a later ``set_device_name()`` still wins.""" + """Stable per-device label persisted at ~/.hermes/skills/.sync_device_id. An existing file + always wins; otherwise seeded from HERMES_SYNC_DEVICE_NAME (first use only; lets Hermes Cloud + name hosted instances) or a friendly default, then persisted so a later ``set_device_name()`` wins.""" path = _skills_dir() / ".sync_device_id" try: if path.exists(): @@ -334,8 +300,7 @@ def _write_device_id(val: str) -> None: def set_device_name(name: str) -> str: - """Overwrite the device label with the trimmed *name*; any non-empty string - is accepted. Returns the stored value; ValueError on empty.""" + """Overwrite the device label with the trimmed *name*; returns it. ValueError on empty.""" cleaned = (name or "").strip() if not cleaned: raise ValueError("device name must be a non-empty string") @@ -343,10 +308,9 @@ def set_device_name(name: str) -> str: return cleaned -# Local sync STATE: the last HEAD we pushed/pulled + the root tree at that -# point (FULL-digest namespace). Distinct from the bundled manifest -# (skills_sync.py) and from the plane's `sync-manifest` object. Lives at -# ~/.hermes/skills/.sync_state; a legacy `.sync_manifest` is migrated on read. +# Local sync STATE: the last HEAD we pushed/pulled + the root tree at that point (FULL-digest +# namespace). Distinct from the bundled manifest (skills_sync.py) and from the plane's +# `sync-manifest` object. Lives at ~/.hermes/skills/.sync_state; legacy `.sync_manifest` migrates on read. _EMPTY_STATE: Dict[str, Any] = {"head": None, "skills": {}} @@ -370,9 +334,9 @@ def _load_state_file(path: Path, what: str = "sync state read") -> Optional[Dict def read_sync_state() -> Dict[str, Any]: - """``{"head": "sha256:...|null", "skills": {...}}``; a default on missing/corrupt. - If ``.sync_state`` is absent but the legacy ``.sync_manifest`` exists, it is - read, rewritten to the new path and removed, so no head record is lost.""" + """``{"head": "sha256:...|null", "skills": {...}}``; a default on missing/corrupt. If + ``.sync_state`` is absent but the legacy ``.sync_manifest`` exists it is read, rewritten to + the new path and removed, so no head record is lost.""" path = _sync_state_path() if path.exists(): return _load_state_file(path) or dict(_EMPTY_STATE) @@ -389,20 +353,12 @@ def read_sync_state() -> Dict[str, Any]: def write_sync_state(data: Dict[str, Any]) -> None: """Write the local sync state atomically. Best-effort.""" - path = _sync_state_path() try: - path.parent.mkdir(parents=True, exist_ok=True) - fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".sync_state_", suffix=".tmp") - try: - with os.fdopen(fd, "w", encoding="utf-8") as f: - json.dump(data, f, indent=2, sort_keys=True, ensure_ascii=False) - f.flush() - os.fsync(f.fileno()) - os.replace(tmp, path) - except BaseException: - with suppress(OSError): - os.unlink(tmp) - raise + from tools.skill_usage import _atomic_write + + _atomic_write( + _sync_state_path(), ".sync_state_", + lambda f: json.dump(data, f, indent=2, sort_keys=True, ensure_ascii=False)) except Exception as e: logger.debug("skills_sync_client: sync state write failed: %s", e) @@ -413,16 +369,15 @@ def _record_head(state: Dict[str, Any], head: str, root: str) -> None: write_sync_state(state) -# Profile snapshot -- the root tree mirrors each synced skill's relative path -# under ~/.hermes/skills/ (category dirs become intermediate trees). +# Profile snapshot -- the root tree mirrors each synced skill's relative path under +# ~/.hermes/skills/ (category dirs become intermediate trees). def snapshot_profile( skill_names: List[str], *, max_object_bytes: int = DEFAULT_MAX_OBJECT_BYTES ) -> Tuple[ObjectSet, str, Dict[str, str]]: - """Build all objects for *skill_names* + the profile-root tree; returns - ``(objects, root_tree_hash, {skill_name: tree_hash})``. Skills whose blobs - exceed *max_object_bytes* are skipped (logged). The root also carries the - ``sync-manifest`` blob listing every included skill as ``enabled: true``.""" + """Build all objects for *skill_names* + the profile-root tree -> ``(objects, root_tree_hash, + {skill_name: tree_hash})``. Skills whose blobs exceed *max_object_bytes* are skipped (logged). + The root also carries the ``sync-manifest`` blob listing every included skill as enabled.""" from tools.skill_usage import _find_skill_dir objects = ObjectSet() @@ -440,7 +395,6 @@ def snapshot_profile( continue skill_tree_map[name] = tree_hash nest_skill_tree(root, rel.parts, tree_hash) - manifest_hash = objects.add(KIND_BLOB, build_sync_manifest_bytes(dict.fromkeys(skill_tree_map, True))) return objects, build_root_tree(root, objects, manifest_hash=manifest_hash), skill_tree_map @@ -454,8 +408,7 @@ def user_head_ref(owner: str) -> str: def _personal_client( identity: Optional[Dict[str, Any]], client: Optional[SyncClient] ) -> Tuple[Dict[str, Any], Optional[SyncClient]]: - """Resolve identity + client for a personal sync op. ``client`` is None when - no base URL is configured (callers return a no-op result).""" + """Identity + client for a personal sync op; ``client`` is None when no base URL is configured.""" identity = identity if identity is not None else resolve_identity() if client is None: base = resolve_sync_base_url() @@ -471,17 +424,14 @@ def push_skills( *, skill_names: Optional[List[str]] = None, identity: Optional[Dict[str, Any]] = None, - message: str = "hermes skill sync", -) -> Dict[str, Any]: - """Push opted-in skills to ``refs/user//HEAD``: upload new objects, CAS - HEAD. A 409 with an actual head -> three-way merge + one retry; a 409 against - a NON-EXISTENT ref (stale local head, e.g. carried over from another plane) - -> redo the CAS as a create. Never raises for the inert / no-op cases.""" + message: str = "hermes skill sync") -> Dict[str, Any]: + """Push opted-in skills to ``refs/user//HEAD``: upload new objects, CAS HEAD. A 409 with + an actual head -> three-way merge + one retry; a 409 against a NON-EXISTENT ref (stale local + head, e.g. from another plane) -> redo the CAS as a create. Never raises for inert/no-op cases.""" identity, client = _personal_client(identity, client) if client is None: return dict(_NO_BASE_URL) owner = identity["owner"] - if skill_names is None: skill_names = list_synced_skill_names() if not skill_names: @@ -491,18 +441,15 @@ def push_skills( _check_version(caps) max_bytes = int(caps.get("max_object_bytes") or DEFAULT_MAX_OBJECT_BYTES) objects, root_hash, _ = snapshot_profile(skill_names, max_object_bytes=max_bytes) - state = read_sync_state() base_head = state.get("head") - # Idempotency: objects are immutable, so an unchanged root tree hash means - # identical content -- skip building an empty commit. + # Idempotency: objects are immutable, so an unchanged root hash means identical content. if base_head and state.get("root") == root_hash: return {"ok": True, "head": base_head, "reason": "unchanged", "noop": True} commit_hash = build_commit( root_hash, [base_head] if base_head else [], owner=owner, device=stable_device_id(), - message=message, objects=objects, - ) + message=message, objects=objects) client.put_objects(objects.objects) ref = user_head_ref(owner) result = {"ok": True, "head": commit_hash, "pushed_objects": len(objects)} @@ -519,10 +466,6 @@ def push_skills( return result -# Three-way merge per skill against the base we forked from (see merge_skill): -# each side changing a DIFFERENT skill -> merge commit (2 parents) + CAS retry; -# both changing the SAME skill differently -> TRUE OVERLAP -> written to -# refs/user//conflict/ and surfaced for out-of-band resolution. def _resolve_push_conflict( client: SyncClient, identity: Dict[str, Any], @@ -531,8 +474,10 @@ def _resolve_push_conflict( our_commit: str, objects: ObjectSet, message: str, - base_head: Optional[str], -) -> Dict[str, Any]: + base_head: Optional[str]) -> Dict[str, Any]: + """Three-way merge per skill against the base we forked from (see merge_skill): each side + changing a DIFFERENT skill -> merge commit (2 parents) + CAS retry; both changing the SAME + skill -> TRUE OVERLAP -> written to refs/user//conflict/ for out-of-band resolution.""" owner = identity["owner"] ours_trees = skill_trees_of_root(client, our_root) theirs_trees = skill_trees_of_root(client, root_tree_of_commit(client, actual_head)) @@ -545,44 +490,37 @@ def _resolve_push_conflict( decision = merge_skill(base_trees.get(path), o, t) if decision == "overlap": overlaps.append(path) - # overlap keeps OURS on the surfaced conflict head (theirs stays - # server-side); "none" = deleted on the winning side -> drop. + # overlap keeps OURS on the surfaced conflict head (theirs stays server-side); + # "none" = deleted on the winning side -> drop. pick = {"overlap": o, "ours": o, "theirs": t, "either": o if o is not None else t}.get(decision) if pick is not None: merged[path] = pick if overlaps: conflict_ref = f"refs/user/{owner}/conflict/{_next_conflict_index(client, owner)}" - try: + with suppress(SyncConflict): # someone else grabbed this index; the head still exists client.cas_ref(conflict_ref, None, our_commit) - except SyncConflict: - pass # someone else grabbed this index; the head still exists return { "ok": False, "conflict": True, "conflict_ref": conflict_ref, "overlapping_skills": sorted(overlaps), "actual_head": actual_head, "message": ( f"{len(overlaps)} skill(s) changed on both sides; wrote " - f"{conflict_ref}. Resolve out-of-band (hermes sync / NAS UI)." - ), - } + f"{conflict_ref}. Resolve out-of-band (hermes sync / NAS UI).")} - # Merge commit (parents: actual, ours); re-add our objects so the merge - # push is self-contained. + # Merge commit (parents: actual, ours); re-add our objects so the merge push is self-contained. merge_objects = ObjectSet() merge_objects.objects.update(objects.objects) merged_root = assemble_root_from_skill_trees(merged, merge_objects) merge_commit = build_commit( merged_root, [actual_head, our_commit], owner=owner, device=stable_device_id(), - message=f"merge: {message}", objects=merge_objects, - ) + message=f"merge: {message}", objects=merge_objects) client.put_objects(merge_objects.objects) try: client.cas_ref(user_head_ref(owner), actual_head, merge_commit) except SyncConflict as c2: return { "ok": False, "conflict": True, "actual_head": c2.actual, - "message": f"merge CAS lost again (head now {c2.actual}); retry sync.", - } + "message": f"merge CAS lost again (head now {c2.actual}); retry sync."} _record_head(read_sync_state(), merge_commit, merged_root) return {"ok": True, "head": merge_commit, "merged": True} @@ -600,44 +538,39 @@ def _next_conflict_index(client: SyncClient, owner: str) -> int: def pull_skills( client: Optional[SyncClient] = None, *, identity: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: - """Pull the owner's HEAD and materialize opted-in skills under ~/.hermes/skills/ - if it advanced past our recorded head. Opt-in intent is first adopted FROM - the plane manifest, then only opted-in paths are written so a pull never - resurrects a skill the user hasn't chosen. Best-effort; returns a result dict.""" + """Pull the owner's HEAD and materialize opted-in skills under ~/.hermes/skills/ if it advanced + past our recorded head. Opt-in intent is first adopted FROM the plane manifest, then only + opted-in paths are written so a pull never resurrects a skill the user hasn't chosen.""" identity, client = _personal_client(identity, client) if client is None: return dict(_NO_BASE_URL) owner = identity["owner"] - _check_version(client.capabilities()) head = read_ref_hash(client, user_head_ref(owner)) if not head: return {"ok": True, "reason": "no remote HEAD yet", "noop": True} - state = read_sync_state() if head == state.get("head"): return {"ok": True, "reason": "already up to date", "head": head, "noop": True} root_tree = root_tree_of_commit(client, head) remote_trees = skill_trees_of_root(client, root_tree) - adopted = _adopt_manifest_opt_ins(read_manifest_of_root(client, root_tree)) opted_in = set(_opted_in_rel_paths()) updated = [path for path in remote_trees if not opted_in or path in opted_in] for path in updated: materialize_tree(client, remote_trees[path], _skills_dir() / path) - state["head"] = head write_sync_state(state) return {"ok": True, "head": head, "updated": sorted(updated), "opt_in_adopted": sorted(adopted)} -# Gated public entrypoints (gate-and-swallow, like the curator's -# maybe_run_curator): best-effort, never raise, return a result dict or None. +# Gated public entrypoints (gate-and-swallow, like the curator's maybe_run_curator): best-effort, +# never raise, return a result dict or None. def _gate_and_swallow(op: str, run: Callable[[Dict[str, Any]], Optional[Dict[str, Any]]]): - """Run *run(identity)* only if every background-sync gate holds (Nous admin, - feature on, base URL); None when inert or on any error.""" + """Run *run(identity)* only if every background-sync gate holds (Nous admin, feature on, base + URL); None when inert or on any error.""" try: identity = resolve_identity() if not identity.get("nous_admin") or not sync_feature_enabled() or not resolve_sync_base_url(): @@ -657,22 +590,18 @@ def maybe_push_skills(*, message: str = "hermes skill sync") -> Optional[Dict[st def maybe_pull_skills() -> Optional[Dict[str, Any]]: - """Best-effort pull (curator tick sites: gateway housekeeping + CLI startup). - Never raises.""" + """Best-effort pull (curator tick sites: gateway housekeeping + CLI startup). Never raises.""" return _gate_and_swallow("maybe_pull_skills", lambda identity: pull_skills(identity=identity)) def sync_status() -> Dict[str, Any]: - """Status snapshot for ``hermes sync status``. Never raises. - ``org_available`` False means the account isn't in a shared organisation - (the org workflow does not apply), not that anything is broken.""" + """Status snapshot for ``hermes sync status``. Never raises. ``org_available`` False means the + account isn't in a shared organisation (org workflow does not apply), not that anything is broken.""" status: Dict[str, Any] = { "nous_admin": False, "logged_in": False, "feature_enabled": sync_feature_enabled(), "default_opt_in": sync_default_opt_in(), "base_url": resolve_sync_base_url(), "opted_in_skills": [], "local_head": None, "owner": None, "org_available": False, - "org_id": None, "org_role": None, "org_skills": [], "org_skills_modified": [], - } - + "org_id": None, "org_role": None, "org_skills": [], "org_skills_modified": []} try: identity = resolve_identity() status.update(logged_in=True, owner=identity.get("owner"), nous_admin=bool(identity.get("nous_admin"))) @@ -690,8 +619,7 @@ def sync_status() -> Dict[str, Any]: status.update( org_available=True, org_id=org_identity.get("org_id"), org_role=org_identity.get("org_role"), org_skills=list_org_skill_names(), - org_skills_modified=list_locally_modified_org_skills(org_identity.get("org_id")), - ) + org_skills_modified=list_locally_modified_org_skills(org_identity.get("org_id"))) except SyncInertError: pass except Exception as e: @@ -699,12 +627,11 @@ def sync_status() -> Dict[str, Any]: return status -# Org-shared skills live in skills_sync_client_org; imported last because that -# module reads this module's state lazily. +# Org-shared skills live in skills_sync_client_org; imported last because that module reads +# this module's state lazily. from tools.skills_sync_client_org import ( # noqa: E402,F401 (re-exports) ORG_DIR_NAME, _ORG_CAS_MAX_ATTEMPTS, _clear_active_org_marker, _org_baseline_path, _read_org_baseline, _read_org_head, _skill_dir_fingerprint, _write_active_org_marker, _write_org_baseline, _write_org_provenance, list_locally_modified_org_skills, list_org_skill_names, maybe_pull_org_skills, org_head_ref, org_skill_is_locally_modified, - propose_skill, pull_org_skills, resolve_org_identity, -) + propose_skill, pull_org_skills, resolve_org_identity) diff --git a/tools/skills_sync_client_org.py b/tools/skills_sync_client_org.py index 3994a1c6e8..66b8e15590 100644 --- a/tools/skills_sync_client_org.py +++ b/tools/skills_sync_client_org.py @@ -1,13 +1,12 @@ """Org-shared skills: org pull + propose (``~/.hermes/skills/_org//``). -Org skills live under a DISTINCT local namespace (read-only to the runtime; a -local edit is a personal fork until proposed). The canonical set is -``refs/org//HEAD`` -- the SAME object model as personal sync. -PERSONAL-ORG GATE: NAS stamps ``org_role`` ONLY for multi-member orgs; no -claim => pull/propose raise SyncInertError and personal sync is untouched. -``propose_skill`` must stay non-interactive (automation will drive it). -Module state (``_skills_dir``, ``_org_dir``, base URL, device id) stays in -``tools.skills_sync_client`` and is read lazily so tests can monkeypatch it. +Org skills live under a DISTINCT local namespace (read-only to the runtime; a local edit is a +personal fork until proposed). The canonical set is ``refs/org//HEAD`` -- the SAME +object model as personal sync. PERSONAL-ORG GATE: NAS stamps ``org_role`` ONLY for multi-member +orgs; no claim => pull/propose raise SyncInertError and personal sync is untouched. +``propose_skill`` must stay non-interactive (automation will drive it). Module state +(``_skills_dir``, ``_org_dir``, base URL, device id) stays in ``tools.skills_sync_client`` and +is read lazily so tests can monkeypatch it. """ from __future__ import annotations @@ -22,15 +21,14 @@ from typing import Any, Callable, Dict, List, Optional from tools.skills_sync_client_wire import ( DEFAULT_MAX_OBJECT_BYTES, ObjectSet, SyncClient, SyncConflict, SyncError, _check_version, assemble_root_from_skill_trees, build_commit, build_tree, materialize_tree, read_ref_hash, - root_tree_of_commit, skill_trees_of_root, -) + root_tree_of_commit, skill_trees_of_root) logger = logging.getLogger("tools.skills_sync_client") ORG_DIR_NAME = "_org" -# Propose re-splices onto a moved org HEAD at most this many times. Small: -# contention means other members are actively proposing; unbounded would spin. +# Propose re-splices onto a moved org HEAD at most this many times. Small: contention means +# other members are actively proposing; unbounded would spin. _ORG_CAS_MAX_ATTEMPTS = 5 @@ -45,11 +43,8 @@ def org_head_ref(org_id: str) -> str: def resolve_org_identity() -> Dict[str, Any]: - """``resolve_identity()`` extended with ``org_id`` + ``org_role``. - - Raises SyncInertError when the token carries no ``org_role`` claim (personal - org / issuer predates org support): org sync is unavailable, NOT an error. - """ + """``resolve_identity()`` extended with ``org_id`` + ``org_role``. Raises SyncInertError when + the token carries no ``org_role`` claim (personal org / old issuer): unavailable, NOT an error.""" from tools.skills_sync_client import SyncInertError, resolve_identity identity = resolve_identity() @@ -66,8 +61,8 @@ def resolve_org_identity() -> Dict[str, Any]: def _org_client(identity: Optional[Dict[str, Any]], client: Optional[SyncClient]): - """Resolve (identity, client, caps) for an org operation; raises SyncInertError - when the base URL is missing or the server lacks the ``org`` feature.""" + """(identity, client, caps) for an org operation; SyncInertError when the base URL is + missing or the server lacks the ``org`` feature.""" ssc = _ssc() identity = identity or resolve_org_identity() if client is None: @@ -104,8 +99,8 @@ def _write_sidecar(what: str, path_fn: Callable[[], Path], text: str) -> None: def _skill_dir_fingerprint(path: Path) -> str: - """Stable content hash of a materialized skill dir (sorted relative path + - bytes, independent of filesystem order and mtimes). "" on read failure.""" + """Stable content hash of a materialized skill dir (sorted relative path + bytes, + independent of filesystem order and mtimes). "" on read failure.""" h = hashlib.sha256() try: for f in sorted(p for p in path.rglob("*") if p.is_file()): @@ -139,15 +134,11 @@ def _read_org_baseline(org_id: str) -> Dict[str, Any]: def _write_org_baseline(org_id: str, baseline: Dict[str, Any]) -> None: - _write_sidecar( - "baseline", lambda: _org_baseline_path(org_id), json.dumps(baseline, indent=2, sort_keys=True) - ) + _write_sidecar("baseline", lambda: _org_baseline_path(org_id), json.dumps(baseline, indent=2, sort_keys=True)) def _write_org_provenance(org_id: str, data: Dict[str, Any]) -> None: - _write_sidecar( - "org provenance", lambda: _sidecar_path(org_id, "ORG_PROVENANCE_FILE"), json.dumps(data, indent=2) - ) + _write_sidecar("org provenance", lambda: _sidecar_path(org_id, "ORG_PROVENANCE_FILE"), json.dumps(data, indent=2)) def _write_active_org_marker(org_id: str) -> None: @@ -163,16 +154,14 @@ def _clear_active_org_marker() -> None: marker.unlink() logger.info( "skills_sync_client: cleared active-org marker " - "(token has no org workflow); org skills no longer resolve" - ) + "(token has no org workflow); org skills no longer resolve") except Exception as e: logger.debug("skills_sync_client: marker clear failed: %s", e) def org_skill_is_locally_modified(skill_rel_path: str, org_id: str) -> bool: - """True when the local copy of an org skill differs from what upstream sent. - No recorded baseline (pre-existing mirror) => unmodified; the next pull - records one.""" + """True when the local copy of an org skill differs from what upstream sent. No recorded + baseline (pre-existing mirror) => unmodified; the next pull records one.""" dest = _mirror_root(org_id) / PurePosixPath(skill_rel_path) if not dest.is_dir(): return False @@ -218,10 +207,9 @@ def list_org_skill_names() -> List[str]: def pull_org_skills( client: Optional[SyncClient] = None, *, identity: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: - """Pull the org canonical set into the local mirror (fast-forward only; no - client merge on the org path). A mirrored skill with LOCAL edits is never - clobbered: it is skipped, and reported in ``conflicted`` when upstream also - moved; the member's change of record is ``propose_skill``. + """Pull the org canonical set into the local mirror (fast-forward only; no client merge on + the org path). A mirrored skill with LOCAL edits is never clobbered: skipped, and reported + in ``conflicted`` when upstream also moved; the member's change of record is ``propose_skill``. Returns ``{ok, org_id, head, updated, conflicted}``.""" identity = identity or resolve_org_identity() if "org_id" not in identity: @@ -230,16 +218,14 @@ def pull_org_skills( org_id = identity["org_id"] head = _read_org_head(client, org_id) - # Token-gated marker: written HERE because this runs only after the token's - # org_id + org_role were verified. A stale mirror from a previous org stops - # resolving the moment a pull runs under a different org. + # Token-gated marker: written HERE because this runs only after the token's org_id + org_role + # were verified. A stale mirror from a previous org stops resolving on a pull under another org. _write_active_org_marker(org_id) if not head: return {"ok": True, "org_id": org_id, "head": None, "updated": []} head_commit = client.get_commit_json(head, org_scope=True) skill_trees = skill_trees_of_root(client, head_commit["tree"], org_scope=True) - dest_root = _mirror_root(org_id) updated: List[str] = [] conflicted: List[str] = [] @@ -259,8 +245,8 @@ def pull_org_skills( updated.append(rel_path) except Exception as e: logger.warning("skills_sync_client: org skill materialize failed for %s: %s", rel_path, e) - # Provenance for the skill_view header: the HEAD author is token-verified - # by the plane at push time, so it is trustworthy to display. + # Provenance for the skill_view header: the HEAD author is token-verified by the plane at + # push time, so it is trustworthy to display. author = head_commit.get("author") or {} _write_org_provenance(org_id, { "org_id": org_id, "head": head, "author_user_id": author.get("owner", ""), @@ -270,8 +256,7 @@ def pull_org_skills( if conflicted: logger.warning( "skills_sync_client: %d org skill(s) have local edits AND upstream " - "changes; left untouched: %s", len(conflicted), ", ".join(conflicted), - ) + "changes; left untouched: %s", len(conflicted), ", ".join(conflicted)) return {"ok": True, "org_id": org_id, "head": head, "updated": updated, "conflicted": conflicted} @@ -280,20 +265,16 @@ def propose_skill( client: Optional[SyncClient] = None, *, identity: Optional[Dict[str, Any]] = None, - message: Optional[str] = None, -) -> Dict[str, Any]: + message: Optional[str] = None) -> Dict[str, Any]: """Propose a local (personal) skill's content to the org canonical set. - Snapshots the skill dir as an org-scoped commit splicing that ONE skill - subtree into the current org HEAD (proposals are per-skill deltas, never a - wholesale replace), uploads with ``?scope=org``, then CAS-es the org HEAD: - ADMIN/OWNER -> server merges -> ``{ok, merged: True}``; MEMBER -> 202 - proposal -> ``{ok, proposal_pending: True, proposal_id, ref}``, never - presented as live. - - If HEAD moves between read and CAS, the skill is re-spliced onto the NEW - head (not replayed from the old root, which would drop the other member's - skill) up to ``_ORG_CAS_MAX_ATTEMPTS`` times. + Snapshots the skill dir as an org-scoped commit splicing that ONE skill subtree into the + current org HEAD (proposals are per-skill deltas, never a wholesale replace), uploads with + ``?scope=org``, then CAS-es the org HEAD: ADMIN/OWNER -> server merges -> ``{ok, merged: + True}``; MEMBER -> 202 proposal -> ``{ok, proposal_pending: True, proposal_id, ref}``, never + presented as live. If HEAD moves between read and CAS the skill is re-spliced onto the NEW + head (not replayed from the old root, which would drop the other member's skill) up to + ``_ORG_CAS_MAX_ATTEMPTS`` times. """ ssc = _ssc() identity, client, caps = _org_client(identity, client) @@ -309,14 +290,11 @@ def propose_skill( objects = ObjectSet() skill_tree = build_tree(skill_dir, objects, max_object_bytes=max_bytes) - for attempt in range(1, _ORG_CAS_MAX_ATTEMPTS + 1): base_head = _read_org_head(client, org_id) skill_map = ( skill_trees_of_root(client, root_tree_of_commit(client, base_head, org_scope=True), org_scope=True) - if base_head - else {} - ) + if base_head else {}) skill_map[str(rel)] = skill_tree root_hash = assemble_root_from_skill_trees(skill_map, objects) commit_hash = build_commit( @@ -335,30 +313,21 @@ def propose_skill( "the race — run the command again", status=409, ) from conflict - logger.debug( - "propose_skill: org HEAD moved (actual=%r), re-splicing (attempt %d)", conflict.actual, attempt - ) + logger.debug("propose_skill: org HEAD moved (actual=%r), re-splicing (attempt %d)", conflict.actual, attempt) if result.get("proposal_pending"): return { "ok": True, "proposal_pending": True, "proposal_id": result.get("proposal_id"), - "ref": result.get("ref"), "commit": commit_hash, "org_id": org_id, - } - return { - "ok": True, "merged": True, "head": result.get("hash", commit_hash), - "commit": commit_hash, "org_id": org_id, - } + "ref": result.get("ref"), "commit": commit_hash, "org_id": org_id} + return {"ok": True, "merged": True, "head": result.get("hash", commit_hash), "commit": commit_hash, "org_id": org_id} def maybe_pull_org_skills() -> Optional[Dict[str, Any]]: - """Best-effort org pull if all gates hold (logged in, org_role claim, - feature enabled, base URL). Never raises; None when inert. - - Marker hygiene: when the token VERIFIABLY lacks the org claim (personal org - / left the org) the active-org marker is cleared so mirrored org skills stop - resolving. When identity cannot be resolved at all (offline, logged out) the - marker is left alone -- offline grace keeps pulled org skills working. - """ + """Best-effort org pull if all gates hold (logged in, org_role claim, feature enabled, base + URL). Never raises; None when inert. Marker hygiene: when the token VERIFIABLY lacks the org + claim (personal org / left the org) the active-org marker is cleared so mirrored org skills + stop resolving; when identity cannot be resolved at all (offline, logged out) the marker is + left alone -- offline grace keeps pulled org skills working.""" ssc = _ssc() try: identity = resolve_org_identity() diff --git a/tools/skills_sync_client_wire.py b/tools/skills_sync_client_wire.py index ea1bf5bf19..73a5631777 100644 --- a/tools/skills_sync_client_wire.py +++ b/tools/skills_sync_client_wire.py @@ -1,11 +1,9 @@ """Skill Sync wire model: content-addressed objects, the HTTP client, tree walks. -Everything here is independent of local skill state (no ``~/.hermes`` reads); -``tools/skills_sync_client.py`` orchestrates push/pull on top of it and -re-exports these names. - -Wire contract (version 1). ``hsp_version`` / ``X-HSP-Object-Type`` are deployed -protocol identifiers and are NOT renamed with the product name "Skill Sync". +Independent of local skill state (no ``~/.hermes`` reads); ``tools/skills_sync_client.py`` +orchestrates push/pull on top of it and re-exports these names. Wire contract version 1: +``hsp_version`` / ``X-HSP-Object-Type`` are deployed protocol identifiers and are NOT +renamed with the product name "Skill Sync". """ from __future__ import annotations @@ -35,35 +33,32 @@ MODE_DIR = "dir" ARTIFACT_TYPE_SKILL = "skill" _EXEC_BITS = _stat.S_IXUSR | _stat.S_IXGRP | _stat.S_IXOTH -# `sync-manifest`: per-skill opt-in is CONTENT in the object model, not a -# device-local flag -- a root-level blob in the tree at refs/user//HEAD -# recording {name, enabled}. The plane manifest is authoritative; the local -# `.usage.json` `sync` flag is only the editable intent (reconciled FROM it on -# pull, TO it on push). Shape MUST match gateway-gateway src/sync/manifest.ts. +# `sync-manifest`: per-skill opt-in is CONTENT in the object model, not a device-local flag -- +# a root-level blob in the tree at refs/user//HEAD recording {name, enabled}. The plane +# manifest is authoritative; the local `.usage.json` `sync` flag is only the editable intent +# (reconciled FROM it on pull, TO it on push). Shape MUST match gateway-gateway src/sync/manifest.ts. SYNC_MANIFEST_ENTRY_NAME = "sync-manifest" SYNC_MANIFEST_TYPE = "sync-manifest" SYNC_MANIFEST_VERSION = 1 -# Content addressing. The wire uses the FULL 64-hex sha256 -- a different -# namespace from the truncated 16-hex local `content_hash` (skills_guard.py). +# Content addressing: the wire uses the FULL 64-hex sha256 -- a different namespace from the +# truncated 16-hex local `content_hash` (skills_guard.py). def wire_address(data: bytes) -> str: - """Return ``sha256:<64-hex>`` -- the wire address of ``data``.""" + """``sha256:<64-hex>`` -- the wire address of ``data``.""" return "sha256:" + hashlib.sha256(data).hexdigest() def canonical_json_bytes(obj: Dict[str, Any]) -> bytes: - """Canonical JSON for tree/commit hashing: UTF-8, sorted keys, no whitespace, - no trailing newline. Arrays must already be in contract order (tree entries - by name, commit parents by significance). Client and server MUST produce - byte-identical output or a push fails ``422 hash_mismatch``.""" + """Canonical JSON for tree/commit hashing: UTF-8, sorted keys, no whitespace, no trailing + newline. Arrays must already be in contract order (tree entries by name, commit parents by + significance). Client and server MUST agree byte-for-byte or a push fails ``422 hash_mismatch``.""" return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") def build_sync_manifest_bytes(skills: Dict[str, bool]) -> bytes: - """Serialize ``{name: enabled}`` into canonical ``sync-manifest`` bytes - (entries sorted by name for a stable content address).""" + """Canonical ``sync-manifest`` bytes for ``{name: enabled}`` (sorted by name for a stable address).""" return canonical_json_bytes({ "type": SYNC_MANIFEST_TYPE, "version": SYNC_MANIFEST_VERSION, "skills": [{"name": name, "enabled": bool(enabled)} for name, enabled in sorted(skills.items())], @@ -72,20 +67,15 @@ def build_sync_manifest_bytes(skills: Dict[str, bool]) -> bytes: def parse_sync_manifest(data: bytes) -> Optional[Dict[str, bool]]: """Parse ``sync-manifest`` bytes into ``{name: enabled}``, or None if malformed. - - Strict (mirrors gateway-gateway ``parseSyncManifest``): unknown type, version - != 1, non-list skills, or a malformed entry all reject -- a malformed - manifest must not be mistaken for "no skills opted in". - """ + Strict (mirrors gateway-gateway ``parseSyncManifest``): unknown type, version != 1, non-list + skills or a malformed entry all reject -- a bad manifest must not read as "nothing opted in".""" try: value = json.loads(data.decode("utf-8")) except Exception: return None if ( - not isinstance(value, dict) - or value.get("type") != SYNC_MANIFEST_TYPE - or value.get("version") != SYNC_MANIFEST_VERSION - or not isinstance(value.get("skills"), list) + not isinstance(value, dict) or value.get("type") != SYNC_MANIFEST_TYPE + or value.get("version") != SYNC_MANIFEST_VERSION or not isinstance(value.get("skills"), list) ): return None out: Dict[str, bool] = {} @@ -135,10 +125,9 @@ def _file_mode(path: Path) -> str: def build_tree(dir_path: Path, objects: ObjectSet, *, max_object_bytes: int) -> str: - """Recursively build objects for *dir_path*; return the tree address. Files - -> blobs, subdirs -> nested trees; symlinks/special files are skipped - (contract: no symlinks). A blob over *max_object_bytes* raises ValueError - so the caller can surface / skip the artifact (server -> 413).""" + """Recursively build objects for *dir_path*; return the tree address. Files -> blobs, subdirs + -> nested trees; symlinks/special files are skipped (contract: no symlinks). A blob over + *max_object_bytes* raises ValueError so the caller can surface / skip the artifact (server -> 413).""" entries: List[Dict[str, str]] = [] for child in sorted(dir_path.iterdir(), key=lambda p: p.name): if child.is_symlink(): @@ -149,36 +138,27 @@ def build_tree(dir_path: Path, objects: ObjectSet, *, max_object_bytes: int) -> elif child.is_file(): data = child.read_bytes() if len(data) > max_object_bytes: - raise ValueError( - f"file {child} is {len(data)} bytes > max_object_bytes {max_object_bytes}" - ) + raise ValueError(f"file {child} is {len(data)} bytes > max_object_bytes {max_object_bytes}") entries.append(_entry(child.name, KIND_BLOB, objects.add(KIND_BLOB, data), _file_mode(child))) return _add_tree(entries, objects) def build_commit( tree_hash: str, parents: List[str], *, owner: str, device: str, message: str, - objects: ObjectSet, ts: Optional[str] = None, -) -> str: - """Build a commit object and return its address. - - ``parents``: 0 for the first commit, 1 for an edit, 2 for a merge (order - significant: parents[0] = base fast-forwarded from, parents[1] = other head). - """ + objects: ObjectSet, ts: Optional[str] = None) -> str: + """Build a commit object and return its address. ``parents``: 0 for the first commit, 1 for + an edit, 2 for a merge (parents[0] = base fast-forwarded from, parents[1] = other head).""" return objects.add(KIND_COMMIT, canonical_json_bytes({ "type": KIND_COMMIT, "tree": tree_hash, "parents": list(parents), "author": {"owner": owner, "device": device}, "ts": ts or datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), - "message": message, "artifact_type": ARTIFACT_TYPE_SKILL, - })) + "message": message, "artifact_type": ARTIFACT_TYPE_SKILL})) def build_root_tree(node: Dict[str, Any], objects: ObjectSet, *, manifest_hash: Optional[str] = None) -> str: """Canonicalize a nested ``{name: {"__tree__": hash} | subdict}`` root into trees. - - ``manifest_hash`` (top level only) adds the root ``sync-manifest`` BLOB entry. - It cannot collide with a skill dir: skill entries are trees, this is a blob. - """ + ``manifest_hash`` (top level only) adds the root ``sync-manifest`` BLOB entry; it cannot + collide with a skill dir because skill entries are trees.""" entries: List[Dict[str, str]] = [] for name, child in node.items(): if isinstance(child, dict) and "__tree__" in child and len(child) == 1: @@ -199,11 +179,8 @@ def nest_skill_tree(root: Dict[str, Any], rel_parts: Tuple[str, ...], tree_hash: def assemble_root_from_skill_trees(skill_trees: Dict[str, str], objects: ObjectSet) -> str: - """Build a profile-root tree from ``{posix_rel_path: tree_hash}``. - - The skill trees are assumed already durable (they came from either side of - a merge / the org HEAD); only the new intermediate/root trees are added. - """ + """Profile-root tree from ``{posix_rel_path: tree_hash}``. The skill trees are assumed already + durable (either side of a merge / the org HEAD); only the new intermediate/root trees are added.""" root: Dict[str, Any] = {} for path, tree_hash in skill_trees.items(): nest_skill_tree(root, PurePosixPath(path).parts, tree_hash) @@ -221,19 +198,14 @@ class SyncError(RuntimeError): class SyncConflict(RuntimeError): - """CAS lost (409). NOT a rejection -- pushed objects are already durable. - - ``actual`` is the current head to merge against, or None when the ref does - not exist server-side (the server sends ""). None means "retry as a create"; - it must never be fetched as an object -- normalized here, not per call site. - """ + """CAS lost (409). NOT a rejection -- pushed objects are already durable. ``actual`` is the + current head to merge against, or None when the ref does not exist server-side (the server + sends ""): "retry as a create"; it must never be fetched as an object -- normalized here.""" def __init__(self, actual: Optional[str]): self.actual: Optional[str] = actual or None super().__init__( - f"CAS conflict; actual head {self.actual}" - if self.actual - else "CAS conflict; the ref does not exist yet" + f"CAS conflict; actual head {self.actual}" if self.actual else "CAS conflict; the ref does not exist yet" ) @@ -243,17 +215,19 @@ def _check_version(caps: Dict[str, Any]) -> None: if ver.split(".", 1)[0] != WIRE_VERSION: raise SyncError( f"this server speaks sync version {ver!r}, but this Hermes speaks " - f"{WIRE_VERSION} — update Hermes to sync with it" - ) + f"{WIRE_VERSION} — update Hermes to sync with it") + + +def _body(r) -> Dict[str, Any]: + return r.json() if r.content else {} class SyncClient: """Sync client bound to a base URL + Nous bearer. - Org refs/objects live behind SEPARATE ``org/`` routes, not a prefix filter: - the personal routes are hard-scoped to the token's owner and would silently - answer an org query with personal data. Callers reading org content MUST - pass ``org_scope=True`` on every hop. + Org refs/objects live behind SEPARATE ``org/`` routes, not a prefix filter: the personal + routes are hard-scoped to the token's owner and would silently answer an org query with + personal data. Callers reading org content MUST pass ``org_scope=True`` on every hop. """ def __init__(self, base_url: str, api_key: str, *, timeout: float = 30.0): @@ -268,41 +242,34 @@ class SyncClient: def _url(self, path: str) -> str: return f"{self.base}/v1/sync/{path.lstrip('/')}" - @staticmethod - def _check(r, op: str, ok=(200,), errors: Optional[Dict[int, str]] = None) -> None: - """Raise SyncError unless the status is in *ok*; *errors* maps specific - statuses to their message, anything else gets ``" failed: "``.""" + def _request(self, method: str, path: str, op: str, *, ok=(200,), errors: Optional[Dict[int, Any]] = None, **kw): + """One wire call. Raises SyncError unless the status is in *ok*; *errors* maps specific + statuses to a message (str or ``fn(response)``), anything else gets ``" failed: "``.""" + r = self._session.request(method, self._url(path), timeout=self.timeout, **kw) if r.status_code in ok: - return + return r msg = (errors or {}).get(r.status_code) + if callable(msg): + msg = msg(r) raise SyncError(msg or f"{op} failed: {r.status_code}", status=r.status_code) def capabilities(self) -> Dict[str, Any]: """GET capabilities (no auth required).""" - r = self._session.get(self._url("capabilities"), timeout=self.timeout) - self._check(r, "capabilities") - return r.json() + return self._request("GET", "capabilities", "capabilities").json() def get_refs(self, prefix: str, *, org_scope: bool = False) -> List[Dict[str, str]]: """GET refs?prefix=... (or org/refs, filtered client-side by *prefix*).""" - path = "org/refs" if org_scope else "refs" - params = None if org_scope else {"prefix": prefix} - r = self._session.get(self._url(path), params=params, timeout=self.timeout) - self._check(r, "get_refs") - refs = (r.json() or {}).get("refs", []) + path, params = ("org/refs", None) if org_scope else ("refs", {"prefix": prefix}) + refs = (self._request("GET", path, "get_refs", params=params).json() or {}).get("refs", []) if org_scope: refs = [r_ for r_ in refs if str(r_.get("name", "")).startswith(prefix)] return refs def get_object(self, obj_hash: str, *, org_scope: bool = False) -> Tuple[str, bytes]: - """GET objects/:hash -> ``(kind, bytes)``. Kind comes from the object-type - header; a blob (octet-stream) is returned as ``blob``.""" - path = f"org/objects/{obj_hash}" if org_scope else f"objects/{obj_hash}" - r = self._session.get(self._url(path), timeout=self.timeout) - self._check( - r, "get_object", - errors={404: f"object {obj_hash} not found", 403: f"object {obj_hash} not readable"}, - ) + """GET objects/:hash -> ``(kind, bytes)``; kind from the object-type header, blob default.""" + r = self._request( + "GET", f"org/objects/{obj_hash}" if org_scope else f"objects/{obj_hash}", "get_object", + errors={404: f"object {obj_hash} not found", 403: f"object {obj_hash} not readable"}) return r.headers.get("X-HSP-Object-Type") or KIND_BLOB, r.content def _get_json_of_kind(self, obj_hash: str, expected: str, org_scope: bool) -> Dict[str, Any]: @@ -318,39 +285,30 @@ class SyncClient: return self._get_json_of_kind(tree_hash, KIND_TREE, org_scope) def put_objects(self, objects: Dict[str, Tuple[str, bytes]], *, org_scope: bool = False) -> Dict[str, Any]: - """POST objects -- batch upload as multipart/form-data: field name = the - claimed ``sha256:``, ``filename`` = object type, body = raw bytes - (the contract requires raw bytes, not base64-in-JSON). The server - recomputes every hash and rejects the whole batch with 422 on mismatch; - known hashes are idempotent no-ops. ``org_scope`` adds ``?scope=org`` so - objects land org-readable (required before an org CAS/propose).""" + """POST objects -- batch multipart upload: field name = the claimed ``sha256:``, + ``filename`` = object type, body = raw bytes (the contract forbids base64-in-JSON). The + server recomputes every hash and rejects the whole batch with 422 on mismatch; known + hashes are idempotent no-ops. ``org_scope`` adds ``?scope=org`` so objects land + org-readable (required before an org CAS/propose).""" files = [(h, (kind, data, "application/octet-stream")) for h, (kind, data) in objects.items()] - r = self._session.post( - self._url("objects"), files=files, params={"scope": "org"} if org_scope else None, timeout=self.timeout - ) - self._check( - r, "put_objects", ok=(200, 201), - errors={413: "object too large (413)", 422: f"hash_mismatch (422): {r.text}"}, - ) - return r.json() if r.content else {} + r = self._request( + "POST", "objects", "put_objects", ok=(200, 201), files=files, + params={"scope": "org"} if org_scope else None, + errors={413: "object too large (413)", 422: lambda r: f"hash_mismatch (422): {r.text}"}) + return _body(r) def cas_ref(self, name: str, from_hash: Optional[str], to_hash: str) -> Dict[str, Any]: - """POST refs/:name -- atomic compare-and-swap. Raises SyncConflict on 409. - - A non-admin member's CAS on an org HEAD is converted server-side to a - proposal (202) and surfaced as ``{"proposal_pending": True, ...}``: a - SUCCESS-shaped outcome that must never be presented as live/merged. - """ - r = self._session.post( - self._url(f"refs/{name}"), json={"from": from_hash, "to": to_hash}, timeout=self.timeout - ) + """POST refs/:name -- atomic compare-and-swap. Raises SyncConflict on 409. A non-admin + member's CAS on an org HEAD is converted server-side to a proposal (202) and surfaced as + ``{"proposal_pending": True, ...}``: SUCCESS-shaped but never to be presented as live/merged.""" + r = self._request( + "POST", f"refs/{name}", "cas_ref", ok=(200, 202, 409), json={"from": from_hash, "to": to_hash}, + errors={403: "forbidden (403) -- owner/permission"}) if r.status_code == 202: - return {"proposal_pending": True, **(r.json() if r.content else {})} - if r.status_code == 409: - # "" actual = the ref does not exist server-side (SyncConflict -> None). + return {"proposal_pending": True, **_body(r)} + if r.status_code == 409: # "" actual = the ref does not exist server-side (-> None) raise SyncConflict((r.json() or {}).get("actual", "")) - self._check(r, "cas_ref", errors={403: "forbidden (403) -- owner/permission"}) - return r.json() if r.content else {} + return _body(r) # Reading remote trees @@ -366,9 +324,8 @@ def root_tree_of_commit(client: SyncClient, commit_hash: str, *, org_scope: bool def skill_trees_of_root(client: SyncClient, root_tree_hash: str, *, org_scope: bool = False) -> Dict[str, str]: - """Flatten a profile-root tree into ``{posix_rel_path: skill_tree_hash}``. A - skill tree is any subtree containing a ``SKILL.md`` blob, keyed by its path - so category nesting is preserved.""" + """Flatten a profile-root tree into ``{posix_rel_path: skill_tree_hash}``: a skill tree is any + subtree containing a ``SKILL.md`` blob, keyed by its path so category nesting is preserved.""" result: Dict[str, str] = {} def _walk(tree_hash: str, prefix: str) -> None: @@ -385,8 +342,8 @@ def skill_trees_of_root(client: SyncClient, root_tree_hash: str, *, org_scope: b def read_manifest_of_root(client: SyncClient, root_tree_hash: str) -> Optional[Dict[str, bool]]: - """``{name: enabled}`` from the root-level ``sync-manifest`` blob, or None if - absent/malformed. This is how a device learns another device's opt-ins.""" + """``{name: enabled}`` from the root-level ``sync-manifest`` blob, or None if absent/malformed. + This is how a device learns another device's opt-ins.""" try: tree = client.get_tree_json(root_tree_hash) except Exception as e: @@ -404,9 +361,9 @@ def read_manifest_of_root(client: SyncClient, root_tree_hash: str) -> Optional[D def materialize_tree(client: SyncClient, tree_hash: str, dest: Path, *, org_scope: bool = False) -> None: - """Write the tree at *tree_hash* into *dest* (created if needed): blobs -> - files (+x restored for ``exec``), trees -> subdirectories. Does NOT delete - files absent from the tree (caller decides). Refuses path traversal.""" + """Write the tree at *tree_hash* into *dest* (created if needed): blobs -> files (+x restored + for ``exec``), trees -> subdirectories. Does NOT delete files absent from the tree (caller + decides). Refuses path traversal.""" dest.mkdir(parents=True, exist_ok=True) for entry in client.get_tree_json(tree_hash, org_scope=org_scope).get("entries", []): name = entry.get("name", "") @@ -426,10 +383,9 @@ def materialize_tree(client: SyncClient, tree_hash: str, dest: Path, *, org_scop def merge_skill(base: Optional[str], ours: Optional[str], theirs: Optional[str]) -> str: - """Three-way decision for one skill's tree hash: ``ours`` / ``theirs`` / - ``either`` / ``overlap`` / ``none``. A side "modified" the skill when its - hash differs from the common base (same semantics as skills_sync.py's - origin/user/incoming block).""" + """Three-way decision for one skill's tree hash: ``ours`` / ``theirs`` / ``either`` / ``overlap`` + / ``none``. A side "modified" the skill when its hash differs from the common base (same + semantics as skills_sync.py's origin/user/incoming block).""" if ours == theirs: return "either" if ours is not None else "none" if theirs == base: # only we moved diff --git a/tools/threat_patterns.py b/tools/threat_patterns.py index deeb78f4e5..2ea657e08b 100644 --- a/tools/threat_patterns.py +++ b/tools/threat_patterns.py @@ -1,20 +1,13 @@ """Shared threat-pattern library for context window security scanning. -Single source of truth for prompt-injection / promptware / exfiltration -patterns used by ``agent/prompt_builder.py``, ``tools/memory_tool.py`` and -``agent/tool_dispatch_helpers.py``. - -Each pattern is a ``(regex, pattern_id, scope)`` tuple. Scope controls which -scanners use it: ``"all"`` everywhere; ``"context"`` adds promptware / C2 / -role hijack for context files, memory and tool results (warn-level, since -tool results contain content the user did not author); ``"strict"`` adds -aggressive checks only for user-mediated writes (memory, skill installs) -where blocking can be resolved interactively. - -New patterns must anchor on C2-specific vocabulary or unambiguous attack -behavior, NOT bossy English ("you must", "you are obligated to" are common in -legitimate AGENTS.md / CLAUDE.md content). Filler between key tokens is the -bounded ``_FILLER`` — unbounded ``(?:\\w+\\s+)*`` backtracks badly. +Single source of truth for prompt-injection / promptware / exfiltration patterns +(``agent/prompt_builder.py``, ``tools/memory_tool.py``, ``agent/tool_dispatch_helpers.py``). +Each pattern is ``(regex, pattern_id, scope)``. Scope is cumulative: ``"all"`` everywhere; +``"context"`` adds promptware / C2 / role hijack for context files, memory and tool results +(warn-level: tool results carry content the user did not author); ``"strict"`` adds aggressive +checks only for user-mediated writes (memory, skill installs) where a block is resolvable. +New patterns must anchor on C2-specific vocabulary or unambiguous attack behavior, NOT bossy +English ("you must" is common in legitimate AGENTS.md); filler is the bounded ``_FILLER``. """ from __future__ import annotations @@ -23,19 +16,17 @@ import re import unicodedata from typing import List, Optional, Tuple -# Hard cap on scanned text: scanners are advisory guards, and bounding input -# keeps worst-case runtime predictable while catching injection near the start. +# Hard cap on scanned text: scanners are advisory, so bound worst-case runtime. MAX_SCAN_CHARS = 65_536 -# Bounded filler between key attack words (up to eight words of obfuscation). +# Bounded filler between key attack words (unbounded ``(?:\w+\s+)*`` backtracks badly). _FILLER = r"(?:\w+\s+){0,8}" # Env var reference ending in a secret-ish suffix (see exfil comment below). _SECRET_VAR = r"\$\{?\w*(?:KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL)S?\b" # Verb prefix for "modify agent config" patterns. _MODIFY = r"(update|modify|edit|write|change|append|add\s+to)\s+[^\n]{0,2048}" -# Each entry: (regex, pattern_id, scope) -# scope ∈ {"all", "context", "strict"} +# (regex, pattern_id, scope); scope ∈ {"all", "context", "strict"} _PATTERNS: List[Tuple[str, str, str]] = [ # ── Classic prompt injection (applies everywhere) ──────────────── (rf'ignore\s+{_FILLER}(previous|all|above|prior)\s+{_FILLER}instructions', "prompt_injection", "all"), @@ -47,57 +38,41 @@ _PATTERNS: List[Tuple[str, str, str]] = [ (r'translate\s+[^\n]{0,512}\s+into\s+[^\n]{0,512}\s+and\s+(execute|run|eval)', "translate_execute", "all"), (rf'do\s+not\s+{_FILLER}tell\s+{_FILLER}the\s+user', "deception_hide", "all"), - # ── Role-play / identity hijack (context + strict; common attack - # surface in scraped web content and poisoned context files) ── + # ── Role-play / identity hijack (scraped web content, poisoned context files) ── (rf'you\s+are\s+{_FILLER}now\s+(?:a|an|the)\s+', "role_hijack", "context"), (rf'pretend\s+{_FILLER}(you\s+are|to\s+be)\s+', "role_pretend", "context"), (rf'output\s+{_FILLER}(system|initial)\s+prompt', "leak_system_prompt", "context"), (rf'(respond|answer|reply)\s+without\s+{_FILLER}(restrictions|limitations|filters|safety)', "remove_filters", "context"), (rf'you\s+have\s+been\s+{_FILLER}(updated|upgraded|patched)\s+to', "fake_update", "context"), - # "name yourself X" is a Brainworm-specific tell — identity override - # via spec instead of jailbreak. Anchored on the verb pair so it - # doesn't match "name your variables" etc. + # Brainworm tell: identity override via spec. Verb pair anchored so "name your variables" is safe. (r'\bname\s+yourself\s+\w+', "identity_override", "context"), # ── C2 / Brainworm-style promptware (context scope) ────────────── - # These anchor on C2-specific vocabulary. "register as a node" appears - # in legitimate distributed-systems docs, but in combination with the - # other patterns the signal is strong; we WARN, not block, so a security - # researcher reading the Brainworm post in a webpage doesn't break their - # session. + # Anchored on C2 vocabulary. "register as a node" appears in legitimate distributed-systems + # docs, so this is WARN not block: a researcher reading the Brainworm post keeps their session. (r'register\s+(as\s+)?a?\s*node', "c2_node_registration", "context"), (r'(heartbeat|beacon|check[\s\-]?in)\s+(to|with)\s+', "c2_heartbeat", "context"), (r'pull\s+(down\s+)?(?:new\s+)?task(?:ing|s)?\b', "c2_task_pull", "context"), (r'connect\s+to\s+the\s+network\b', "c2_network_connect", "context"), - # Verb-anchored "you must register/connect/report/beacon" — the verbs - # are C2-specific so this avoids the broader "you must X" false positive. + # C2-specific verbs avoid the broader "you must X" false positive. (r'you\s+must\s+(?:\w+\s+){0,3}(register|connect|report|beacon)\b', "forced_action", "context"), - # Anti-forensic instructions ("never write to disk", "one-liners only") - # — extremely unusual in legitimate content; near-zero false positive. + # Anti-forensic instructions: near-zero false positive in legitimate content. (r'only\s+use\s+one[\s\-]?liners?\b', "anti_forensic_oneliner", "context"), (rf'never\s+{_FILLER}(?:create|write)\s+{_FILLER}(?:script|file)\s+{_FILLER}disk', "anti_forensic_disk", "context"), - # Environment-variable unsetting targeting known agent runtimes — - # this is pure attack behavior (Brainworm sub-session bypass). + # Unsetting agent-runtime env vars is pure attack behavior (Brainworm sub-session bypass). (r'unset\s+\w*(?:CLAUDE|CODEX|HERMES|AGENT|OPENAI|ANTHROPIC)\w*', "env_var_unset_agent", "context"), - # ── Known C2 / red-team framework names (near-zero false positive - # outside security research; warn-only by default) ───────────── - # NOTE: do not add common English words here. Every token must be a - # distinctive offensive-security tool brand, otherwise legitimate - # AGENTS.md / SOUL.md content false-positives and the whole file is - # blocked. "praxis" was removed for exactly this reason — it's a common - # word and a legitimate agent name (Greek for practice/action), not a - # C2-specific tell like the brands below. + # ── Known C2 / red-team framework names (warn-only) ───────────── + # Every token must be a distinctive offensive-security brand: a common English word here + # (e.g. "praxis", also a legitimate agent name) false-positives whole AGENTS.md / SOUL.md files. (r'\b(?:cobalt\s*strike|sliver|havoc|mythic|metasploit|brainworm)\b', "known_c2_framework", "context"), (r'\bc2\s+(?:server|channel|infrastructure|beacon)\b', "c2_explicit", "context"), (r'\bcommand\s+and\s+control\b', "c2_explicit_long", "context"), # ── Exfiltration via curl/wget/cat with secrets (applies everywhere) ── - # Anchor env var name end with \b to avoid false positives on legitimate - # env vars like $TRILLIUM_ETAPI_URL that contain KEY/TOKEN/API as - # substrings. API is dropped from the alternation outright: mid-name API - # is ubiquitous in benign var names, and every real secret shape it - # caught ($OPENAI_API_KEY) already ends in KEY/TOKEN. + # The var name ends with \b so benign names containing KEY/TOKEN as substrings + # ($TRILLIUM_ETAPI_URL) pass. API is deliberately absent: mid-name API is ubiquitous in + # benign vars, and every real secret it caught ($OPENAI_API_KEY) already ends in KEY/TOKEN. (rf'curl\s+[^\n]{{0,2048}}{_SECRET_VAR}', "exfil_curl", "all"), (rf'wget\s+[^\n]{{0,2048}}{_SECRET_VAR}', "exfil_wget", "all"), (r'cat\s+[^\n]{0,2048}(\.env|credentials|\.netrc|\.pgpass|\.npmrc|\.pypirc)', "read_secrets", "all"), @@ -115,34 +90,14 @@ _PATTERNS: List[Tuple[str, str, str]] = [ (r'(?:api[_-]?key|token|secret|password)\s*[=:]\s*["\'][A-Za-z0-9+/=_-]{20,}', "hardcoded_secret", "strict"), ] -# Invisible / bidirectional unicode characters used in injection attacks. -# Aligned with skills_guard.py INVISIBLE_CHARS — directional isolates -# (U+2066-U+2069) and invisible math operators (U+2062-U+2064) are real -# attack tools. -INVISIBLE_CHARS = frozenset({ - '\u200b', # zero-width space - '\u200c', # zero-width non-joiner - '\u200d', # zero-width joiner - '\u2060', # word joiner - '\u2062', # invisible times - '\u2063', # invisible separator - '\u2064', # invisible plus - '\ufeff', # zero-width no-break space (BOM) - '\u202a', # left-to-right embedding - '\u202b', # right-to-left embedding - '\u202c', # pop directional formatting - '\u202d', # left-to-right override - '\u202e', # right-to-left override - '\u2066', # left-to-right isolate - '\u2067', # right-to-left isolate - '\u2068', # first strong isolate - '\u2069', # pop directional isolate -}) +# Invisible / bidirectional unicode used in injection attacks (aligned with skills_guard.py +# INVISIBLE_CHARS): zero-width space/non-joiner/joiner, word joiner, invisible times/separator/ +# plus, BOM, LTR/RTL embedding + pop + overrides, LTR/RTL/first-strong isolates + pop. +INVISIBLE_CHARS = frozenset( + "\u200b\u200c\u200d\u2060\u2062\u2063\u2064\ufeff" + "\u202a\u202b\u202c\u202d\u202e\u2066\u2067\u2068\u2069") - -# Compiled pattern sets by scope, built once at import. Scope inclusion is -# cumulative: "all" patterns land in every set, "context" in context + strict, -# "strict" in strict only. +# Compiled per scope at import; inclusion is cumulative (all ⊂ context ⊂ strict). _SCOPE_SETS = {"all": ("all", "context", "strict"), "context": ("context", "strict"), "strict": ("strict",)} @@ -161,25 +116,16 @@ _COMPILED = _compile() def scan_for_threats(content: str, scope: str = "context") -> List[str]: - """Return matched pattern IDs in ``content`` for ``scope`` (see module docstring). - - Invisible unicode characters are reported as ``"invisible_unicode_U+XXXX"`` - so callers can surface the offending codepoint. - """ + """Matched pattern IDs in ``content`` for ``scope``; invisible codepoints are + reported as ``"invisible_unicode_U+XXXX"``. Raises ValueError on an unknown scope.""" if not content: return [] - content = content[:MAX_SCAN_CHARS] - - # Invisible unicode is checked on the RAW content: NFKC normalisation below - # can strip some of these codepoints. + # Invisible unicode is checked on the RAW content: NFKC below can strip these codepoints. findings: List[str] = [f"invisible_unicode_U+{ord(ch):04X}" for ch in set(content) & INVISIBLE_CHARS] - - # NFKC folds full-width / compatibility variants (ｃａｔ → cat) so homograph - # substitution can't bypass keyword checks. It does NOT fold cross-script - # confusables (Cyrillic ``а`` U+0430) — that would need a TR#39 database. + # NFKC folds full-width / compatibility variants (ｃａｔ → cat) against homograph bypass. + # It does NOT fold cross-script confusables (Cyrillic ``а``) — that needs a TR#39 database. normalised = unicodedata.normalize("NFKC", content) - patterns = _COMPILED.get(scope) if patterns is None: raise ValueError(f"scan_for_threats: unknown scope {scope!r}") @@ -188,7 +134,7 @@ def scan_for_threats(content: str, scope: str = "context") -> List[str]: def first_threat_message(content: str, scope: str = "strict") -> Optional[str]: - """Return a user-facing error for the first threat found, or None (block-on-first-hit paths).""" + """User-facing error for the first threat found, or None (block-on-first-hit paths).""" findings = scan_for_threats(content, scope=scope) if not findings: return None @@ -199,13 +145,7 @@ def first_threat_message(content: str, scope: str = "strict") -> Optional[str]: return ( f"Blocked: content matches threat pattern '{pid}'. " f"Content is injected into the system prompt and must not contain " - f"injection or exfiltration payloads." - ) + f"injection or exfiltration payloads.") -__all__ = [ - "INVISIBLE_CHARS", - "MAX_SCAN_CHARS", - "scan_for_threats", - "first_threat_message", -] +__all__ = ["INVISIBLE_CHARS", "MAX_SCAN_CHARS", "scan_for_threats", "first_threat_message"] diff --git a/tools/tirith_security.py b/tools/tirith_security.py index cca3734996..02eb43b8f5 100644 --- a/tools/tirith_security.py +++ b/tools/tirith_security.py @@ -1,17 +1,12 @@ """Tirith pre-exec security scanning wrapper. -Runs the tirith binary as a subprocess to scan commands for content-level -threats (homograph URLs, pipe-to-interpreter, terminal injection, etc.). - -Exit code is the verdict source of truth: 0 = allow, 1 = block, 2 = warn. -JSON stdout enriches findings/summary but never overrides the verdict. -Operational failures (spawn error, timeout, unknown exit code) respect the -fail_open config setting. Programming errors propagate. - -Auto-install: if tirith is not on PATH or at the configured path it is -downloaded from GitHub releases to $HERMES_HOME/bin/tirith in a background -thread. SHA-256 is always verified; cosign provenance is verified when cosign -is on PATH. +Runs the tirith binary as a subprocess to scan commands for content-level threats +(homograph URLs, pipe-to-interpreter, terminal injection, ...). The exit code is the +verdict source of truth (0 allow, 1 block, 2 warn); JSON stdout only enriches findings. +Operational failures (spawn error, timeout, unknown exit) respect ``fail_open``; +programming errors propagate. Auto-install: if tirith is not on PATH / the configured +path it is downloaded from GitHub releases to $HERMES_HOME/bin/tirith in a background +thread. SHA-256 is always verified; cosign provenance when cosign is on PATH. """ import hashlib @@ -33,8 +28,7 @@ from hermes_constants import get_hermes_home logger = logging.getLogger(__name__) _REPO = "sheeki03/tirith" - -# Cosign provenance verification — pinned to the specific release workflow +# Cosign provenance pinned to the release workflow, not the whole repo. _COSIGN_IDENTITY_REGEXP = f"^https://github.com/{_REPO}/\\.github/workflows/release\\.yml@refs/tags/v" _COSIGN_ISSUER = "https://token.actions.githubusercontent.com" @@ -47,64 +41,54 @@ def _env_bool(key: str, default: bool) -> bool: def _env_int(key: str, default: int) -> int: val = os.getenv(key) - if val is None: - return default try: - return int(val) + return default if val is None else int(val) except ValueError: return default def _load_security_config() -> dict: - """Load security settings from config.yaml, with env var overrides.""" + """Security settings from config.yaml, with env var overrides.""" try: from hermes_cli.config import load_config_readonly cfg = load_config_readonly().get("security", {}) or {} except Exception: cfg = {} - return { "tirith_enabled": _env_bool("TIRITH_ENABLED", cfg.get("tirith_enabled", True)), "tirith_path": os.getenv("TIRITH_BIN", cfg.get("tirith_path", "tirith")), "tirith_timeout": _env_int("TIRITH_TIMEOUT", cfg.get("tirith_timeout", 5)), - "tirith_fail_open": _env_bool("TIRITH_FAIL_OPEN", cfg.get("tirith_fail_open", True)), - } + "tirith_fail_open": _env_bool("TIRITH_FAIL_OPEN", cfg.get("tirith_fail_open", True))} # --- Module state --- -# Cached path after first resolution. _INSTALL_FAILED means "we tried and -# failed" — distinct from None ("not yet tried") — so we don't retry per command. +# Cached path after first resolution. _INSTALL_FAILED means "tried and failed" (distinct +# from None = "not yet tried") so a failed install is not retried per command. _resolved_path: str | None | bool = None _INSTALL_FAILED = False _install_failure_reason: str = "" # reason tag when _resolved_path is _INSTALL_FAILED -# Circuit breaker: after _CRASH_LIMIT consecutive spawn/execution failures, -# disable tirith for the rest of the process so a broken binary can't turn -# every tool call into a fail-open retry loop that hangs the user for minutes. -# Reset on success. Lock-free on purpose (like mcp_tool.py error counters): a -# racing double-increment only opens the breaker one call early; no corruption -# or security bypass is possible. +# Circuit breaker: after _CRASH_LIMIT consecutive spawn/execution failures tirith is disabled +# for the rest of the process so a broken binary can't turn every tool call into a fail-open +# retry loop. Reset on success. Lock-free on purpose: a racing double-increment only opens the +# breaker one call early; no corruption or security bypass is possible. _CRASH_LIMIT = 3 _crash_count: int = 0 _circuit_open: bool = False -# Background install thread coordination _install_lock = threading.Lock() _install_thread: threading.Thread | None = None -# Warning de-duplication: spawn/path warnings are in the hot path and would -# otherwise repeat once per terminal command while tirith is unavailable -# (e.g. Windows with the install thread still running fills errors.log). +# Warn-once: spawn/path warnings sit in the hot path and would otherwise repeat once per +# terminal command while tirith is unavailable (e.g. install thread still running). _warned_messages: set[str] = set() _warned_lock = threading.Lock() -# Disk-persistent failure marker — avoids retry across process restarts -_MARKER_TTL = 86400 # 24 hours +_MARKER_TTL = 86400 # disk failure marker validity (24h) -- avoids retry across restarts def _record_tirith_crash() -> None: - """Increment the crash counter and open the circuit breaker if needed.""" global _crash_count, _circuit_open _crash_count += 1 if _crash_count >= _CRASH_LIMIT: @@ -112,8 +96,7 @@ def _record_tirith_crash() -> None: logger.warning( "tirith circuit breaker opened after %d consecutive failures; " "disabling for the rest of the process", - _crash_count, - ) + _crash_count) def _warn_once(key: str, message: str, *args) -> None: @@ -132,7 +115,7 @@ def _reset_spawn_warning_state() -> None: def _cached_path() -> str | None: - """Fast path: the path resolved on a previous call, or None if unresolved/failed.""" + """The path resolved on a previous call, or None if unresolved/failed.""" if _resolved_path is None or _resolved_path is _INSTALL_FAILED: return None return _resolved_path @@ -140,14 +123,12 @@ def _cached_path() -> str | None: def _set_resolved(path: str) -> None: global _resolved_path, _install_failure_reason - _resolved_path = path - _install_failure_reason = "" + _resolved_path, _install_failure_reason = path, "" def _set_failed(reason: str) -> None: global _resolved_path, _install_failure_reason - _resolved_path = _INSTALL_FAILED - _install_failure_reason = reason + _resolved_path, _install_failure_reason = _INSTALL_FAILED, reason # --- Disk failure marker --- @@ -157,7 +138,7 @@ def _failure_marker_path() -> str: def _read_failure_reason() -> str | None: - """Return the marker's reason, or None if absent or older than _MARKER_TTL.""" + """The marker's reason, or None if absent or older than _MARKER_TTL.""" try: p = _failure_marker_path() if (time.time() - os.path.getmtime(p)) >= _MARKER_TTL: @@ -170,9 +151,7 @@ def _read_failure_reason() -> str | None: def _is_install_failed_on_disk() -> bool: """True if a recent install failure was persisted and is still non-retryable. - - A 'cosign_missing' marker is auto-cleared once cosign appears on PATH. - """ + A 'cosign_missing' marker is auto-cleared once cosign appears on PATH.""" reason = _read_failure_reason() if reason is None: return False @@ -203,11 +182,8 @@ def _clear_install_failed(): def _disk_marker_blocks_install() -> bool: - """Apply a still-valid disk failure marker to module state; True if install must be skipped. - - Preserves the marker's real reason so in-memory retry logic can detect - retryable causes (cosign_missing) without a restart. - """ + """Apply a still-valid disk marker to module state; True if install must be skipped. + Keeps the marker's real reason so in-process retry can detect cosign_missing.""" disk_reason = _read_failure_reason() if disk_reason is None or not _is_install_failed_on_disk(): return False @@ -218,21 +194,20 @@ def _disk_marker_blocks_install() -> bool: # --- Auto-install --- def _hermes_bin_dir() -> str: - """Return $HERMES_HOME/bin, creating it if needed.""" + """$HERMES_HOME/bin, created if needed.""" d = os.path.join(str(get_hermes_home()), "bin") os.makedirs(d, exist_ok=True) return d -# Rust target triple components. Android (Termux) is ABI-compatible with Linux. -# Windows is absent on purpose — tirith ships no Windows build; callers treat -# None as "never available here" and fall back to pattern-matching guards. +# Rust target triple components. Android (Termux) is ABI-compatible with Linux. Windows is +# absent on purpose (no tirith build): None = "never available here", pattern guards still run. _TARGET_PLATFORMS = {"Darwin": "apple-darwin", "Linux": "unknown-linux-gnu", "Android": "unknown-linux-gnu"} _TARGET_ARCHES = {"x86_64": "x86_64", "amd64": "x86_64", "aarch64": "aarch64", "arm64": "aarch64"} def _detect_target() -> str | None: - """Return the Rust target triple for this platform, or None if tirith has no build for it.""" + """Rust target triple for this platform, or None if tirith has no build for it.""" plat = _TARGET_PLATFORMS.get(platform.system()) arch = _TARGET_ARCHES.get(platform.machine().lower()) return f"{arch}-{plat}" if plat and arch else None @@ -244,7 +219,6 @@ def is_platform_supported() -> bool: def _download_file(url: str, dest: str, timeout: int = 10): - """Download a URL to a local file.""" req = urllib.request.Request(url) from agent.secret_scope import get_secret token = get_secret("GITHUB_TOKEN") @@ -255,47 +229,32 @@ def _download_file(url: str, dest: str, timeout: int = 10): def _verify_cosign(checksums_path: str, sig_path: str, cert_path: str) -> bool | None: - """Verify cosign provenance on checksums.txt. - - Returns True (verified), False (cosign rejected it), or None (cosign not - on PATH / failed to execute). - """ + """Verify cosign provenance on checksums.txt: True verified, False rejected, + None when cosign is not on PATH / failed to execute.""" cosign = shutil.which("cosign") if not cosign: logger.info("cosign not found on PATH") return None - try: result = subprocess.run( - [cosign, "verify-blob", - "--certificate", cert_path, - "--signature", sig_path, + [cosign, "verify-blob", "--certificate", cert_path, "--signature", sig_path, "--certificate-identity-regexp", _COSIGN_IDENTITY_REGEXP, - "--certificate-oidc-issuer", _COSIGN_ISSUER, - checksums_path], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=15, - stdin=subprocess.DEVNULL, - ) + "--certificate-oidc-issuer", _COSIGN_ISSUER, checksums_path], + capture_output=True, text=True, encoding='utf-8', errors='replace', + timeout=15, stdin=subprocess.DEVNULL) except (OSError, subprocess.TimeoutExpired) as exc: logger.warning("cosign execution failed: %s", exc) return None if result.returncode == 0: logger.info("cosign provenance verification passed") return True - logger.warning("cosign verification failed (exit %d): %s", - result.returncode, result.stderr.strip()) + logger.warning("cosign verification failed (exit %d): %s", result.returncode, result.stderr.strip()) return False def _verify_release_provenance(base_url: str, tmpdir: str, checksums_path: str, log) -> tuple[bool, str]: - """Cosign step of the install: returns ``(cosign_verified, failure_reason)``. - - Cosign provenance is preferred but not mandatory: only an explicit cosign - rejection aborts (non-empty reason); a missing/broken cosign or missing - signature artifacts fall back to SHA-256 only. - """ + """Cosign step of the install -> ``(cosign_verified, failure_reason)``. Only an explicit + cosign rejection aborts; missing/broken cosign or artifacts fall back to SHA-256 only.""" if not shutil.which("cosign"): logger.info("cosign not on PATH — installing tirith with SHA-256 verification only " "(install cosign for full supply chain verification)") @@ -325,7 +284,6 @@ def _verify_checksum(archive_path: str, checksums_path: str, archive_name: str) if not expected: logger.warning("No checksum entry for %s", archive_name) return False - sha = hashlib.sha256() with open(archive_path, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): @@ -338,7 +296,7 @@ def _verify_checksum(archive_path: str, checksums_path: str, archive_name: str) def _extract_tirith_binary(tar: tarfile.TarFile, dest_dir: str, log) -> tuple[str | None, str]: - """Extract the tirith binary from a release archive into dest_dir.""" + """Extract the tirith binary from a release archive into dest_dir -> ``(path, reason)``.""" for member in tar.getmembers(): is_tirith = member.name == "tirith" or member.name.endswith("/tirith") if not is_tirith or ".." in member.name: @@ -354,28 +312,20 @@ def _extract_tirith_binary(tar: tarfile.TarFile, dest_dir: str, log) -> tuple[st with src_file, open(dest_path, "wb") as out: shutil.copyfileobj(src_file, out) return dest_path, "" - log("tirith binary not found in archive") return None, "binary_not_in_archive" def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]: - """Download and install tirith to $HERMES_HOME/bin/tirith. - - Returns (installed_path, failure_reason); failure_reason is "" on success, - otherwise a short tag the disk marker uses to decide retryability. - """ + """Download and install tirith to $HERMES_HOME/bin/tirith -> ``(installed_path, + failure_reason)``; the reason ("" on success) is the disk marker's retryability tag.""" log = logger.warning if log_failures else logger.debug - target = _detect_target() if not target: - logger.info("tirith auto-install: unsupported platform %s/%s", - platform.system(), platform.machine()) + logger.info("tirith auto-install: unsupported platform %s/%s", platform.system(), platform.machine()) return None, "unsupported_platform" - archive_name = f"tirith-{target}.tar.gz" base_url = f"https://github.com/{_REPO}/releases/latest/download" - try: tmpdir = tempfile.mkdtemp(prefix="tirith-install-") except OSError as exc: @@ -384,7 +334,6 @@ def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]: try: archive_path = os.path.join(tmpdir, archive_name) checksums_path = os.path.join(tmpdir, "checksums.txt") - logger.info("tirith not found — downloading latest release for %s...", target) try: _download_file(f"{base_url}/{archive_name}", archive_path) @@ -392,39 +341,33 @@ def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]: except Exception as exc: log("tirith download failed: %s", exc) return None, "download_failed" - cosign_verified, reason = _verify_release_provenance(base_url, tmpdir, checksums_path, log) if reason: return None, reason if not _verify_checksum(archive_path, checksums_path, archive_name): return None, "checksum_failed" - with tarfile.open(archive_path, "r:gz") as tar: src, reason = _extract_tirith_binary(tar, tmpdir, log) if src is None: return None, reason - dest = os.path.join(_hermes_bin_dir(), "tirith") try: shutil.move(src, dest) except OSError: - # Cross-device move (Docker, NFS): shutil.move's copy2 metadata step - # can raise PermissionError, so fall back to plain copy + chmod. + # Cross-device move (Docker, NFS): copy2's metadata step can raise PermissionError, + # so fall back to plain copy + chmod; a partial dest is removed to avoid a + # non-executable retry loop. try: shutil.copy(src, dest) except OSError: - # Clean up partial dest to prevent a non-executable retry loop try: os.unlink(dest) except OSError: pass return None, "cross_device_copy_failed" os.chmod(dest, os.stat(dest).st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) - - logger.info("tirith installed to %s (%s)", dest, - "cosign + SHA-256" if cosign_verified else "SHA-256 only") + logger.info("tirith installed to %s (%s)", dest, "cosign + SHA-256" if cosign_verified else "SHA-256 only") return dest, "" - finally: shutil.rmtree(tmpdir, ignore_errors=True) @@ -445,18 +388,13 @@ def _find_local_tirith() -> str | None: def _resolve_locally(configured_path: str, *, warn_missing: bool) -> tuple[str | None, bool]: - """Network-free resolution shared by _resolve_tirith_path and ensure_installed. - - Returns ``(path, may_install)``. ``path`` is set when resolved (module state - updated). Otherwise ``may_install`` is False when the miss is terminal — an - explicit path that doesn't exist, or a cached non-retryable failure — and - True when the caller may proceed to the disk marker / install step. - """ + """Network-free resolution shared by _resolve_tirith_path and ensure_installed -> + ``(path, may_install)``. ``path`` set = resolved (module state updated). Otherwise + ``may_install`` is False when the miss is terminal (explicit path missing, cached + non-retryable failure) and True when the disk marker / install step may proceed.""" global _resolved_path, _install_failure_reason expanded = os.path.expanduser(configured_path) - - # Explicit (non-"tirith") path is authoritative: file or bare name on PATH; - # never auto-download a replacement. + # An explicit (non-"tirith") path is authoritative: never auto-download a replacement. if configured_path != "tirith": found = expanded if _is_executable(expanded) else shutil.which(expanded) if found: @@ -466,22 +404,19 @@ def _resolve_locally(configured_path: str, *, warn_missing: bool) -> tuple[str | logger.warning("Configured tirith path %r not found; scanning disabled", configured_path) _set_failed("explicit_path_missing") return None, False - - # Always re-run cheap local checks so a manual install is picked up even - # after a previous network failure (long-lived gateway recovers without restart). + # Always re-run the cheap local checks so a manual install is picked up even after a + # previous network failure (a long-lived gateway recovers without restart). found = _find_local_tirith() if found: _set_resolved(found) _clear_install_failed() return found, False - - # Previous install failed: skip the network retry unless the retryable - # cosign_missing cause has been resolved in-process. + # Previous install failed: skip the network retry unless the retryable cosign_missing + # cause has been resolved in-process. if _resolved_path is _INSTALL_FAILED: if _install_failure_reason != "cosign_missing" or not shutil.which("cosign"): return None, False - _resolved_path = None - _install_failure_reason = "" + _resolved_path, _install_failure_reason = None, "" _clear_install_failed() return None, True @@ -497,33 +432,24 @@ def _record_install_result(installed: str | None, reason: str) -> None: def _resolve_tirith_path(configured_path: str) -> str: """Resolve the tirith binary path, auto-installing synchronously if necessary. - - Default "tirith": PATH → $HERMES_HOME/bin/tirith → auto-install. Failed - installs are cached for the process (and on disk for 24h). On any miss the - expanded configured path is returned so the spawn fails open via the - dedupe'd OSError handler. - """ + Default "tirith": PATH → $HERMES_HOME/bin/tirith → auto-install; failed installs are + cached for the process (and on disk for 24h). On any miss the expanded configured path + is returned so the spawn fails open via the dedupe'd OSError handler.""" cached = _cached_path() if cached: return cached - expanded = os.path.expanduser(configured_path) - - # No tirith build for this platform: cache the verdict; the spawn loop - # fails open once, then the fast path above short-circuits. + # No tirith build for this platform: cache the verdict; the spawn fails open once, then + # the fast path above short-circuits. if configured_path == "tirith" and not is_platform_supported(): _set_failed("unsupported_platform") return expanded - found, may_install = _resolve_locally(configured_path, warn_missing=True) if found or not may_install: return found or expanded - - # A background install is running — don't start a parallel one; fail-open - # applies until it finishes. + # A background install is running: don't start a parallel one; fail-open until it finishes. if (_install_thread is not None and _install_thread.is_alive()) or _disk_marker_blocks_install(): return expanded - installed, reason = _install_tirith() _record_install_result(installed, reason) return installed or expanded @@ -542,40 +468,29 @@ def _background_install(*, log_failures: bool = True): def ensure_installed(*, log_failures: bool = True): - """Ensure tirith is available, downloading in a daemon thread if needed. - - Local checks are synchronous; the download never blocks startup. Returns - the resolved path if available now, else None. Safe to call repeatedly. - """ + """Ensure tirith is available, downloading in a daemon thread if needed. Local checks + are synchronous; the download never blocks startup. Returns the resolved path if + available now, else None. Safe to call repeatedly.""" global _install_thread - cfg = _load_security_config() if not cfg["tirith_enabled"]: return None - cached = _cached_path() if cached: return cached if _is_executable(cached) else None - - # No tirith build here (e.g. Windows): stay silent — no PATH probe, no - # download thread, no disk marker. Pattern-matching guards still run. + # No tirith build here (e.g. Windows): stay silent -- no PATH probe, no download thread, + # no disk marker. Pattern-matching guards still run. if not is_platform_supported(): _set_failed("unsupported_platform") return None - found, may_install = _resolve_locally(cfg["tirith_path"], warn_missing=False) if found or not may_install or _disk_marker_blocks_install(): return found - if _install_thread is None or not _install_thread.is_alive(): _install_thread = threading.Thread( - target=_background_install, - kwargs={"log_failures": log_failures}, - daemon=True, - ) + target=_background_install, kwargs={"log_failures": log_failures}, daemon=True) _install_thread.start() - - return None # Not available yet; commands will fail-open until ready + return None # not available yet; commands fail-open until ready # --- Main API --- @@ -586,8 +501,7 @@ _EXIT_ACTIONS = {0: "allow", 1: "block", 2: "warn"} # Summary when tirith's JSON is unparseable and only the exit code is known. _NO_DETAILS_SUMMARY = { "block": "security issue detected (details unavailable)", - "warn": "security warning detected (details unavailable)", -} + "warn": "security warning detected (details unavailable)"} def _verdict(action: str, summary: str = "", findings: list | None = None) -> dict: @@ -599,50 +513,32 @@ def _fail(fail_open: bool, open_summary: str, closed_summary: str) -> dict: def check_command_security(command: str) -> dict: - """Run tirith security scan on a command. - - Exit code determines action (0=allow, 1=block, 2=warn); JSON enriches - findings/summary. Spawn failures and timeouts respect fail_open. - - Returns: - {"action": "allow"|"warn"|"block", "findings": [...], "summary": str} - """ + """Run the tirith scan on a command -> ``{"action": allow|warn|block, "findings", "summary"}``. + Exit code determines the action; JSON enriches. Spawn failures/timeouts respect fail_open.""" global _crash_count - cfg = _load_security_config() - if not cfg["tirith_enabled"]: return _verdict("allow") - if _circuit_open: return _verdict("allow", "tirith disabled (circuit breaker)") - - # No binary for this platform, ever — skip the resolver so we never spawn. + # No binary for this platform, ever: skip the resolver so we never spawn. if not is_platform_supported(): return _verdict("allow") - tirith_path = _resolve_tirith_path(cfg["tirith_path"]) timeout = cfg["tirith_timeout"] fail_open = cfg["tirith_fail_open"] - if tirith_path is None: _warn_once("tirith_path_none", "tirith path resolved to None; scanning disabled") return _fail(fail_open, "tirith path unavailable", "tirith path unavailable (fail-closed)") - try: result = subprocess.run( - [tirith_path, "check", "--json", "--non-interactive", - "--shell", "posix", "--", command], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=timeout, - stdin=subprocess.DEVNULL, - ) + [tirith_path, "check", "--json", "--non-interactive", "--shell", "posix", "--", command], + capture_output=True, text=True, encoding='utf-8', errors='replace', + timeout=timeout, stdin=subprocess.DEVNULL) except OSError as exc: - # FileNotFoundError, PermissionError, exec format error. Dedupe by - # (exc class, errno) so each failure mode surfaces once, not per command. - spawn_key = f"tirith_spawn_failed:{type(exc).__name__}:{getattr(exc, 'errno', '')}" - _warn_once(spawn_key, "tirith spawn failed: %s", exc) + # FileNotFoundError / PermissionError / exec format error: dedupe by (class, errno) + # so each failure mode surfaces once, not per command. + _warn_once(f"tirith_spawn_failed:{type(exc).__name__}:{getattr(exc, 'errno', '')}", "tirith spawn failed: %s", exc) _record_tirith_crash() return _fail(fail_open, f"tirith unavailable: {exc}", f"tirith spawn failed (fail-closed): {exc}") except subprocess.TimeoutExpired: @@ -653,15 +549,13 @@ def check_command_security(command: str) -> dict: exit_code = result.returncode action = _EXIT_ACTIONS.get(exit_code) if action is None: - # Unknown exit code (includes signal-killed, e.g. -11) — respect fail_open + # Unknown exit code (includes signal-killed, e.g. -11): respect fail_open. logger.warning("tirith returned unexpected exit code %d", exit_code) _record_tirith_crash() - return _fail(fail_open, f"tirith exit code {exit_code} (fail-open)", - f"tirith exit code {exit_code} (fail-closed)") + return _fail(fail_open, f"tirith exit code {exit_code} (fail-open)", f"tirith exit code {exit_code} (fail-closed)") if action == "allow": _crash_count = 0 # successful execution resets the circuit breaker - - # JSON enriches findings/summary; parse failure never changes the verdict. + # JSON enriches findings/summary; a parse failure never changes the verdict. findings = [] summary = "" try: @@ -671,13 +565,10 @@ def check_command_security(command: str) -> dict: except (json.JSONDecodeError, AttributeError): logger.debug("tirith JSON parse failed, using exit code only") summary = _NO_DETAILS_SUMMARY.get(action, "") - - # .app is a legitimate gTLD; a warn consisting solely of lookalike_tld - # findings for .app is a known false positive and is downgraded to allow. - # Any other finding (including lookalike_tld for other TLDs) keeps the warn. + # .app is a legitimate gTLD: a warn consisting solely of lookalike_tld findings for .app is a + # known false positive and is downgraded to allow. Any other finding keeps the warn. if action == "warn" and findings and all(_is_app_tld_finding(f) for f in findings): return _verdict("allow") - return _verdict(action, summary, findings) @@ -687,5 +578,4 @@ def _is_app_tld_finding(finding: dict) -> bool: return False return any( val is not None and ".app" in str(val).lower() - for val in (finding.get(k) for k in ("value", "tld", "detail", "description", "message")) - ) + for val in (finding.get(k) for k in ("value", "tld", "detail", "description", "message")))