feat: v1.3.0 — Censys, ZoomEye, FOFA, GreyNoise OSINT integration

4 new internet intelligence APIs, all following TDD/SDD methodology:

Censys (search.censys.io/api/v2):
- censys_lookup(): host services, labels, last_updated
- censys_search(): query → list of IPs for bulk scanning
- Auth: HTTP Basic (ARGUS_CENSYS_ID + ARGUS_CENSYS_SECRET)

ZoomEye (api.zoomeye.org):
- zoomeye_lookup(): host info, port/service/geo data
- zoomeye_search(): dork → IPs
- Auth: JWT token header (ARGUS_ZOOMEYE_KEY)

FOFA (fofa.info):
- fofa_lookup(): domain/IP → hosts with port/protocol/country/product
- fofa_search(): arbitrary query → IPs
- Auth: email + API key, query as base64 (ARGUS_FOFA_EMAIL + ARGUS_FOFA_KEY)

GreyNoise (api.greynoise.io/v3):
- greynoise_lookup(): IP reputation — noise/riot/classification/name
- Community endpoint free (no key), Enterprise with ARGUS_GREYNOISE_KEY
- Shows if IP is known scanner, botnet, or benign CDN/DNS

All 4 APIs run in orchestrator Group 0 (OSINT parallel) alongside Shodan/VT.
HTML report: new "OSINT Intelligence" section with GreyNoise badge,
Censys services, ZoomEye/FOFA tables.

argus bulk now supports:
  --censys "services.port:443 AND labels:cloud"
  --zoomeye "hostname:example.com"
  --fofa 'domain="example.com"'

Config: 6 new api_keys fields + env overrides
Tests: 33 new tests (8 Censys + 7 ZoomEye + 9 FOFA + 8 GreyNoise)
Total: 524 tests passing

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: LuxVTZ

src/argus_lite/cli.py

@@ -323,7 +323,10 @@ def run_template(ctx: click.Context, template_path: str, target: str | None) ->
 
 @main.command("bulk")
 @click.argument("sources", nargs=-1)
-@click.option("--shodan", "shodan_query", default=None, help="Shodan search query (requires ARGUS_SHODAN_KEY)")
+@click.option("--shodan", "shodan_query", default=None, help="Shodan query (requires ARGUS_SHODAN_KEY)")
+@click.option("--censys", "censys_query", default=None, help="Censys query (requires ARGUS_CENSYS_ID + ARGUS_CENSYS_SECRET)")
+@click.option("--zoomeye", "zoomeye_query", default=None, help="ZoomEye dork (requires ARGUS_ZOOMEYE_KEY)")
+@click.option("--fofa", "fofa_query", default=None, help="FOFA query (requires ARGUS_FOFA_EMAIL + ARGUS_FOFA_KEY)")
 @click.option("--preset", type=click.Choice(["bulk", "quick", "web", "full", "recon"]), default="bulk",
               help="Scan preset for each target")
 @click.option("--concurrency", type=int, default=5, help="Max parallel scans")
@@ -333,21 +336,27 @@ def run_template(ctx: click.Context, template_path: str, target: str | None) ->
 def bulk_scan(
     sources: tuple[str, ...],
     shodan_query: str | None,
+    censys_query: str | None,
+    zoomeye_query: str | None,
+    fofa_query: str | None,
     preset: str,
     concurrency: int,
     max_targets: int,
     output_format: str,
     no_confirm: bool,
 ) -> None:
-    """Bulk scan multiple targets: files, CIDRs, ASNs, or Shodan queries.
+    """Bulk scan multiple targets: files, CIDRs, ASNs, or OSINT queries.
 
     \b
     Examples:
       argus bulk targets.txt
       argus bulk 192.168.1.0/24
       argus bulk AS12345
       argus bulk targets.txt 10.0.1.0/24 --preset web
-      argus bulk --shodan "org:MyCompany" --concurrency 3
+      argus bulk --shodan "org:MyCompany"
+      argus bulk --censys "services.port:443 AND labels:cloud"
+      argus bulk --zoomeye "hostname:example.com"
+      argus bulk --fofa 'domain="example.com"'
 
     Generates individual reports per target + a combined summary.html.
     IMPORTANT: Only scan systems you have written permission to test.
