1  """hermes webhook — manage dynamic webhook subscriptions from the CLI.
  2  
  3  Usage:
  4      hermes webhook subscribe <name> [options]
  5      hermes webhook list
  6      hermes webhook remove <name>
  7      hermes webhook test <name> [--payload '{"key": "value"}']
  8  
  9  Subscriptions persist to ~/.hermes/webhook_subscriptions.json and are
 10  hot-reloaded by the webhook adapter without a gateway restart.
 11  """
 12  
 13  import json
 14  import re
 15  import secrets
 16  import time
 17  from pathlib import Path
 18  from typing import Dict
 19  
 20  from hermes_constants import display_hermes_home
 21  from utils import atomic_replace
 22  from hermes_cli.config import cfg_get
 23  
 24  
 25  _SUBSCRIPTIONS_FILENAME = "webhook_subscriptions.json"
 26  
 27  
 28  def _hermes_home() -> Path:
 29      from hermes_constants import get_hermes_home
 30      return get_hermes_home()
 31  
 32  
 33  def _subscriptions_path() -> Path:
 34      return _hermes_home() / _SUBSCRIPTIONS_FILENAME
 35  
 36  
 37  def _load_subscriptions() -> Dict[str, dict]:
 38      path = _subscriptions_path()
 39      if not path.exists():
 40          return {}
 41      try:
 42          data = json.loads(path.read_text(encoding="utf-8"))
 43          return data if isinstance(data, dict) else {}
 44      except Exception:
 45          return {}
 46  
 47  
 48  def _save_subscriptions(subs: Dict[str, dict]) -> None:
 49      path = _subscriptions_path()
 50      path.parent.mkdir(parents=True, exist_ok=True)
 51      tmp_path = path.with_suffix(".tmp")
 52      tmp_path.write_text(
 53          json.dumps(subs, indent=2, ensure_ascii=False),
 54          encoding="utf-8",
 55      )
 56      atomic_replace(tmp_path, path)
 57  
 58  
 59  def _get_webhook_config() -> dict:
 60      """Load webhook platform config. Returns {} if not configured."""
 61      try:
 62          from hermes_cli.config import load_config
 63          cfg = load_config()
 64          return cfg_get(cfg, "platforms", "webhook", default={})
 65      except Exception:
 66          return {}
 67  
 68  
 69  def _is_webhook_enabled() -> bool:
 70      return bool(_get_webhook_config().get("enabled"))
 71  
 72  
 73  def _get_webhook_base_url() -> str:
 74      wh = _get_webhook_config().get("extra", {})
 75      host = wh.get("host", "0.0.0.0")
 76      port = wh.get("port", 8644)
 77      display_host = "localhost" if host == "0.0.0.0" else host
 78      return f"http://{display_host}:{port}"
 79  
 80  
 81  def _setup_hint() -> str:
 82      _dhh = display_hermes_home()
 83      return f"""
 84    Webhook platform is not enabled. To set it up:
 85  
 86    1. Run the gateway setup wizard:
 87       hermes gateway setup
 88  
 89    2. Or manually add to {_dhh}/config.yaml:
 90       platforms:
 91         webhook:
 92           enabled: true
 93           extra:
 94             host: "0.0.0.0"
 95             port: 8644
 96             secret: "your-global-hmac-secret"
 97  
 98    3. Or set environment variables in {_dhh}/.env:
 99       WEBHOOK_ENABLED=true
100       WEBHOOK_PORT=8644
101       WEBHOOK_SECRET=your-global-secret
102  
103    Then start the gateway: hermes gateway run
104  """
105  
106  
107  def _require_webhook_enabled() -> bool:
108      """Check webhook is enabled. Print setup guide and return False if not."""
109      if _is_webhook_enabled():
110          return True
111      print(_setup_hint())
112      return False
113  
114  
115  def webhook_command(args):
116      """Entry point for 'hermes webhook' subcommand."""
117      sub = getattr(args, "webhook_action", None)
118  
119      if not sub:
120          print("Usage: hermes webhook {subscribe|list|remove|test}")
121          print("Run 'hermes webhook --help' for details.")
122          return
123  
124      if not _require_webhook_enabled():
125          return
126  
127      if sub in ("subscribe", "add"):
128          _cmd_subscribe(args)
129      elif sub in ("list", "ls"):
130          _cmd_list(args)
131      elif sub in ("remove", "rm"):
132          _cmd_remove(args)
133      elif sub == "test":
134          _cmd_test(args)
135  
136  
137  def _cmd_subscribe(args):
138      name = args.name.strip().lower().replace(" ", "-")
139      if not re.match(r'^[a-z0-9][a-z0-9_-]*$', name):
140          print(f"Error: Invalid name '{name}'. Use lowercase alphanumeric with hyphens/underscores.")
141          return
142  
143      subs = _load_subscriptions()
144      is_update = name in subs
145  
146      secret = args.secret or secrets.token_urlsafe(32)
147      events = [e.strip() for e in args.events.split(",")] if args.events else []
148  
149      route = {
150          "description": args.description or f"Agent-created subscription: {name}",
151          "events": events,
152          "secret": secret,
153          "prompt": args.prompt or "",
154          "skills": [s.strip() for s in args.skills.split(",")] if args.skills else [],
155          "deliver": args.deliver or "log",
156          "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
157      }
158  
159      if getattr(args, "deliver_only", False):
160          if route["deliver"] == "log":
161              print(
162                  "Error: --deliver-only requires --deliver to be a real target "
163                  "(telegram, discord, slack, github_comment, etc.) — not 'log'."
164              )
165              return
166          route["deliver_only"] = True
167  
168      if args.deliver_chat_id:
169          route["deliver_extra"] = {"chat_id": args.deliver_chat_id}
170  
171      subs[name] = route
172      _save_subscriptions(subs)
173  
174      base_url = _get_webhook_base_url()
175      status = "Updated" if is_update else "Created"
176  
177      print(f"\n  {status} webhook subscription: {name}")
178      print(f"  URL:    {base_url}/webhooks/{name}")
179      print(f"  Secret: {secret}")
180      if events:
181          print(f"  Events: {', '.join(events)}")
182      else:
183          print("  Events: (all)")
184      print(f"  Deliver: {route['deliver']}")
185      if route.get("deliver_only"):
186          print("  Mode: direct delivery (no agent, zero LLM cost)")
187      if route.get("prompt"):
188          prompt_preview = route["prompt"][:80] + ("..." if len(route["prompt"]) > 80 else "")
189          label = "Message" if route.get("deliver_only") else "Prompt"
190          print(f"  {label}: {prompt_preview}")
191      print(f"\n  Configure your service to POST to the URL above.")
192      print(f"  Use the secret for HMAC-SHA256 signature validation.")
193      print(f"  The gateway must be running to receive events (hermes gateway run).\n")
194  
195  
196  def _cmd_list(args):
197      subs = _load_subscriptions()
198      if not subs:
199          print("  No dynamic webhook subscriptions.")
200          print("  Create one with: hermes webhook subscribe <name>")
201          return
202  
203      base_url = _get_webhook_base_url()
204      print(f"\n  {len(subs)} webhook subscription(s):\n")
205      for name, route in subs.items():
206          events = ", ".join(route.get("events", [])) or "(all)"
207          deliver = route.get("deliver", "log")
208          if route.get("deliver_only"):
209              deliver = f"{deliver} (direct — no agent)"
210          desc = route.get("description", "")
211          print(f"  ◆ {name}")
212          if desc:
213              print(f"    {desc}")
214          print(f"    URL:     {base_url}/webhooks/{name}")
215          print(f"    Events:  {events}")
216          print(f"    Deliver: {deliver}")
217          print()
218  
219  
220  def _cmd_remove(args):
221      name = args.name.strip().lower()
222      subs = _load_subscriptions()
223  
224      if name not in subs:
225          print(f"  No subscription named '{name}'.")
226          print("  Note: Static routes from config.yaml cannot be removed here.")
227          return
228  
229      del subs[name]
230      _save_subscriptions(subs)
231      print(f"  Removed webhook subscription: {name}")
232  
233  
234  def _cmd_test(args):
235      """Send a test POST to a webhook route."""
236      name = args.name.strip().lower()
237      subs = _load_subscriptions()
238  
239      if name not in subs:
240          print(f"  No subscription named '{name}'.")
241          return
242  
243      route = subs[name]
244      secret = route.get("secret", "")
245      base_url = _get_webhook_base_url()
246      url = f"{base_url}/webhooks/{name}"
247  
248      payload = args.payload or '{"test": true, "event_type": "test", "message": "Hello from hermes webhook test"}'
249  
250      import hmac
251      import hashlib
252      sig = "sha256=" + hmac.new(
253          secret.encode(), payload.encode(), hashlib.sha256
254      ).hexdigest()
255  
256      print(f"  Sending test POST to {url}")
257      try:
258          import urllib.request
259          req = urllib.request.Request(
260              url,
261              data=payload.encode(),
262              headers={
263                  "Content-Type": "application/json",
264                  "X-Hub-Signature-256": sig,
265                  "X-GitHub-Event": "test",
266              },
267              method="POST",
268          )
269          with urllib.request.urlopen(req, timeout=10) as resp:
270              body = resp.read().decode()
271              print(f"  Response ({resp.status}): {body}")
272      except Exception as e:
273          print(f"  Error: {e}")
274          print("  Is the gateway running? (hermes gateway run)")