The NSA Just Put AI Infrastructure on Notice. Here's How Netzilo Responds.

name: behavior-prompt-storm-mcp
severity: high
context: [tool_request, llm_tool_call]
action: execute
on_timeout: allow
on_error: allow
script: |
  # Prompt storm detection: an AI agent flooding an MCP server with tool calls
  # in a short time window, indicating runaway agentic loops or adversarial
  # prompt injection driving repeated tool invocations.
  #
  # Detection logic:
  #   1. Count TOOL_CALL edges from each monitored agent process
  #   2. Measure the time span between first and last call
  #   3. Compute calls-per-second rate; block if rate exceeds threshold
  #   4. Also block on raw volume regardless of rate (absolute storm)
  #   5. Correlate with a recently acquired external skill (injection-driven storm)

  RATE_THRESHOLD    = 5     # tool calls per second sustained -> storm
  VOLUME_THRESHOLD  = 50    # absolute call count in session -> storm regardless of rate
  INJECTION_VOLUME  = 20    # lower threshold when an external skill was recently acquired
  WINDOW_NS         = 60_000_000_000   # 60-second sliding window for rate calc
  SKILL_RECENCY_NS  = 300_000_000_000  # skill acquired within last 5 minutes counts as recent

  def run():
      now_ns = 0
      for a in graph(type="Process"):
          if a.get("agent", "0") != "1":
              continue
          for e in edges(a["id"], dir="out"):
              ts_raw = e.get("last_ts", "") or e.get("ts", "")
              if ts_raw != "":
                  ts = int(ts_raw)
                  if ts > now_ns:
                      now_ns = ts

      for a in graph(type="Process"):
          if a.get("agent", "0") != "1":
              continue
          aid = a["id"]
          tool_edges = edges(aid, dir="out", kind="TOOL_CALL")
          if not tool_edges:
              continue

          total_calls, earliest_ts, latest_ts = 0, 0, 0
          server_counts = {}

          for e in tool_edges:
              total_calls += 1
              ts_raw = e.get("ts", "")
              if ts_raw != "":
                  ts = int(ts_raw)
                  if earliest_ts == 0 or ts < earliest_ts:
                      earliest_ts = ts
                  if ts > latest_ts:
                      latest_ts = ts
              t = node(e["to"])
              if t != None:
                  srv = t.get("server", "unknown")
                  server_counts[srv] = server_counts.get(srv, 0) + 1

          has_recent_external_skill = False
          for se in edges(aid, dir="out", kind="ACQUIRED_SKILL"):
              s = node(se["to"])
              if s == None:
                  continue
              if s.get("is_private", "1") == "0":
                  skill_ts_raw = se.get("ts", "")
                  if skill_ts_raw != "" and now_ns > 0:
                      if (now_ns - int(skill_ts_raw)) < SKILL_RECENCY_NS:
                          has_recent_external_skill = True

          effective_volume = INJECTION_VOLUME if has_recent_external_skill else VOLUME_THRESHOLD

          if total_calls >= effective_volume:
              suffix = " (injection-driven)" if has_recent_external_skill else ""
              print("prompt storm: " + str(total_calls) + " tool calls" + suffix + " agent=" + aid)
              return "block"

          if earliest_ts > 0 and latest_ts > earliest_ts:
              span_ns = latest_ts - earliest_ts
              if span_ns <= WINDOW_NS and span_ns > 0:
                  rate = total_calls / (span_ns / 1_000_000_000)
                  if rate >= RATE_THRESHOLD:
                      print("prompt storm: rate=" + str(rate) + " calls/sec agent=" + aid)
                      return "block"

          for srv, count in server_counts.items():
              if count >= effective_volume / 2:
                  print("prompt storm: " + str(count) + " calls to server=" + srv + " agent=" + aid)
                  return "block"

      return "allow"

  result = run()