diff --git a/.github/actions/install/action.yml b/.github/actions/install/action.yml index 1f2f53685..0951f4a91 100644 --- a/.github/actions/install/action.yml +++ b/.github/actions/install/action.yml @@ -20,7 +20,7 @@ runs: - name: Install secator with pipx shell: bash - run: pipx install -e .[dev,ai] + run: pipx install -e .[dev,ai,shodan] - name: Add secator to $PATH shell: bash diff --git a/pyproject.toml b/pyproject.toml index b9c23c7cb..ed5ff78ac 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -94,6 +94,9 @@ ai = [ 'litellm < 2', 'safecmd' ] +shodan = [ + 'shodan < 2' +] [project.scripts] secator = 'secator.cli:cli' diff --git a/secator/config.py b/secator/config.py index bea53e099..e489a976c 100644 --- a/secator/config.py +++ b/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() diff --git a/secator/configs/workflows/domain_recon.yaml b/secator/configs/workflows/domain_recon.yaml index 2263c06e4..8379f4e7d 100644 --- a/secator/configs/workflows/domain_recon.yaml +++ b/secator/configs/workflows/domain_recon.yaml @@ -47,6 +47,11 @@ tasks: dnsx: description: Resolve DNS records + shodan: + description: Passive DNS records via Shodan + operation: dns + if: opts.passive + wafw00f: description: Check WAF targets_: diff --git a/secator/tasks/shodan.py b/secator/tasks/shodan.py new file mode 100644 index 000000000..ec3ddb45e --- /dev/null +++ b/secator/tasks/shodan.py @@ -0,0 +1,285 @@ +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, STRING +from secator.output_types import ( + Error, Ip, Port, Record, 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, STRING] + output_types = [Ip, Subdomain, Port, Technology, Vulnerability, Tag, Record] + tags = ['shodan', 'recon', 'osint', 'passive'] + install_cmd = 'pip install shodan' + opts = { + 'operation': {'type': str, 'default': 'host', 'short': 'op', 'help': 'Operation: host | dns | search'}, + # 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)'}, + 'resolver': {'type': str, 'default': 'local', 'help': 'host mode: hostname resolver — local | shodan'}, + 'record_types': {'type': list, 'default': ['A', 'AAAA', 'CNAME', 'MX', 'NS', 'TXT', 'SOA'], + 'help': 'dns mode: DNS record types to emit'}, + 'limit': {'type': int, 'default': 100, 'help': 'search mode: max results (one page = 100)'}, + } + + 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 + + # Validate operation first so unknown ops get a clean error regardless of key. + operation = self.get_opt_value('operation') or 'host' + if operation not in ('host', 'dns', 'search'): + yield Error(message=f"Unknown Shodan operation '{operation}' (expected host | dns | search).") + 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) + if operation == 'host': + yield from self._run_host(api, shodan_sdk) + elif operation == 'dns': + yield from self._run_dns(api, shodan_sdk) + elif operation == 'search': + yield from self._run_search(api, shodan_sdk) + + def _run_host(self, api, shodan_sdk): + 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 + if (self.get_opt_value('resolver') or 'local') == 'shodan': + try: + ip = self._shodan_resolve(api, target) + except shodan_sdk.APIError as e: + yield Error(message=f'Shodan DNS resolve failed for {target}: {e}') + continue + if not ip: + yield Error(message=f'Shodan DNS has no A record for {target}') + continue + else: + 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 _run_dns(self, api, shodan_sdk): + record_types = [str(t).upper() for t in (self.get_opt_value('record_types') or [])] + for domain in self.inputs: + try: + info = api.dns.domain_info(domain) + except shodan_sdk.APIError as e: + msg = str(e) + if 'No information' in msg or 'Invalid' in msg: + yield Warning(message=f'No Shodan DNS data for {domain}') + else: + yield Error(message=f'Shodan DNS error for {domain}: {msg}') + continue + yield from self._map_dns(domain, info, record_types) + + def _map_dns(self, domain, info, record_types): + for r in (info.get('data') or []): + rtype = str(r.get('type') or '').upper() + if record_types and rtype not in record_types: + continue + sub = r.get('subdomain') or '' + fqdn = f'{sub}.{domain}' if sub else domain + value = r.get('value') + yield Record( + name=fqdn, type=rtype, host=domain, + extra_data=self._compact({'value': value, 'last_seen': r.get('last_seen'), 'ports': r.get('ports')}), + tags=['shodan'], + ) + if rtype in ('A', 'AAAA') and value and self._is_public_ip(value): + yield Ip(ip=value, host=fqdn, alive=True, tags=['shodan']) + seen = set() + for sub in (info.get('subdomains') or []): + host = f'{sub}.{domain}' + if host not in seen: + seen.add(host) + yield Subdomain(host=host, domain=domain, sources=['shodan']) + + def _shodan_resolve(self, api, host): + """Resolve a hostname to an IP via Shodan DNS (no local resolver). Returns the + first matching A-record value, or None.""" + domain = self._registered_domain(host) + sub = host[:-len(domain)].rstrip('.') if host != domain else '' + info = api.dns.domain_info(domain) + for r in (info.get('data') or []): + if str(r.get('type')) == 'A' and (r.get('subdomain') or '') == sub: + return r.get('value') + return None + + def _run_search(self, api, shodan_sdk): + query = ' '.join(self.inputs).strip() + if not query: + yield Error(message='Shodan search requires a query (pass it as the input).') + return + limit = self.get_opt_value('limit') or 100 + try: + result = api.search(query, limit=limit) + except shodan_sdk.APIError as e: + yield Error(message=f'Shodan search error: {e}') + return + yield Tag(name='shodan_search_total', value=str(result.get('total', 0)), + match=query, category='info', tags=['shodan']) + for match in (result.get('matches') or []): + ip_str = match.get('ip_str') + if not ip_str: + continue + host = (match.get('hostnames') or [''])[0] + yield Ip( + ip=ip_str, host=host, alive=True, + extra_data=self._compact({'os': match.get('os'), 'org': match.get('org'), + 'isp': match.get('isp'), 'asn': match.get('asn')}), + tags=['shodan'], + ) + seen = set() + for name in (match.get('hostnames') or []): + if name and name not in seen: + seen.add(name) + yield Subdomain(host=name, domain=self._registered_domain(name), sources=['shodan']) + yield from self._map_banner(match, ip_str, 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 []): + yield from self._map_banner(b, ip_str, host) + + def _map_banner(self, b, ip_str, host): + port = b.get('port') + try: + port = int(port) + except (TypeError, ValueError): + return + 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 _is_public_ip(value): + try: + return ipaddress.ip_address(value).is_global + 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() + + @staticmethod + def validate_input(self, inputs): + # In search mode the input is a free-text Shodan query (e.g. "apache country:US"), + # not a HOST/IP — accept it. host/dns inputs are still typed via input_types. + return True diff --git a/secator/utils_test.py b/secator/utils_test.py index d0afd95ff..aa06e3dc7 100644 --- a/secator/utils_test.py +++ b/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 + }, } diff --git a/tests/fixtures/shodan_dns_output.json b/tests/fixtures/shodan_dns_output.json new file mode 100644 index 000000000..af96ceb1a --- /dev/null +++ b/tests/fixtures/shodan_dns_output.json @@ -0,0 +1,13 @@ +{ + "domain": "example.com", + "tags": ["ipv6"], + "subdomains": ["www", "mail"], + "data": [ + {"subdomain": "", "type": "A", "value": "93.184.216.34", "last_seen": "2026-06-01T00:00:00"}, + {"subdomain": "www", "type": "A", "value": "93.184.216.34", "last_seen": "2026-06-01T00:00:00"}, + {"subdomain": "", "type": "AAAA", "value": "2606:2800:220:1:248:1893:25c8:1946", "last_seen": "2026-06-01T00:00:00"}, + {"subdomain": "mail", "type": "MX", "value": "mail.example.com", "last_seen": "2026-06-01T00:00:00"}, + {"subdomain": "", "type": "TXT", "value": "v=spf1 -all", "last_seen": "2026-06-01T00:00:00"}, + {"subdomain": "internal", "type": "A", "value": "10.0.0.5", "last_seen": "2026-06-01T00:00:00"} + ] +} diff --git a/tests/fixtures/shodan_output.json b/tests/fixtures/shodan_output.json new file mode 100644 index 000000000..1aa937e9a --- /dev/null +++ b/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" + } + ] +} diff --git a/tests/fixtures/shodan_search_output.json b/tests/fixtures/shodan_search_output.json new file mode 100644 index 000000000..03e295b1f --- /dev/null +++ b/tests/fixtures/shodan_search_output.json @@ -0,0 +1,17 @@ +{ + "total": 2, + "matches": [ + { + "ip_str": "10.0.0.1", "org": "Example Org", "isp": "Example ISP", "asn": "AS65000", + "hostnames": ["a.example.com"], + "port": 80, "transport": "tcp", "product": "nginx", "version": "1.21.0", + "cpe": ["cpe:/a:nginx:nginx:1.21.0"], "data": "HTTP/1.1 200 OK\r\n", + "vulns": {"CVE-2021-23017": {"cvss": "9.8", "summary": "nginx resolver off-by-one"}} + }, + { + "ip_str": "10.0.0.2", "org": "Example Org", "hostnames": ["b.example.com"], + "port": 22, "transport": "tcp", "product": "OpenSSH", "version": "8.2p1", + "data": "SSH-2.0-OpenSSH_8.2p1\r\n" + } + ] +} diff --git a/tests/integration/test_tasks.py b/tests/integration/test_tasks.py index 7c60e8b59..fef63b815 100644 --- a/tests/integration/test_tasks.py +++ b/tests/integration/test_tasks.py @@ -80,7 +80,7 @@ def test_tasks(self): TASKS = [t for t in tasks if t.__name__ in test_tasks_names] for cls in TASKS: - if cls.__name__ == 'msfconsole': # skip msfconsole test as it's stuck + if cls.__name__ in ('msfconsole', 'shodan'): # msfconsole is stuck; shodan needs a live SHODAN_API_KEY not available in CI continue with self.subTest(name=cls.__name__): input = INPUTS_TASKS.get(cls.__name__) diff --git a/tests/unit/test_shodan.py b/tests/unit/test_shodan.py new file mode 100644 index 000000000..a26ded742 --- /dev/null +++ b/tests/unit/test_shodan.py @@ -0,0 +1,209 @@ +import json +import os +import unittest +import unittest.mock + + +def _load_dns_fixture(): + path = os.path.join(os.path.dirname(__file__), '..', 'fixtures', 'shodan_dns_output.json') + with open(path) as f: + return json.load(f) + + +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) + + +class TestShodanDns(unittest.TestCase): + def _run(self, record_types=None): + from secator.tasks.shodan import shodan + task = shodan.__new__(shodan) + types = record_types or ['A', 'AAAA', 'CNAME', 'MX', 'NS', 'TXT', 'SOA'] + return list(task._map_dns('example.com', _load_dns_fixture(), types)) + + def test_emits_record_per_type(self): + from secator.output_types import Record + recs = [r for r in self._run() if isinstance(r, Record)] + types = sorted({r.type for r in recs}) + self.assertEqual(types, ['A', 'AAAA', 'MX', 'TXT']) + a = next(r for r in recs if r.type == 'A' and r.name == 'www.example.com') + self.assertEqual(a.host, 'example.com') + self.assertEqual(a.extra_data.get('value'), '93.184.216.34') + + def test_ip_only_for_public_a_aaaa(self): + from secator.output_types import Ip + ips = sorted({i.ip for i in self._run() if isinstance(i, Ip)}) + # public A + AAAA only; the 10.0.0.5 private A is excluded + self.assertEqual(ips, ['2606:2800:220:1:248:1893:25c8:1946', '93.184.216.34']) + + def test_emits_subdomains_from_list(self): + from secator.output_types import Subdomain + subs = sorted({s.host for s in self._run() if isinstance(s, Subdomain)}) + self.assertEqual(subs, ['mail.example.com', 'www.example.com']) + + def test_record_types_filter(self): + from secator.output_types import Record + recs = [r for r in self._run(record_types=['MX']) if isinstance(r, Record)] + self.assertEqual({r.type for r in recs}, {'MX'}) + + +def _load_search_fixture(): + path = os.path.join(os.path.dirname(__file__), '..', 'fixtures', 'shodan_search_output.json') + with open(path) as f: + return json.load(f) + + +class TestShodanSearch(unittest.TestCase): + def _run(self): + import unittest.mock as m + from secator.tasks.shodan import shodan + task = shodan.__new__(shodan) + task.inputs = ['apache country:US'] + task.run_opts = {'limit': 100} + mock_api = m.MagicMock() + mock_api.search.return_value = _load_search_fixture() + import shodan as sdk + with m.patch('shodan.Shodan', return_value=mock_api): + return list(task._run_search(mock_api, sdk)) + + def test_emits_total_tag(self): + from secator.output_types import Tag + tags = [t for t in self._run() if isinstance(t, Tag) and t.name == 'shodan_search_total'] + self.assertEqual(len(tags), 1) + self.assertEqual(tags[0].value, '2') + + def test_emits_ip_per_match(self): + from secator.output_types import Ip + ips = sorted(i.ip for i in self._run() if isinstance(i, Ip)) + self.assertEqual(ips, ['10.0.0.1', '10.0.0.2']) + + def test_emits_ports_low_confidence(self): + from secator.output_types import Port + ports = [p for p in self._run() if isinstance(p, Port)] + self.assertEqual(sorted(p.port for p in ports), [22, 80]) + for p in ports: + self.assertEqual(p.confidence, 'low') + + def test_emits_banner_vuln_low_confidence(self): + from secator.output_types import Vulnerability + vulns = [v for v in self._run() if isinstance(v, Vulnerability)] + self.assertTrue(any(v.name == 'CVE-2021-23017' and v.confidence == 'low' + and v.cvss_score == 9.8 for v in vulns))