feat: Shodan host-lookup task (passive recon via the Shodan SDK)

pyproject.toml

@@ -94,6 +94,9 @@ ai = [
 	'litellm < 2',
 	'safecmd'
 ]
+shodan = [
+	'shodan < 2'
+]
 
 [project.scripts]
 secator = 'secator.cli:cli'

secator/config.py

@@ -225,6 +225,11 @@ class VulnersAddon(StrictModel):
 	api_key: str = ''
 
 
+class ShodanAddon(StrictModel):
+	enabled: bool = False
+	api_key: str = ''
+
+
 class AiAddon(StrictModel):
 	enabled: bool = False
 	api_key: str = ''
@@ -315,6 +320,7 @@ class Addons(StrictModel):
 	mongodb: MongodbAddon = MongodbAddon()
 	sqlite: SqliteAddon = SqliteAddon()
 	vulners: VulnersAddon = VulnersAddon()
+	shodan: ShodanAddon = ShodanAddon()
 	discord: DiscordAddon = DiscordAddon()
 	api: ApiAddon = ApiAddon()
 	ai: AiAddon = AiAddon()

secator/tasks/shodan.py

@@ -0,0 +1,163 @@
+import contextlib
+import ipaddress
+import os
+import socket
+import unittest.mock
+
+from secator.config import CONFIG
+from secator.decorators import task
+from secator.definitions import HOST, IP
+from secator.output_types import (
+	Error, Ip, Port, Subdomain, Tag, Technology, Vulnerability, Warning
+)
+from secator.runners import PythonRunner
+
+
+@task()
+class shodan(PythonRunner):
+	"""Passive host recon via the Shodan API (ports, services, CVEs, hostnames)."""
+	input_types = [HOST, IP]
+	output_types = [Ip, Subdomain, Port, Technology, Vulnerability, Tag]
+	tags = ['shodan', 'recon', 'osint', 'passive']
+	install_cmd = 'pip install shodan'
+	opts = {
+		# Empty default + runtime fallback (never a CONFIG default — it would leak
+		# the configured key into the secator-api UI form, like the `ai` task).
+		'api_key': {'type': str, 'default': '', 'help': 'Shodan API key (defaults to configured key)'},
+		'history': {'is_flag': True, 'default': False, 'help': 'Include historical (non-current) banners'},
+		'minify': {'is_flag': True, 'default': False, 'help': 'Only ports + general host info (no banners)'},
+	}
+
+	def yielder(self):
+		try:
+			import shodan as shodan_sdk
+		except ImportError:
+			yield Error(message="The 'shodan' package is not installed. Run: pip install shodan")
+			return
+
+		api_key = (
+			self.get_opt_value('api_key')
+			or CONFIG.addons.shodan.api_key
+			or os.environ.get('SHODAN_API_KEY', '')
+		)
+		if not api_key:
+			yield Error(message='Shodan API key not configured (set the api_key opt, '
+								'CONFIG.addons.shodan.api_key, or the SHODAN_API_KEY env var).')
+			return
+
+		api = shodan_sdk.Shodan(api_key)
+		history = self.get_opt_value('history')
+		minify = self.get_opt_value('minify')
+
+		for target in self.inputs:
+			ip, hostname = target, ''
+			if not self._is_ip(target):
+				hostname = target
+				try:
+					ip = socket.gethostbyname(target)
+				except (socket.gaierror, OSError) as e:
+					yield Error(message=f'Could not resolve {target}: {e}')
+					continue
+			try:
+				data = api.host(ip, history=history, minify=minify)
+			except shodan_sdk.APIError as e:
+				msg = str(e)
+				if 'No information available' in msg:
+					yield Warning(message=f'No Shodan data for {ip}')
+				else:
+					yield Error(message=f'Shodan API error for {ip}: {msg}')
+				continue
+			host = hostname or (data.get('hostnames') or [''])[0]
+			yield from self._map_host(data, ip, host)
+
+	def _map_host(self, h, ip, host):
+		ip_str = h.get('ip_str', ip)
+		yield Ip(
+			ip=ip_str, host=host, alive=True,
+			extra_data=self._compact({
+				'os': h.get('os'), 'org': h.get('org'), 'isp': h.get('isp'),
+				'asn': h.get('asn'), 'country': h.get('country_name'),
+			}),
+			tags=['shodan'],
+		)
+		seen = set()
+		for name in (h.get('hostnames') or []) + (h.get('domains') or []):
+			if name and name not in seen:
+				seen.add(name)
+				yield Subdomain(host=name, domain=self._registered_domain(name), sources=['shodan'])
+		for key, label in (('org', 'shodan_org'), ('isp', 'shodan_isp'),
+						   ('asn', 'shodan_asn'), ('os', 'shodan_os')):
+			val = h.get(key)
+			if val:
+				yield Tag(name=label, value=str(val), match=ip_str, category='info', tags=['shodan'])
+		for cve in (h.get('vulns') or []):
+			yield Vulnerability(name=cve, id=cve, matched_at=ip_str, ip=ip_str,
+								provider='shodan', confidence='low', tags=['shodan'])
+		for b in (h.get('data') or []):
+			port = b.get('port')
+			try:
+				port = int(port)
+			except (TypeError, ValueError):
+				continue
+			yield Port(
+				port=port, ip=ip_str, host=host, state='open',
+				protocol=b.get('transport', 'tcp'),
+				service_name=b.get('product', '') or '',
+				cpes=b.get('cpe', []) or [],
+				confidence='low', service_confidence='low',
+				extra_data=self._compact({'version': b.get('version'), 'banner': self._excerpt(b.get('data'))}),
+				tags=['shodan'],
+			)
+			product = b.get('product')
+			if product:
+				yield Technology(
+					product=product, match=f'{ip_str}:{port}', version=b.get('version'),
+					extra_data=self._compact({'cpe': b.get('cpe')}), tags=['shodan'],
+				)
+			for cve, meta in (b.get('vulns') or {}).items():
+				cvss = 0.0
+				if isinstance(meta, dict) and meta.get('cvss') is not None:
+					try:
+						cvss = float(meta.get('cvss'))
+					except (TypeError, ValueError):
+						cvss = 0.0
+				yield Vulnerability(
+					name=cve, id=cve, matched_at=f'{ip_str}:{port}', ip=ip_str,
+					provider='shodan', confidence='low', cvss_score=cvss,
+					description=(meta.get('summary', '') if isinstance(meta, dict) else ''),
+					tags=['shodan'],
+				)
+
+	@staticmethod
+	def _is_ip(value):
+		try:
+			ipaddress.ip_address(value)
+			return True
+		except ValueError:
+			return False
+
+	@staticmethod
+	def _registered_domain(hostname):
+		parts = hostname.split('.')
+		return '.'.join(parts[-2:]) if len(parts) >= 2 else hostname
+
+	@staticmethod
+	def _excerpt(text, length=500):
+		return (text or '')[:length]
+
+	@staticmethod
+	def _compact(d):
+		return {k: v for k, v in d.items() if v not in (None, '', [], {})}
+
+	@classmethod
+	def get_mock_context(cls, fixture):
+		"""Mock the Shodan SDK + DNS for the PythonRunner unit-test harness (no network)."""
+		@contextlib.contextmanager
+		def _ctx():
+			mock_api = unittest.mock.MagicMock()
+			mock_api.host.return_value = fixture
+			patch_shodan = unittest.mock.patch('shodan.Shodan', return_value=mock_api)
+			patch_dns = unittest.mock.patch('socket.gethostbyname', return_value='10.0.0.1')
+			with patch_shodan, patch_dns:
+				yield
+		return _ctx()