@@ -371,25 +380,34 @@ def bulk_scan(
     config.bulk.max_concurrent = concurrency
     config.bulk.max_targets = max_targets
 
-    # Build source list
-    all_sources = list(sources)
-    if shodan_query:
-        # Shodan handled via expand_shodan() separately
-        all_sources_for_expand = list(sources)
-    else:
-        all_sources_for_expand = all_sources
+    has_query = any([shodan_query, censys_query, zoomeye_query, fofa_query])
 
-    if not all_sources_for_expand and not shodan_query:
+    if not sources and not has_query:
         console.print("[red]No sources provided. Use positional args or --shodan.[/red]")
         raise SystemExit(1)
 
     # Expand targets
     expander = TargetExpander(config)
-    if shodan_query:
-        targets = asyncio.get_event_loop().run_until_complete(expander.expand_shodan(shodan_query))
+    targets: list[str] = []
+
+    # Expand positional sources (file, CIDR, ASN, plain hosts)
+    if sources:
         targets += asyncio.get_event_loop().run_until_complete(expander.expand(list(sources)))
-    else:
-        targets = asyncio.get_event_loop().run_until_complete(expander.expand(all_sources_for_expand))
+
+    # Expand OSINT queries
+    loop = asyncio.get_event_loop()
+    if shodan_query:
+        console.print(f"[dim]Querying Shodan: {shodan_query}[/dim]")
+        targets += loop.run_until_complete(expander.expand_shodan(shodan_query))
+    if censys_query:
+        console.print(f"[dim]Querying Censys: {censys_query}[/dim]")
+        targets += loop.run_until_complete(expander.expand_censys(censys_query))
+    if zoomeye_query:
+        console.print(f"[dim]Querying ZoomEye: {zoomeye_query}[/dim]")
+        targets += loop.run_until_complete(expander.expand_zoomeye(zoomeye_query))
+    if fofa_query:
+        console.print(f"[dim]Querying FOFA: {fofa_query}[/dim]")
+        targets += loop.run_until_complete(expander.expand_fofa(fofa_query))
 
     # Deduplicate & cap
     seen: set[str] = set()

src/argus_lite/core/config.py

@@ -79,6 +79,12 @@ class ApiKeysConfig(BaseModel):
     shodan: str = ""
     virustotal: str = ""
     nvd_api_key: str = ""
+    censys_api_id: str = ""
+    censys_api_secret: str = ""
+    zoomeye_api_key: str = ""
+    fofa_email: str = ""
+    fofa_api_key: str = ""
+    greynoise_api_key: str = ""
 
 
 class NotificationConfig(BaseModel):
@@ -163,6 +169,18 @@ def _apply_env_overrides(config: AppConfig) -> None:
     if nvd_key:
         config.api_keys.nvd_api_key = nvd_key
 
+    for env_name, attr in [
+        ("ARGUS_CENSYS_ID", "censys_api_id"),
+        ("ARGUS_CENSYS_SECRET", "censys_api_secret"),
+        ("ARGUS_ZOOMEYE_KEY", "zoomeye_api_key"),
+        ("ARGUS_FOFA_EMAIL", "fofa_email"),
+        ("ARGUS_FOFA_KEY", "fofa_api_key"),
+        ("ARGUS_GREYNOISE_KEY", "greynoise_api_key"),
+    ]:
+        val = os.environ.get(env_name)
+        if val:
+            setattr(config.api_keys, attr, val)
+
     for env_name, attr in [
         ("ARGUS_TELEGRAM_TOKEN", "telegram_token"),
         ("ARGUS_TELEGRAM_CHAT_ID", "telegram_chat_id"),

src/argus_lite/core/orchestrator.py

@@ -174,11 +174,14 @@ async def _run_subtask(self, name: str, coro) -> None:
             self._on_progress(name, "fail")
 
     async def _run_recon(self) -> None:
+        from argus_lite.modules.recon.censys_api import censys_lookup
         from argus_lite.modules.recon.certificates import certificate_info
         from argus_lite.modules.recon.dns import dns_enumerate
         from argus_lite.modules.recon.dnsx_resolve import parse_dnsx_output
+        from argus_lite.modules.recon.fofa_api import fofa_lookup
         from argus_lite.modules.recon.gau_urls import gau_discover
         from argus_lite.modules.recon.gowitness import gowitness_capture
+        from argus_lite.modules.recon.greynoise_api import greynoise_lookup
         from argus_lite.modules.recon.httpx_probe import httpx_probe, httpx_probe_multi
         from argus_lite.modules.recon.katana_crawl import katana_crawl
         from argus_lite.modules.recon.securitytrails_api import st_lookup
@@ -187,8 +190,9 @@ async def _run_recon(self) -> None:
         from argus_lite.modules.recon.tlsx_certs import tlsx_scan
         from argus_lite.modules.recon.virustotal_api import vt_lookup
         from argus_lite.modules.recon.whois import whois_lookup
+        from argus_lite.modules.recon.zoomeye_api import zoomeye_lookup
 
-        # Group 0: OSINT APIs (no tools needed, run in parallel, fire-and-forget)
+        # Group 0: OSINT APIs (no tools needed, run in parallel)
         api_tasks = []
         api_keys = self.config.api_keys
 
@@ -204,14 +208,40 @@ async def do_vt():
                 self._tools_used.append("virustotal-api")
             api_tasks.append(do_vt())
 
+        if api_keys.censys_api_id and api_keys.censys_api_secret:
+            async def do_censys():
+                self._recon_result.censys_info = await censys_lookup(
+                    self.target, api_id=api_keys.censys_api_id, api_secret=api_keys.censys_api_secret)
+                self._tools_used.append("censys-api")
+            api_tasks.append(do_censys())
+
+        if api_keys.zoomeye_api_key:
+            async def do_zoomeye():
+                self._recon_result.zoomeye_info = await zoomeye_lookup(
+                    self.target, api_key=api_keys.zoomeye_api_key)
+                self._tools_used.append("zoomeye-api")
+            api_tasks.append(do_zoomeye())
+
+        if api_keys.fofa_email and api_keys.fofa_api_key:
+            async def do_fofa():
+                self._recon_result.fofa_info = await fofa_lookup(
+                    self.target, email=api_keys.fofa_email, api_key=api_keys.fofa_api_key)
+                self._tools_used.append("fofa-api")
+            api_tasks.append(do_fofa())
+
+        # GreyNoise: enriches the main target IP (works even without API key via community endpoint)
+        async def do_greynoise():
+            self._recon_result.greynoise_info = await greynoise_lookup(
+                self.target, api_key=api_keys.greynoise_api_key)
+            self._tools_used.append("greynoise-api")
+        api_tasks.append(do_greynoise())
+
         async def do_st():
-            # SecurityTrails uses env var ARGUS_SECURITYTRAILS_KEY
             import os
             st_key = os.environ.get("ARGUS_SECURITYTRAILS_KEY", "")
             if st_key:
                 self._recon_result.securitytrails_info = await st_lookup(self.target, st_key)
                 self._tools_used.append("securitytrails-api")
-
         api_tasks.append(do_st())
 
         if api_tasks:

src/argus_lite/core/target_expander.py

@@ -67,6 +67,41 @@ async def expand_shodan(self, query: str) -> list[str]:
         """Expand a Shodan search query → list of IP addresses."""
         return await self._expand_shodan(query)
 
+    async def expand_censys(self, query: str) -> list[str]:
+        """Expand a Censys search query → list of IP addresses."""
+        from argus_lite.modules.recon.censys_api import censys_search
+        keys = self._config.api_keys
+        if not keys.censys_api_id or not keys.censys_api_secret:
+            logger.warning("Censys search requires ARGUS_CENSYS_ID + ARGUS_CENSYS_SECRET")
+            return []
+        results = await censys_search(query, api_id=keys.censys_api_id,
+                                      api_secret=keys.censys_api_secret,
+                                      max_results=self._max_targets)
+        return results[:self._max_targets]
+
+    async def expand_zoomeye(self, query: str) -> list[str]:
+        """Expand a ZoomEye dork → list of IP addresses."""
+        from argus_lite.modules.recon.zoomeye_api import zoomeye_search
+        api_key = self._config.api_keys.zoomeye_api_key
+        if not api_key:
+            logger.warning("ZoomEye search requires ARGUS_ZOOMEYE_KEY")
+            return []
+        results = await zoomeye_search(query, api_key=api_key,
+                                       max_results=self._max_targets)
+        return results[:self._max_targets]
+
+    async def expand_fofa(self, query: str) -> list[str]:
+        """Expand a FOFA query → list of IP addresses."""
+        from argus_lite.modules.recon.fofa_api import fofa_search
+        keys = self._config.api_keys
+        if not keys.fofa_email or not keys.fofa_api_key:
+            logger.warning("FOFA search requires ARGUS_FOFA_EMAIL + ARGUS_FOFA_KEY")
+            return []
+        results = await fofa_search(query, email=keys.fofa_email,
+                                    api_key=keys.fofa_api_key,
+                                    max_results=self._max_targets)
+        return results[:self._max_targets]
+
     # ------------------------------------------------------------------
     # Source type detection
     # ------------------------------------------------------------------

src/argus_lite/models/recon.py

@@ -116,6 +116,41 @@ class Screenshot(BaseModel):
     response_time_ms: int = 0
 
 
+class CensysServiceInfo(BaseModel):
+    port: int = 0
+    transport: str = ""
+    service_name: str = ""
+    banner: str = ""
+
+
+class CensysHostInfo(BaseModel):
+    ip: str = ""
+    services: list[CensysServiceInfo] = []
+    labels: list[str] = []
+    last_updated: str = ""
+    total_results: int = 0
+
+
+class ZoomEyeHostInfo(BaseModel):
+    total: int = 0
+    matches: list[dict] = []   # ip, portinfo.port, geoinfo.country/city
+
+
+class FofaHostInfo(BaseModel):
+    total: int = 0
+    results: list[dict] = []   # ip, port, protocol, country, city, product
+
+
+class GreyNoiseInfo(BaseModel):
+    ip: str = ""
+    noise: bool = False          # True = observed scanning the internet
+    riot: bool = False           # True = known benign service (CDN, DNS, etc.)
+    classification: str = ""     # benign | malicious | unknown
+    name: str = ""               # e.g. "Cloudflare", "Google Public DNS"
+    last_seen: str = ""
+    message: str = ""
+
+
 class ReconResult(BaseModel):
     """Aggregated result from recon module."""
 
@@ -132,3 +167,7 @@ class ReconResult(BaseModel):
     shodan_info: ShodanHostInfo | None = None
     virustotal_info: VirusTotalInfo | None = None
     securitytrails_info: SecurityTrailsInfo | None = None
+    censys_info: CensysHostInfo | None = None
+    zoomeye_info: ZoomEyeHostInfo | None = None
+    fofa_info: FofaHostInfo | None = None
+    greynoise_info: GreyNoiseInfo | None = None

src/argus_lite/modules/recon/censys_api.py

@@ -0,0 +1,89 @@
+"""Censys Search API v2 integration."""
+
+from __future__ import annotations
+
+import logging
+
+import httpx
+
+from argus_lite.models.recon import CensysHostInfo, CensysServiceInfo
+
+logger = logging.getLogger(__name__)
+
+_BASE_URL = "https://search.censys.io/api/v2"
+
+
+def parse_censys_host_response(data: dict) -> CensysHostInfo:
+    """Parse Censys /v2/hosts/{ip} response into CensysHostInfo."""
+    result = data.get("result", {})
+    if not result:
+        return CensysHostInfo()
+
+    services = []
+    for svc in result.get("services", []):
+        services.append(CensysServiceInfo(
+            port=svc.get("port", 0),
+            transport=svc.get("transport_protocol", ""),
+            service_name=svc.get("service_name", ""),
+            banner=(svc.get("banner", "") or "")[:200],
+        ))
+
+    return CensysHostInfo(
+        ip=result.get("ip", ""),
+        services=services,
+        labels=result.get("labels", []),
+        last_updated=result.get("last_updated_at", ""),
+        total_results=1,
+    )
+
+
+def parse_censys_search_response(data: dict) -> list[str]:
+    """Parse Censys /v2/hosts/search response → list of IPs."""
+    result = data.get("result", {})
+    hits = result.get("hits", [])
+    return [h.get("ip", "") for h in hits if h.get("ip")]
+
+
+async def censys_lookup(target: str, api_id: str, api_secret: str) -> CensysHostInfo:
+    """Look up a host on Censys by IP or domain. Returns CensysHostInfo."""
+    if not api_id or not api_secret:
+        return CensysHostInfo()
+
+    # Resolve domain to IP if needed
+    import socket
+    try:
+        ip = socket.gethostbyname(target)
+    except socket.gaierror:
+        ip = target
+
+    url = f"{_BASE_URL}/hosts/{ip}"
+    try:
+        async with httpx.AsyncClient(timeout=30) as client:
+            resp = await client.get(url, auth=(api_id, api_secret))
+        if resp.status_code != 200:
+            logger.debug("Censys returned %s for %s", resp.status_code, ip)
+            return CensysHostInfo()
+        return parse_censys_host_response(resp.json())
+    except Exception as exc:
+        logger.debug("Censys lookup failed for %s: %s", target, exc)
+        return CensysHostInfo()
+
+
+async def censys_search(query: str, api_id: str, api_secret: str,
+                        max_results: int = 100) -> list[str]:
+    """Search Censys with a query string → list of IP addresses."""
+    if not api_id or not api_secret:
+        return []
+
+    url = f"{_BASE_URL}/hosts/search"
+    params = {"q": query, "per_page": min(max_results, 100)}
+    try:
+        async with httpx.AsyncClient(timeout=30) as client:
+            resp = await client.get(url, params=params, auth=(api_id, api_secret))
+        if resp.status_code != 200:
+            logger.debug("Censys search returned %s", resp.status_code)
+            return []
+        return parse_censys_search_response(resp.json())
+    except Exception as exc:
+        logger.debug("Censys search failed: %s", exc)
+        return []