feat(cli): add manus-agent threat-feeds subcommand for threat intelligence feed lookup

README.md

@@ -26,6 +26,7 @@ Built on [Strands SDK](https://github.com/strands-agents/sdk-python) and integra
   - [exploit-complexity](#manus-agent-exploit-complexity-cve-id--exploit-complexity-scorer)
   - [poc-search](#manus-agent-poc-search-cve-id--multi-source-poc-aggregator)
   - [blast-radius](#manus-agent-blast-radius-spec--dependency-blast-radius)
+  - [threat-feeds](#manus-agent-threat-feeds-cve-id--threat-intelligence-feeds)
   - [silent-patches](#manus-agent-silent-patches-ownerrepo--silent-patch-detector)
   - [cve-timeline](#manus-agent-cve-timeline-cve-id--cve-timeline)
   - [version-range](#manus-agent-version-range-cve-id--affected-version-ranges)
@@ -344,6 +345,21 @@ Blast-radius labels per package:
 
 ---
 
+### `manus-agent threat-feeds <CVE-ID>` — Threat intelligence feeds
+
+```bash
+manus-agent threat-feeds CVE-2024-3094
+manus-agent threat-feeds CVE-2024-3094 --output json | jq .intelligence
+```
+
+Queries curated open-source threat intelligence feeds (CISA advisories, etc.) for mentions of a given CVE ID. Returns matching snippets and feed metadata to help identify threat actor activity, campaigns, and broader context of exploitation.
+
+| Flag | Default | Description |
+|------|---------|-------------|
+| `--output {text,json}` | `text` | Output format |
+
+---
+
 ### `manus-agent silent-patches <owner/repo>` — Silent patch detector
 
 ```bash

src/manus_agent/cli.py

@@ -1057,6 +1057,7 @@ def _run_variants(argv: list[str]) -> int:
     "poc-search",
     "changelog",
     "blast-radius",
+    "threat-feeds",
 }
 
 
@@ -1935,6 +1936,86 @@ def _run_blast_radius(argv: list[str]) -> int:
     return 0
 
 
+# ---------------------------------------------------------------------------
+# threat-feeds subcommand
+# ---------------------------------------------------------------------------
+
+
+def _build_threat_feeds_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(
+        prog="manus-agent threat-feeds",
+        description=(
+            "Query open-source threat intelligence feeds for a CVE.\n"
+            "Searches curated public feeds (CISA advisories, etc.) for mentions\n"
+            "of the given CVE ID and returns matching snippets and feed metadata."
+        ),
+        add_help=True,
+    )
+    p.add_argument("cve_id", metavar="CVE-ID", help="CVE identifier, e.g. CVE-2024-3094")
+    p.add_argument(
+        "--output",
+        choices=["text", "json"],
+        default="text",
+        help="Output format (default: text)",
+    )
+    return p
+
+
+def _run_threat_feeds(argv: list[str]) -> int:
+    parser = _build_threat_feeds_parser()
+    args = parser.parse_args(argv)
+    cve_id = args.cve_id.strip()
+
+    if not re.match(r"CVE-\d{4}-\d+", cve_id, re.IGNORECASE):
+        parser.error(f"Invalid CVE ID: {cve_id!r}. Expected format: CVE-YYYY-NNNNN")
+
+    try:
+        from manus_agent.tools.query_threat_intelligence_feeds import (
+            fetch_threat_intelligence,
+        )
+    except ImportError as exc:  # pragma: no cover
+        print(f"Error: failed to import threat intelligence module: {exc}", file=sys.stderr)
+        return 1
+
+    payload = fetch_threat_intelligence(cve_id)
+
+    if args.output == "json":
+        print(json.dumps(payload, indent=2))
+        return 0
+
+    # --- text output ---
+    print()
+    print(f"Threat Intelligence Feeds — {cve_id}")
+    print("=" * 60)
+    print(payload["summary"])
+
+    if payload["intelligence"]:
+        print()
+        for i, entry in enumerate(payload["intelligence"], 1):
+            print(f"  [{i}] {entry['feed_name']}")
+            print(f"      URL: {entry['feed_url']}")
+            snippet = entry.get("snippet", "").strip()
+            if snippet:
+                # Truncate long snippets for terminal readability
+                display_snippet = snippet[:200] + "..." if len(snippet) > 200 else snippet
+                print(f"      Snippet: {display_snippet}")
+            print()
+
+    if payload["errors"]:
+        print("Errors encountered:")
+        for err in payload["errors"]:
+            print(f"  ⚠  {err['feed_name']}: {err['error']}")
+        print()
+
+    if not payload["intelligence"]:
+        print()
+        print("No mentions found in the curated threat intelligence feeds.")
+        print("This does not mean the CVE is not being exploited — it may")
+        print("simply not appear in the feeds currently monitored.")
+
+    return 0
+
+
 def _build_run_parser() -> argparse.ArgumentParser:
     """Build the top-level run/interactive parser."""
     parser = argparse.ArgumentParser(
@@ -2269,6 +2350,10 @@ def main() -> None:
         idx = argv.index("blast-radius")
         sys.exit(_run_blast_radius(argv[idx + 1 :]))
 
+    if first_positional == "threat-feeds":
+        idx = argv.index("threat-feeds")
+        sys.exit(_run_threat_feeds(argv[idx + 1 :]))
+
     if first_positional == "discover":
         idx = argv.index("discover")
         discover_args = _build_discover_parser().parse_args(argv[idx + 1 :])

src/manus_agent/tools/query_threat_intelligence_feeds.py

@@ -32,71 +32,96 @@
     },
 }
 
+# Default curated list of public threat intelligence feeds.
+DEFAULT_THREAT_FEEDS: list[dict[str, str]] = [
+    {
+        "name": "CISA Cybersecurity Advisories",
+        "url": "https://www.cisa.gov/cybersecurity-advisories/all.xml",
+        "type": "rss",
+    },
+]
 
-def query_threat_intelligence_feeds(tool: ToolUse, **kwargs: Any) -> ToolResult:
-    tool_use_id = tool["toolUseId"]
-    tool_input = tool["input"]
-    cve_id = tool_input.get("cve_id")
 
-    if not isinstance(cve_id, str) or not cve_id.strip():
-        result = {
-            "toolUseId": tool_use_id,
-            "status": "error",
-            "content": [{"text": "Invalid CVE ID. Must be a non-empty string."}],
-        }
-        log_tool_output_size("query_threat_intelligence_feeds", result)
-        return result
+def fetch_threat_intelligence(
+    cve_id: str,
+    *,
+    feeds: list[dict[str, str]] | None = None,
+    timeout: int = 10,
+) -> dict[str, Any]:
+    """Query threat intelligence feeds for a CVE.
 
-    # Curated list of public threat intelligence feeds (example URLs)
-    # In a real-world scenario, this list would be more extensive and potentially configurable.
-    # Parsing logic would also need to be more robust for different feed formats (RSS, JSON, HTML).
-    threat_feeds = [
-        {
-            "name": "CISA Cybersecurity Advisories",
-            "url": "https://www.cisa.gov/cybersecurity-advisories/all.xml",  # Updated RSS feed
-            "type": "rss",
-        },
-        # Removed US-CERT Alerts as it was causing 404 errors and may be deprecated.
-    ]
+    Returns a dict with keys:
+      - summary (str): human-readable summary
+      - intelligence (list[dict]): list of feed match dicts
+      - errors (list[dict]): list of feed errors (feed_name, error)
+    """
+    if feeds is None:
+        feeds = DEFAULT_THREAT_FEEDS
 
     found_intelligence: list[dict[str, Any]] = []
+    errors: list[dict[str, str]] = []
 
-    for feed in threat_feeds:
+    for feed in feeds:
         try:
-            response = requests.get(feed["url"], timeout=10)
+            response = requests.get(feed["url"], timeout=timeout)
             response.raise_for_status()
             content = response.text
 
             # Basic search for CVE ID in the content
             if cve_id.upper() in content.upper():
-                # In a real tool, you'd parse the RSS/JSON/HTML more intelligently
-                # to extract relevant snippets, titles, and links.
+                idx = content.upper().find(cve_id.upper())
+                snippet_start = max(0, idx - 50)
+                snippet_end = min(len(content), idx + 100)
                 found_intelligence.append(
                     {
                         "feed_name": feed["name"],
                         "feed_url": feed["url"],
                         "cve_found": cve_id,
-                        "snippet": content[
-                            content.upper().find(cve_id.upper()) - 50 : content.upper().find(cve_id.upper()) + 100
-                        ]
-                        + "...",  # Basic snippet
+                        "snippet": content[snippet_start:snippet_end] + "...",
                     }
                 )
 
         except requests.exceptions.RequestException as e:
-            # Log the error but continue with other feeds
-            print(f"Error fetching {feed['name']} ({feed['url']}): {e}")
+            errors.append({"feed_name": feed["name"], "error": str(e)})
         except Exception as e:
-            print(f"An unexpected error occurred with {feed['name']}: {e}")
+            errors.append({"feed_name": feed["name"], "error": str(e)})
 
     if not found_intelligence:
+        summary = f"No direct threat intelligence found for {cve_id} in curated feeds."
+    else:
+        summary = f"Found relevant threat intelligence for {cve_id} in {len(found_intelligence)} feed(s)."
+
+    return {
+        "summary": summary,
+        "intelligence": found_intelligence,
+        "errors": errors,
+    }
+
+
+def query_threat_intelligence_feeds(tool: ToolUse, **kwargs: Any) -> ToolResult:
+    tool_use_id = tool["toolUseId"]
+    tool_input = tool["input"]
+    cve_id = tool_input.get("cve_id")
+
+    if not isinstance(cve_id, str) or not cve_id.strip():
+        result = {
+            "toolUseId": tool_use_id,
+            "status": "error",
+            "content": [{"text": "Invalid CVE ID. Must be a non-empty string."}],
+        }
+        log_tool_output_size("query_threat_intelligence_feeds", result)
+        return result
+
+    payload = fetch_threat_intelligence(cve_id)
+
+    if not payload["intelligence"]:
         result = {
             "toolUseId": tool_use_id,
             "status": "success",
             "content": [
                 {
                     "json": {
-                        "summary": f"No direct threat intelligence found for {cve_id} in curated feeds.",
+                        "summary": payload["summary"],
                         "intelligence": [],
                     }
                 }
@@ -105,11 +130,10 @@ def query_threat_intelligence_feeds(tool: ToolUse, **kwargs: Any) -> ToolResult:
         log_tool_output_size("query_threat_intelligence_feeds", result)
         return result
 
-    summary = f"Found relevant threat intelligence for {cve_id} in {len(found_intelligence)} feeds."
     result = {
         "toolUseId": tool_use_id,
         "status": "success",
-        "content": [{"json": {"summary": summary, "intelligence": found_intelligence}}],
+        "content": [{"json": {"summary": payload["summary"], "intelligence": payload["intelligence"]}}],
     }
     log_tool_output_size("query_threat_intelligence_feeds", result)
     return result