secator/utils_test.py

@@ -76,6 +76,7 @@
 	'maigret': 'Linus__Torvalds',
 	'searchsploit': 'apache',
 	'search_vulns': 'apache 2.4.39',
+	'shodan': '10.0.0.1',
 }
 
 #---------------------#
@@ -137,6 +138,9 @@
 		'sensitive': False,
 		'prompt': 'Run a full reconnaissance on this target',
 	},
+	'shodan': {
+		'api_key': 'test-key',  # bypassed by get_mock_context; just satisfies the key check
+	},
 }
 
 

tests/fixtures/shodan_output.json

@@ -0,0 +1,33 @@
+{
+  "ip_str": "10.0.0.1",
+  "os": "Linux 3.x",
+  "org": "Example Org",
+  "isp": "Example ISP",
+  "asn": "AS65000",
+  "country_name": "Wonderland",
+  "hostnames": ["host1.example.com", "www.example.com"],
+  "domains": ["example.com"],
+  "vulns": ["CVE-2021-40438"],
+  "ports": [80, 443],
+  "data": [
+    {
+      "port": 80,
+      "transport": "tcp",
+      "product": "Apache httpd",
+      "version": "2.4.49",
+      "cpe": ["cpe:/a:apache:http_server:2.4.49"],
+      "data": "HTTP/1.1 200 OK\r\nServer: Apache/2.4.49\r\n",
+      "vulns": {
+        "CVE-2021-41773": {"cvss": "7.5", "summary": "Path traversal in Apache 2.4.49"}
+      }
+    },
+    {
+      "port": 443,
+      "transport": "tcp",
+      "product": "Apache httpd",
+      "version": "2.4.49",
+      "cpe": ["cpe:/a:apache:http_server:2.4.49"],
+      "data": "HTTP/1.1 200 OK\r\n"
+    }
+  ]
+}

tests/integration/inputs.py

@@ -23,6 +23,7 @@
 	'jswhois': 'wikipedia.org',
 	'nuclei': 'http://localhost:3000/',
 	'searchsploit': 'apache 2.4.5',
+	'shodan': '8.8.8.8',
 	'subfinder': 'github.com',
 	'search_vulns': 'apache 2.4.39',
 	'testssl': 'free.fr',

tests/integration/outputs.py

@@ -157,6 +157,9 @@
 	'searchsploit': [
 		Exploit(name='cgi-bin Remote Code Execution', provider='EDB', id='29290', matched_at='apache 2.4.5', confidence='low'),
 	],
+	'shodan': [
+		Ip(ip='8.8.8.8', alive=True, _source='shodan'),
+	],
 	'search_vulns': [
 		Exploit(name='Apache exploit', provider='apache', id='CVE-2019-10081-exploit', matched_at='apache 2.4.39', confidence='high'),
 	],

tests/unit/test_shodan.py

@@ -0,0 +1,126 @@
+import json
+import os
+import unittest
+import unittest.mock
+
+
+class TestShodanConfig(unittest.TestCase):
+    def test_addon_defaults(self):
+        from secator.config import CONFIG
+        self.assertFalse(CONFIG.addons.shodan.enabled)
+        self.assertEqual(CONFIG.addons.shodan.api_key, '')
+
+
+def _load_fixture():
+    path = os.path.join(os.path.dirname(__file__), '..', 'fixtures', 'shodan_output.json')
+    with open(path) as f:
+        return json.load(f)
+
+
+class TestShodanMapping(unittest.TestCase):
+    def _run(self):
+        from secator.output_types import Ip, Subdomain, Port, Technology, Vulnerability, Tag
+        from secator.tasks.shodan import shodan
+        task = shodan.__new__(shodan)
+        return list(task._map_host(_load_fixture(), '10.0.0.1', 'host1.example.com'))
+
+    def test_emits_ip(self):
+        from secator.output_types import Ip
+        ips = [r for r in self._run() if isinstance(r, Ip)]
+        self.assertEqual(len(ips), 1)
+        self.assertEqual(ips[0].ip, '10.0.0.1')
+        self.assertTrue(ips[0].alive)
+        self.assertEqual(ips[0].extra_data.get('org'), 'Example Org')
+
+    def test_emits_subdomains_deduped(self):
+        from secator.output_types import Subdomain
+        subs = [r for r in self._run() if isinstance(r, Subdomain)]
+        hosts = sorted(s.host for s in subs)
+        self.assertEqual(hosts, ['example.com', 'host1.example.com', 'www.example.com'])
+
+    def test_emits_one_port_per_banner(self):
+        from secator.output_types import Port
+        ports = sorted(p.port for p in self._run() if isinstance(p, Port))
+        self.assertEqual(ports, [80, 443])
+
+    def test_ports_are_low_confidence(self):
+        from secator.output_types import Port
+        for p in [r for r in self._run() if isinstance(r, Port)]:
+            self.assertEqual(p.confidence, 'low')
+
+    def test_emits_technology_with_match_hostport(self):
+        from secator.output_types import Technology
+        techs = [r for r in self._run() if isinstance(r, Technology)]
+        self.assertTrue(any(t.product == 'Apache httpd' and t.match == '10.0.0.1:80' for t in techs))
+
+    def test_emits_vulns_low_confidence(self):
+        from secator.output_types import Vulnerability
+        vulns = [r for r in self._run() if isinstance(r, Vulnerability)]
+        names = {v.name for v in vulns}
+        self.assertIn('CVE-2021-40438', names)   # top-level
+        self.assertIn('CVE-2021-41773', names)   # per-banner
+        for v in vulns:
+            self.assertEqual(v.confidence, 'low')
+        banner_vuln = next(v for v in vulns if v.name == 'CVE-2021-41773')
+        self.assertEqual(banner_vuln.cvss_score, 7.5)
+        self.assertEqual(banner_vuln.matched_at, '10.0.0.1:80')
+
+    def test_emits_tags_for_metadata(self):
+        from secator.output_types import Tag
+        tags = {t.name for t in self._run() if isinstance(t, Tag)}
+        self.assertEqual(tags, {'shodan_org', 'shodan_isp', 'shodan_asn', 'shodan_os'})
+
+
+class TestShodanErrorPaths(unittest.TestCase):
+    """Test yielder() error branches: missing key, no-data warning, generic API error."""
+
+    def _make_task(self, **run_opts):
+        from secator.tasks.shodan import shodan
+        task = shodan.__new__(shodan)
+        task.run_opts = run_opts
+        task.inputs = ['10.0.0.1']
+        return task
+
+    def test_missing_api_key_yields_single_error(self):
+        """No api_key opt, no config key, no env var → exactly one Error, no findings."""
+        from secator.output_types import Error
+        task = self._make_task(api_key='')
+        mock_cfg = unittest.mock.MagicMock()
+        mock_cfg.addons.shodan.api_key = ''
+        env_without_key = {k: v for k, v in os.environ.items() if k != 'SHODAN_API_KEY'}
+        with unittest.mock.patch('secator.tasks.shodan.CONFIG', mock_cfg), \
+             unittest.mock.patch.dict(os.environ, env_without_key, clear=True):
+            results = list(task.yielder())
+        errors = [r for r in results if isinstance(r, Error)]
+        self.assertEqual(len(results), 1)
+        self.assertEqual(len(errors), 1)
+        self.assertIn('API key', errors[0].message)
+
+    def test_no_information_available_yields_warning_not_error(self):
+        """shodan.APIError('No information available...') → one Warning, zero Errors."""
+        import shodan as shodan_sdk
+        from secator.output_types import Error, Warning
+        task = self._make_task(api_key='testkey')
+        mock_api = unittest.mock.MagicMock()
+        mock_api.host.side_effect = shodan_sdk.APIError('No information available for that IP.')
+        with unittest.mock.patch('shodan.Shodan', return_value=mock_api):
+            results = list(task.yielder())
+        warnings = [r for r in results if isinstance(r, Warning)]
+        errors = [r for r in results if isinstance(r, Error)]
+        self.assertEqual(len(warnings), 1)
+        self.assertEqual(len(errors), 0)
+
+    def test_generic_api_error_yields_error(self):
+        """Non-'No information' shodan.APIError → one Error containing the message."""
+        import shodan as shodan_sdk
+        from secator.output_types import Error, Warning
+        task = self._make_task(api_key='testkey')
+        mock_api = unittest.mock.MagicMock()
+        mock_api.host.side_effect = shodan_sdk.APIError('Invalid API key')
+        with unittest.mock.patch('shodan.Shodan', return_value=mock_api):
+            results = list(task.yielder())
+        errors = [r for r in results if isinstance(r, Error)]
+        warnings = [r for r in results if isinstance(r, Warning)]
+        self.assertEqual(len(errors), 1)
+        self.assertEqual(len(warnings), 0)
+        self.assertIn('Invalid API key', errors[0].message)