diff --git a/aws/aws/contracts_aws.py b/aws/aws/contracts_aws.py index e9a3db85..35e957d4 100644 --- a/aws/aws/contracts_aws.py +++ b/aws/aws/contracts_aws.py @@ -222,7 +222,7 @@ def build_contract(): ) output_iam_privesc_paths = ContractOutputElement( - type=ContractOutputType.Text, + type=ContractOutputType.Vulnerability, field="privesc_paths", isMultiple=True, isFindingCompatible=True, @@ -255,6 +255,27 @@ def build_contract(): labels=["aws", "ec2", "security_group"], ) + # Public IPv4 addresses parsed out of EC2 / VPC enumeration, exposed as + # finding-compatible IPv4 primitives so they can chain into follow-up + # network injects. Shared across the EC2 instance and VPC contracts. + output_public_ips = ContractOutputElement( + type=ContractOutputType.IPv4, + field="public_ips", + isMultiple=True, + isFindingCompatible=True, + labels=["aws", "ipv4", "public"], + ) + + # Open security-group ports parsed out of EC2 enumeration, exposed as + # finding-compatible Port primitives. + output_open_ports = ContractOutputElement( + type=ContractOutputType.Port, + field="open_ports", + isMultiple=True, + isFindingCompatible=True, + labels=["aws", "ec2", "port"], + ) + # Lambda Outputs output_lambda_functions = ContractOutputElement( type=ContractOutputType.Text, @@ -273,18 +294,20 @@ def build_contract(): labels=["aws", "rds", "database"], ) - # Secrets Manager Outputs + # Secrets Manager Outputs -> Credentials (discovered credential-store + # entries; the enumeration surfaces the identifier, not the plaintext). output_secrets = ContractOutputElement( - type=ContractOutputType.Text, + type=ContractOutputType.Credentials, field="secrets", isMultiple=True, isFindingCompatible=True, labels=["aws", "secretsmanager", "secret"], ) - # SSM Outputs + # SSM Outputs -> Credentials (SSM parameters frequently hold credential + # material; surfaced as credential-store references). output_ssm_parameters = ContractOutputElement( - type=ContractOutputType.Text, + type=ContractOutputType.Credentials, field="parameters", isMultiple=True, isFindingCompatible=True, @@ -531,7 +554,12 @@ def make_contract( EC2_ENUM_INSTANCES_CONTRACT, "AWS - EC2 Enumerate Instances", "AWS - Énumération des instances EC2", - [output_ec2_instances, output_ec2_security_groups], + [ + output_ec2_instances, + output_ec2_security_groups, + output_public_ips, + output_open_ports, + ], attack_patterns=["T1580"], ) @@ -539,7 +567,7 @@ def make_contract( EC2_ENUM_SECURITY_GROUPS_CONTRACT, "AWS - EC2 Enumerate Security Groups", "AWS - Énumération des groupes de sécurité EC2", - [output_ec2_security_groups], + [output_ec2_security_groups, output_open_ports], attack_patterns=["T1580"], ) @@ -607,7 +635,7 @@ def make_contract( VPC_ENUM_CONTRACT, "AWS - VPC Enumerate Networks", "AWS - Énumération des VPC", - [output_vpc_networks], + [output_vpc_networks, output_public_ips], attack_patterns=["T1580"], ) diff --git a/aws/aws/helpers/pacu_executor.py b/aws/aws/helpers/pacu_executor.py index 038f3035..f501bc5c 100644 --- a/aws/aws/helpers/pacu_executor.py +++ b/aws/aws/helpers/pacu_executor.py @@ -3,9 +3,11 @@ This module handles the execution of Pacu commands and parsing of results. """ +import ipaddress import json import os import platform +import re import subprocess from typing import Dict, List, Optional, Tuple @@ -755,8 +757,38 @@ def parse_results(self, results: Dict) -> Dict: elif "ec2__enum" in module_name: stdout = data.get("stdout", "") instances, security_groups = self._parse_ec2_data(stdout) + public_ips = self._extract_public_ipv4s(stdout) + open_ports = self._extract_open_ports(stdout) - message = f"Found {len(instances)} EC2 instances, {len(security_groups)} security groups" + message = ( + f"Found {len(instances)} EC2 instances, " + f"{len(security_groups)} security groups, " + f"{len(public_ips)} public IPs, {len(open_ports)} open ports" + ) + if self.logger: + self.logger.info(message) + + outputs: Dict = { + "instances": instances, + "security_groups": security_groups, + } + if public_ips: + outputs["public_ips"] = public_ips + if open_ports: + outputs["open_ports"] = open_ports + + return { + "success": True, + "message": message, + "outputs": outputs, + } + + # Parse Secrets Manager enumeration -> Credentials findings + elif "secrets" in module_name: + stdout = data.get("stdout", "") or data.get("output", "") + secrets = self._parse_secrets_credentials(stdout) + + message = f"Found {len(secrets)} secrets" if self.logger: self.logger.info(message) @@ -764,8 +796,28 @@ def parse_results(self, results: Dict) -> Dict: "success": True, "message": message, "outputs": { - "instances": instances, - "security_groups": security_groups, + "secrets": secrets, + }, + } + + # Parse SSM parameters enumeration -> Credentials findings + elif ( + "systemsmanager" in module_name + or "ssm" in module_name + or "parameter" in module_name + ): + stdout = data.get("stdout", "") or data.get("output", "") + parameters = self._parse_ssm_credentials(stdout) + + message = f"Found {len(parameters)} SSM parameters" + if self.logger: + self.logger.info(message) + + return { + "success": True, + "message": message, + "outputs": { + "parameters": parameters, }, } @@ -865,18 +917,63 @@ def _parse_iam_roles(self, stdout: str) -> List[str]: roles.append(role_name) return roles - def _parse_privesc_paths(self, stdout: str) -> List[str]: - """Parse privilege escalation paths from Pacu output""" - privesc_paths = [] + # Negative / failure phrases that also contain a privesc keyword but report + # the ABSENCE of a finding (e.g. "No potential privilege escalation methods + # worked."). Lines matching any of these must never be emitted as a + # Vulnerability finding. + _PRIVESC_NEGATIVE_MARKERS = ( + "no potential", + "no privilege", + "no privesc", + "no escalation", + "no exploit", + "no methods", + "no method", + "no paths", + "no path", + "not vulnerable", + "none found", + "not found", + "did not", + "does not", + "could not", + "unable to", + ) + + def _parse_privesc_paths(self, stdout: str) -> List[Dict]: + """Parse privilege escalation paths from Pacu output into Vulnerability + findings. + + Each detected path is shaped as the platform Vulnerability output + processor expects: ``name`` and ``status`` are required, ``details`` + carries the raw line. Negative / failure summaries (e.g. "No potential + privilege escalation methods worked.") also contain a privesc keyword, + so they are explicitly rejected to avoid emitting a false VULNERABLE + finding. + """ + privesc_paths: List[Dict] = [] + seen = set() for line in stdout.split("\n"): line = line.strip() + lowered = line.lower() # Look for privilege escalation indicators if any( - keyword in line.lower() + keyword in lowered for keyword in ["escalation", "privesc", "vulnerable", "exploit"] ): - if line not in privesc_paths: - privesc_paths.append(line) + if not line or line in seen: + continue + # Skip lines that report the absence of a finding. + if any(neg in lowered for neg in self._PRIVESC_NEGATIVE_MARKERS): + continue + seen.add(line) + privesc_paths.append( + { + "name": line[:120], + "status": "VULNERABLE", + "details": line, + } + ) return privesc_paths def _parse_ec2_data(self, stdout: str) -> Tuple[List[str], List[str]]: @@ -939,6 +1036,7 @@ def _generic_parse(self, module_name: str, stdout: str) -> Dict: ) elif "vpc" in module_name: outputs["vpcs"] = self._extract_items(stdout, ["vpc-", "VPC:", "VpcId:"]) + outputs["public_ips"] = self._extract_public_ipv4s(stdout) elif "cloudtrail" in module_name: outputs["events"] = self._extract_items( stdout, ["Event:", "EventName:", "Trail:"] @@ -998,6 +1096,84 @@ def _generic_parse(self, module_name: str, stdout: str) -> Dict: return outputs + def _parse_secrets_credentials(self, stdout: str) -> List[Dict]: + """Parse Secrets Manager identifiers into Credentials findings.""" + identifiers = self._extract_items(stdout, ["SecretName:", "Secret:", "ARN:"]) + return self._identifiers_to_credentials(identifiers) + + def _parse_ssm_credentials(self, stdout: str) -> List[Dict]: + """Parse SSM parameter identifiers into Credentials findings.""" + identifiers = self._extract_items(stdout, ["Parameter:", "Name:", "SSM:"]) + return self._identifiers_to_credentials(identifiers) + + def _identifiers_to_credentials(self, identifiers: List[str]) -> List[Dict]: + """Shape credential-store identifiers as Credentials findings. + + The enumeration modules surface the identifier of a secret / parameter, + not the plaintext value, so the identifier is used as both the username + and the hash to satisfy the platform Credentials validator (username + + password OR hash). The finding acts as a lead for credential-reuse + injects. + """ + credentials: List[Dict] = [] + seen = set() + for identifier in identifiers: + if not identifier or identifier in seen: + continue + seen.add(identifier) + credentials.append({"username": identifier, "hash": identifier}) + return credentials + + def _extract_public_ipv4s(self, stdout: str) -> List[str]: + """Extract distinct, globally-routable IPv4 addresses from Pacu output.""" + public_ips: List[str] = [] + seen = set() + for candidate in re.findall(r"\b(?:\d{1,3}\.){3}\d{1,3}\b", stdout): + if candidate in seen: + continue + try: + address = ipaddress.ip_address(candidate) + except ValueError: + continue + if address.version != 4 or not address.is_global: + continue + seen.add(candidate) + public_ips.append(candidate) + return public_ips + + def _extract_open_ports(self, stdout: str) -> List[int]: + """Extract distinct open security-group ports from Pacu output. + + Security-group rules are expressed as ``FromPort`` / ``ToPort`` pairs. A + pair only identifies a single open port when ``FromPort == ToPort``; a + genuine range (e.g. ``FromPort: 80 ToPort: 82``) cannot be represented by + the contract's list-of-single-``Port`` output, so it is skipped rather + than misrepresented as just its two endpoints. Explicit "open port N" + lines are parsed separately. + """ + open_ports: List[int] = [] + seen = set() + + def _add(value: int) -> None: + if 0 < value <= 65535 and value not in seen: + seen.add(value) + open_ports.append(value) + + # Single-port security-group rules (FromPort == ToPort). ``\D+`` between + # the two values never crosses another digit, so it cannot pair a + # FromPort with a ToPort from a different rule. + for from_port, to_port in re.findall( + r"FromPort\W{0,3}(\d{1,5})\D+ToPort\W{0,3}(\d{1,5})", stdout + ): + if from_port == to_port: + _add(int(from_port)) + + # Explicit "open port N" mentions. + for match in re.findall(r"open port\W{0,3}(\d{1,5})", stdout, re.IGNORECASE): + _add(int(match)) + + return open_ports + def _extract_items(self, text: str, patterns: List[str]) -> List[str]: """Extract items matching any of the given patterns""" items = [] diff --git a/aws/test/__init__.py b/aws/test/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/aws/test/test_pacu_executor_findings.py b/aws/test/test_pacu_executor_findings.py new file mode 100644 index 00000000..9fb600d7 --- /dev/null +++ b/aws/test/test_pacu_executor_findings.py @@ -0,0 +1,128 @@ +"""Unit tests for the AWS Pacu result parser semantic finding shapes. + +Covers the P2 semantic typing: Secrets Manager / SSM -> Credentials, IAM privesc +-> Vulnerability, and EC2 / VPC -> IPv4 (public IPs) + Port (open SG ports). The +parser imports only the standard library, so these tests need neither pacu nor +awscli installed. +""" + +from aws.helpers.pacu_executor import PacuExecutor + + +def _executor(): + return PacuExecutor(logger=None) + + +def test_privesc_paths_emit_vulnerability_findings(): + results = { + "success": True, + "module": "iam__privesc_scan", + "data": { + "stdout": ( + "Potential privilege escalation found: CreateAccessKey\n" + "Vulnerable path via iam:PassRole to admin role\n" + "No potential privilege escalation methods worked.\n" + "just an informational line with no signal\n" + ) + }, + } + + outputs = _executor().parse_results(results)["outputs"] + paths = outputs["privesc_paths"] + + assert len(paths) == 2 + details = {p["details"] for p in paths} + assert "No potential privilege escalation methods worked." not in details + for path in paths: + assert path["status"] == "VULNERABLE" + assert path["name"] + assert path["details"] + + +def test_privesc_negative_summaries_are_not_emitted(): + """Failure/absence summaries must not become VULNERABLE findings.""" + stdout = ( + "No privilege escalation paths found.\n" + "Target is not vulnerable to any known privesc.\n" + "Could not find an exploit chain.\n" + ) + paths = _executor()._parse_privesc_paths(stdout) + assert paths == [] + + +def test_secrets_manager_emits_credentials(): + results = { + "success": True, + "module": "secrets__enum", + "data": {"stdout": "SecretName: prod/db/password\nSecret: staging/api-key\n"}, + } + + creds = _executor().parse_results(results)["outputs"]["secrets"] + + assert {c["username"] for c in creds} == { + "prod/db/password", + "staging/api-key", + } + for cred in creds: + assert cred["hash"] == cred["username"] + + +def test_ssm_parameters_emit_credentials(): + results = { + "success": True, + "module": "systemsmanager__download_parameters", + "data": {"stdout": "Parameter: /prod/token\nName: /prod/other\n"}, + } + + params = _executor().parse_results(results)["outputs"]["parameters"] + + assert params + for cred in params: + assert cred["username"] + assert cred["hash"] + + +def test_ec2_enum_emits_public_ips_and_open_ports(): + stdout = ( + "Instance i-0123456789abcdef0 in sg-0abc1234\n" + "PublicIpAddress: 51.38.220.153\n" + "PrivateIpAddress: 10.0.0.5\n" + "SecurityGroup rule FromPort: 22 ToPort: 22\n" + "Open port 443 to 0.0.0.0/0\n" + ) + results = { + "success": True, + "module": "ec2__enum", + "data": {"stdout": stdout}, + } + + outputs = _executor().parse_results(results)["outputs"] + + assert "51.38.220.153" in outputs["public_ips"] + assert "10.0.0.5" not in outputs["public_ips"] + assert 22 in outputs["open_ports"] + assert 443 in outputs["open_ports"] + assert outputs["instances"] == ["i-0123456789abcdef0"] + + +def test_extract_public_ipv4s_excludes_private_ranges(): + executor = _executor() + ips = executor._extract_public_ipv4s( + "8.8.8.8 10.1.2.3 192.168.1.1 172.16.0.1 1.1.1.1 not.an.ip" + ) + assert set(ips) == {"8.8.8.8", "1.1.1.1"} + + +def test_extract_open_ports_single_port_rules_and_explicit_lines(): + executor = _executor() + ports = executor._extract_open_ports( + "SecurityGroup rule FromPort: 22 ToPort: 22\n" "Open port 443 to 0.0.0.0/0\n" + ) + assert ports == [22, 443] + + +def test_extract_open_ports_skips_multi_port_ranges(): + """A genuine FromPort/ToPort range must not be emitted as its endpoints.""" + executor = _executor() + ports = executor._extract_open_ports("FromPort: 8000 ToPort: 8010") + assert ports == [] diff --git a/shodan/shodan/contracts/cloud_provider_asset_discovery/contract.py b/shodan/shodan/contracts/cloud_provider_asset_discovery/contract.py index fe67d082..d18591bf 100644 --- a/shodan/shodan/contracts/cloud_provider_asset_discovery/contract.py +++ b/shodan/shodan/contracts/cloud_provider_asset_discovery/contract.py @@ -11,6 +11,8 @@ SupportedLanguage, ) +from shodan.contracts.finding_outputs import ipv4_output + if TYPE_CHECKING: from shodan.contracts.shodan_contracts import TargetSelectorField @@ -207,7 +209,7 @@ def contract_with_specific_fields( def contract_with_specific_outputs( base_outputs: list[ContractOutputElement], ) -> list[ContractOutputElement]: - specific_outputs = [] + specific_outputs = [ipv4_output()] contract_outputs = ( ContractBuilder() .add_outputs(base_outputs + specific_outputs) diff --git a/shodan/shodan/contracts/critical_ports_and_exposed_admin_interface/contract.py b/shodan/shodan/contracts/critical_ports_and_exposed_admin_interface/contract.py index c4cfc33b..8b9a75f9 100644 --- a/shodan/shodan/contracts/critical_ports_and_exposed_admin_interface/contract.py +++ b/shodan/shodan/contracts/critical_ports_and_exposed_admin_interface/contract.py @@ -11,6 +11,12 @@ SupportedLanguage, ) +from shodan.contracts.finding_outputs import ( + cve_output, + ipv4_output, + ports_scan_output, +) + if TYPE_CHECKING: from shodan.contracts.shodan_contracts import TargetSelectorField @@ -222,7 +228,7 @@ def contract_with_specific_fields( def contract_with_specific_outputs( base_outputs: list[ContractOutputElement], ) -> list[ContractOutputElement]: - specific_outputs = [] + specific_outputs = [ports_scan_output(), ipv4_output(), cve_output()] contract_outputs = ( ContractBuilder() .add_outputs(base_outputs + specific_outputs) diff --git a/shodan/shodan/contracts/custom_query/contract.py b/shodan/shodan/contracts/custom_query/contract.py index cb341fb8..97c74c05 100644 --- a/shodan/shodan/contracts/custom_query/contract.py +++ b/shodan/shodan/contracts/custom_query/contract.py @@ -10,6 +10,12 @@ SupportedLanguage, ) +from shodan.contracts.finding_outputs import ( + cve_output, + ipv4_output, + ports_scan_output, +) + if TYPE_CHECKING: from shodan.contracts.shodan_contracts import TargetSelectorField @@ -149,7 +155,7 @@ def contract_with_specific_fields( def contract_with_specific_outputs( base_outputs: list[ContractOutputElement], ) -> list[ContractOutputElement]: - specific_outputs = [] + specific_outputs = [ports_scan_output(), ipv4_output(), cve_output()] contract_outputs = ( ContractBuilder() .add_outputs(base_outputs + specific_outputs) diff --git a/shodan/shodan/contracts/cve_enumeration/contract.py b/shodan/shodan/contracts/cve_enumeration/contract.py index f3f8207e..24b72c25 100644 --- a/shodan/shodan/contracts/cve_enumeration/contract.py +++ b/shodan/shodan/contracts/cve_enumeration/contract.py @@ -10,6 +10,12 @@ SupportedLanguage, ) +from shodan.contracts.finding_outputs import ( + cve_output, + ipv4_output, + ports_scan_output, +) + if TYPE_CHECKING: from shodan.contracts.shodan_contracts import TargetSelectorField @@ -184,7 +190,7 @@ def contract_with_specific_fields( def contract_with_specific_outputs( base_outputs: list[ContractOutputElement], ) -> list[ContractOutputElement]: - specific_outputs = [] + specific_outputs = [cve_output(), ports_scan_output(), ipv4_output()] contract_outputs = ( ContractBuilder() .add_outputs(base_outputs + specific_outputs) diff --git a/shodan/shodan/contracts/cve_specific_watchlist/contract.py b/shodan/shodan/contracts/cve_specific_watchlist/contract.py index ede821d0..2bffb955 100644 --- a/shodan/shodan/contracts/cve_specific_watchlist/contract.py +++ b/shodan/shodan/contracts/cve_specific_watchlist/contract.py @@ -11,6 +11,12 @@ SupportedLanguage, ) +from shodan.contracts.finding_outputs import ( + cve_output, + ipv4_output, + ports_scan_output, +) + if TYPE_CHECKING: from shodan.contracts.shodan_contracts import TargetSelectorField @@ -201,7 +207,7 @@ def contract_with_specific_fields( def contract_with_specific_outputs( base_outputs: list[ContractOutputElement], ) -> list[ContractOutputElement]: - specific_outputs = [] + specific_outputs = [cve_output(), ports_scan_output(), ipv4_output()] contract_outputs = ( ContractBuilder() .add_outputs(base_outputs + specific_outputs) diff --git a/shodan/shodan/contracts/domain_discovery/contract.py b/shodan/shodan/contracts/domain_discovery/contract.py index e404654c..3491f6d6 100644 --- a/shodan/shodan/contracts/domain_discovery/contract.py +++ b/shodan/shodan/contracts/domain_discovery/contract.py @@ -10,6 +10,8 @@ SupportedLanguage, ) +from shodan.contracts.finding_outputs import ipv4_output, ports_scan_output + if TYPE_CHECKING: from shodan.contracts.shodan_contracts import TargetSelectorField @@ -184,7 +186,7 @@ def contract_with_specific_fields( def contract_with_specific_outputs( base_outputs: list[ContractOutputElement], ) -> list[ContractOutputElement]: - specific_outputs = [] + specific_outputs = [ipv4_output(), ports_scan_output()] contract_outputs = ( ContractBuilder() .add_outputs(base_outputs + specific_outputs) diff --git a/shodan/shodan/contracts/finding_outputs.py b/shodan/shodan/contracts/finding_outputs.py new file mode 100644 index 00000000..7c088ebe --- /dev/null +++ b/shodan/shodan/contracts/finding_outputs.py @@ -0,0 +1,58 @@ +"""Finding-compatible contract outputs for the Shodan injector. + +Shodan already retrieves ports, IPs, hostnames and CVEs for every query, but +historically rendered them into a display table only, declaring a single +non-finding-compatible ``Asset`` output. These builders expose that same data as +finding-compatible ``ContractOutputElement`` so downstream attack-path chaining +can trigger off the ``IPv4``, ``PortsScan`` and ``CVE`` finding primitives, +following the nmap / netexec / censys reference pattern. + +The output-dict field names below MUST stay in sync with the keys produced by +``ShodanFindingsParser`` and with the platform OutputProcessor field contracts: + +- IPv4 -> primitive IPv4 string values (IPv4OutputProcessor) +- PortsScan -> ``{host, port, service, asset_id}`` (PortScanOutputProcessor) +- CVE -> ``{id, host, severity, asset_id}`` (CVEOutputProcessor) + +A declared finding type with no extractor stays empty, so these declarations are +always shipped together with ``ShodanFindingsParser``. +""" + +from pyoaev.contracts.contract_config import ContractOutputElement, ContractOutputType + +IPV4_FIELD = "ipv4" +PORTS_SCAN_FIELD = "ports_scan" +CVE_FIELD = "cve" + + +def ipv4_output() -> ContractOutputElement: + """Build the finding-compatible IPv4 output element (primitive strings).""" + return ContractOutputElement( + type=ContractOutputType.IPv4, + field=IPV4_FIELD, + isMultiple=True, + isFindingCompatible=True, + labels=["shodan", "ipv4"], + ) + + +def ports_scan_output() -> ContractOutputElement: + """Build the finding-compatible PortsScan output element (host/port/service).""" + return ContractOutputElement( + type=ContractOutputType.PortsScan, + field=PORTS_SCAN_FIELD, + isMultiple=True, + isFindingCompatible=True, + labels=["shodan", "port"], + ) + + +def cve_output() -> ContractOutputElement: + """Build the finding-compatible CVE output element (id/host/severity).""" + return ContractOutputElement( + type=ContractOutputType.CVE, + field=CVE_FIELD, + isMultiple=True, + isFindingCompatible=True, + labels=["shodan", "cve"], + ) diff --git a/shodan/shodan/contracts/ip_enumeration/contract.py b/shodan/shodan/contracts/ip_enumeration/contract.py index 10442d52..c2ab9af5 100644 --- a/shodan/shodan/contracts/ip_enumeration/contract.py +++ b/shodan/shodan/contracts/ip_enumeration/contract.py @@ -11,6 +11,12 @@ SupportedLanguage, ) +from shodan.contracts.finding_outputs import ( + cve_output, + ipv4_output, + ports_scan_output, +) + if TYPE_CHECKING: from shodan.contracts.shodan_contracts import TargetSelectorField @@ -173,7 +179,7 @@ def contract_with_specific_fields( def contract_with_specific_outputs( base_outputs: list[ContractOutputElement], ) -> list[ContractOutputElement]: - specific_outputs = [] + specific_outputs = [ports_scan_output(), ipv4_output(), cve_output()] contract_outputs = ( ContractBuilder() .add_outputs(base_outputs + specific_outputs) diff --git a/shodan/shodan/injector/openaev_shodan.py b/shodan/shodan/injector/openaev_shodan.py index 4bbc6ffb..16b326d1 100644 --- a/shodan/shodan/injector/openaev_shodan.py +++ b/shodan/shodan/injector/openaev_shodan.py @@ -24,7 +24,7 @@ ContractType, NormalizeInputData, ) -from shodan.services import ShodanClientAPI, Utils +from shodan.services import ShodanClientAPI, ShodanFindingsParser, Utils LOG_PREFIX = "[SHODAN_INJECTOR]" @@ -38,6 +38,7 @@ def __init__(self, config: ConfigLoader, helper: OpenAEVInjectorHelper): self.helper = helper self.shodan_client_api = ShodanClientAPI(self.config, self.helper) self.utils = Utils() + self.findings_parser = ShodanFindingsParser() def _prepare_output_message( self, normalize_input_data: NormalizeInputData, results, user_info: dict @@ -442,9 +443,16 @@ def _shodan_execution(self, data: dict): ) # Preparation and creation of auto_create_assets - output_structured = "" + output_structured = {} if normalize_input_data.inject_content.auto_create_assets: - output_structured = self._prepare_output_structured(shodan_results) + output_structured.update(self._prepare_output_structured(shodan_results)) + + # Parse Shodan matches into finding-compatible structured outputs + # (IPv4 / PortsScan / CVE). Emitted for every execution, independently + # of auto_create_assets, so downstream chaining can trigger off them. + output_structured.update( + self.findings_parser.parse(normalize_input_data, shodan_results) + ) # Preparation and creation of output_message output_message = self._prepare_output_message( diff --git a/shodan/shodan/services/__init__.py b/shodan/shodan/services/__init__.py index 53803ca8..e00e9caa 100644 --- a/shodan/shodan/services/__init__.py +++ b/shodan/shodan/services/__init__.py @@ -1,7 +1,9 @@ from shodan.services.client_api import ShodanClientAPI +from shodan.services.findings_parser import ShodanFindingsParser from shodan.services.utils import Utils __all__ = [ "ShodanClientAPI", + "ShodanFindingsParser", "Utils", ] diff --git a/shodan/shodan/services/findings_parser.py b/shodan/shodan/services/findings_parser.py new file mode 100644 index 00000000..e57e4c97 --- /dev/null +++ b/shodan/shodan/services/findings_parser.py @@ -0,0 +1,152 @@ +"""Parse Shodan search results into finding-compatible structured outputs. + +Turns the raw Shodan matches (already retrieved for every contract) into the +``IPv4``, ``PortsScan`` and ``CVE`` finding shapes declared in +``shodan.contracts.finding_outputs``. The output-dict keys MUST match the +declared ``ContractOutputElement`` fields and the platform OutputProcessor +contracts. + +Only findings that satisfy the platform validators are emitted: + +- IPv4 -> a valid IPv4 address string +- PortsScan -> host + numeric port + service +- CVE -> id + host + severity + +``asset_id`` is attached only when a match maps back to a resolved target asset, +mirroring the netexec / ai-redteam convention of never emitting a null asset id. +""" + +import ipaddress + +from shodan.contracts.finding_outputs import CVE_FIELD, IPV4_FIELD, PORTS_SCAN_FIELD + + +class ShodanFindingsParser: + """Extract IPv4 / PortsScan / CVE findings from Shodan search results.""" + + @staticmethod + def _is_ipv4(value) -> bool: + try: + return isinstance(value, str) and ipaddress.ip_address(value).version == 4 + except ValueError: + return False + + @staticmethod + def _build_asset_map(targets: dict) -> dict: + """Map every known target IP / seen IP / hostname to its asset id.""" + asset_map: dict = {} + for asset in targets.get("assets", []) or []: + asset_id = asset.get("asset_id") + if not asset_id: + continue + for ip in asset.get("asset_ips") or []: + if ip: + asset_map.setdefault(ip, asset_id) + seen_ip = asset.get("asset_seen_ip") + if seen_ip: + asset_map.setdefault(seen_ip, asset_id) + hostname = asset.get("asset_hostname") + if hostname: + asset_map.setdefault(hostname, asset_id) + return asset_map + + @staticmethod + def _iter_elements(shodan_results: dict): + """Yield every Shodan match across all target responses.""" + for item in shodan_results.get("data", []) or []: + if not isinstance(item, dict) or item.get("is_error"): + continue + result = item.get("result") or {} + if not isinstance(result, dict): + continue + elements = result.get("matches") + if elements is None: + # CVE_SPECIFIC_WATCHLIST responses expose matches under "data". + elements = result.get("data", []) + for element in elements or []: + if isinstance(element, dict): + yield element + + @staticmethod + def _resolve_asset_id(element: dict, ip_str, asset_map: dict): + if ip_str and ip_str in asset_map: + return asset_map[ip_str] + for hostname in element.get("hostnames", []) or []: + if hostname in asset_map: + return asset_map[hostname] + return None + + @staticmethod + def _iter_cves(vulns): + """Yield (cve_id, severity) pairs from a Shodan ``vulns`` field. + + Shodan exposes ``vulns`` either as a mapping ``{CVE-ID: {cvss: float}}`` + or, more rarely, as a plain list of CVE identifiers. + """ + if isinstance(vulns, dict): + for cve_id, meta in vulns.items(): + severity = "Unknown" + if isinstance(meta, dict) and meta.get("cvss") is not None: + severity = str(meta.get("cvss")) + yield str(cve_id).upper(), severity + elif isinstance(vulns, list): + for cve_id in vulns: + if isinstance(cve_id, str): + yield cve_id.upper(), "Unknown" + + def parse(self, normalize_input_data, shodan_results: dict) -> dict: + """Return the IPv4 / PortsScan / CVE findings for a Shodan execution.""" + targets = normalize_input_data.targets.model_dump() + asset_map = self._build_asset_map(targets) + + ipv4_seen: set = set() + ipv4_findings: list = [] + ports_scan_seen: set = set() + ports_scan_findings: list = [] + cve_seen: set = set() + cve_findings: list = [] + + for element in self._iter_elements(shodan_results): + ip_str = element.get("ip_str") + asset_id = self._resolve_asset_id(element, ip_str, asset_map) + host = ip_str or "" + + if self._is_ipv4(ip_str) and ip_str not in ipv4_seen: + ipv4_seen.add(ip_str) + ipv4_findings.append(ip_str) + + port = element.get("port") + if host and isinstance(port, int) and not isinstance(port, bool): + key = (host, port) + if key not in ports_scan_seen: + ports_scan_seen.add(key) + finding = { + "host": host, + "port": port, + "service": ( + element.get("product") + or element.get("transport") + or "unknown" + ), + } + if asset_id: + finding["asset_id"] = asset_id + ports_scan_findings.append(finding) + + for cve_id, severity in self._iter_cves(element.get("vulns")): + if not host: + continue + key = (cve_id, host) + if key in cve_seen: + continue + cve_seen.add(key) + finding = {"id": cve_id, "host": host, "severity": severity} + if asset_id: + finding["asset_id"] = asset_id + cve_findings.append(finding) + + return { + IPV4_FIELD: ipv4_findings, + PORTS_SCAN_FIELD: ports_scan_findings, + CVE_FIELD: cve_findings, + } diff --git a/shodan/tests/shodan_contracts/test_finding_outputs.py b/shodan/tests/shodan_contracts/test_finding_outputs.py new file mode 100644 index 00000000..ae8a49aa --- /dev/null +++ b/shodan/tests/shodan_contracts/test_finding_outputs.py @@ -0,0 +1,53 @@ +"""Every Shodan contract must declare its finding-compatible outputs. + +Shodan retrieves ports, IPs and CVEs for every query; these must be exposed as +finding-compatible ContractOutputElement so downstream chaining can trigger off +them. This test locks in the per-contract finding declarations. +""" + +import json + +from shodan.contracts.shodan_contracts import ShodanContractId, ShodanContracts + +# Expected finding-compatible output types (serialized enum values) per contract. +EXPECTED_FINDING_TYPES = { + ShodanContractId.CLOUD_PROVIDER_ASSET_DISCOVERY.value: {"ipv4"}, + ShodanContractId.CRITICAL_PORTS_AND_EXPOSED_ADMIN_INTERFACE.value: { + "portscan", + "ipv4", + "cve", + }, + ShodanContractId.CUSTOM_QUERY.value: {"portscan", "ipv4", "cve"}, + ShodanContractId.CVE_ENUMERATION.value: {"portscan", "ipv4", "cve"}, + ShodanContractId.CVE_SPECIFIC_WATCHLIST.value: {"portscan", "ipv4", "cve"}, + ShodanContractId.DOMAIN_DISCOVERY.value: {"ipv4", "portscan"}, + ShodanContractId.IP_ENUMERATION.value: {"portscan", "ipv4", "cve"}, +} + + +def _finding_types(contract): + content = json.loads(contract["contract_content"]) + return { + output["type"] for output in content["outputs"] if output["isFindingCompatible"] + } + + +def test_every_contract_declares_expected_finding_outputs(): + contracts = ShodanContracts().contracts() + assert len(contracts) == len(ShodanContractId) + + for contract in contracts: + contract_id = contract["contract_id"] + assert contract_id in EXPECTED_FINDING_TYPES, contract_id + assert ( + _finding_types(contract) == EXPECTED_FINDING_TYPES[contract_id] + ), contract_id + + +def test_base_asset_output_is_not_finding_compatible(): + for contract in ShodanContracts().contracts(): + content = json.loads(contract["contract_content"]) + asset_outputs = [o for o in content["outputs"] if o["type"] == "asset"] + assert asset_outputs, contract["contract_id"] + for asset_output in asset_outputs: + assert asset_output["isFindingCompatible"] is False diff --git a/shodan/tests/unit/test_findings_parser.py b/shodan/tests/unit/test_findings_parser.py new file mode 100644 index 00000000..bac66796 --- /dev/null +++ b/shodan/tests/unit/test_findings_parser.py @@ -0,0 +1,144 @@ +"""Unit tests for ShodanFindingsParser (IPv4 / PortsScan / CVE extraction).""" + +from unittest.mock import MagicMock + +from shodan.services.findings_parser import ShodanFindingsParser + + +def _normalize_input_data(assets): + """Build a stand-in NormalizeInputData exposing targets.model_dump().""" + nid = MagicMock() + nid.targets.model_dump.return_value = {"assets": assets} + return nid + + +def _results(matches): + return { + "targets": ["target"], + "data": [ + { + "target": "target", + "url": "GET https://api.shodan.io/...", + "result": {"matches": matches, "total": len(matches)}, + } + ], + } + + +def test_parse_extracts_ipv4_portscan_and_cve_with_asset_mapping(): + assets = [ + { + "asset_id": "asset-1", + "asset_hostname": None, + "asset_ips": ["51.38.220.153"], + "asset_seen_ip": None, + } + ] + matches = [ + { + "ip_str": "51.38.220.153", + "hostnames": ["automation.filigran.io"], + "port": 443, + "product": "nginx", + "vulns": { + "CVE-2023-44487": {"cvss": 7.5}, + "cve-2025-23419": {"cvss": 4.3}, + }, + } + ] + + outputs = ShodanFindingsParser().parse( + _normalize_input_data(assets), _results(matches) + ) + + assert outputs["ipv4"] == ["51.38.220.153"] + assert outputs["ports_scan"] == [ + { + "host": "51.38.220.153", + "port": 443, + "service": "nginx", + "asset_id": "asset-1", + } + ] + assert {(c["id"], c["severity"]) for c in outputs["cve"]} == { + ("CVE-2023-44487", "7.5"), + ("CVE-2025-23419", "4.3"), + } + for cve in outputs["cve"]: + assert cve["host"] == "51.38.220.153" + assert cve["asset_id"] == "asset-1" + + +def test_parse_manual_mode_omits_asset_id_and_falls_back_on_service(): + matches = [ + { + "ip_str": "1.1.1.1", + "hostnames": [], + "port": 22, + "transport": "tcp", + }, + { + "ip_str": "8.8.8.8", + "hostnames": [], + "port": 53, + }, + ] + + outputs = ShodanFindingsParser().parse(_normalize_input_data([]), _results(matches)) + + assert outputs["ipv4"] == ["1.1.1.1", "8.8.8.8"] + assert outputs["ports_scan"] == [ + {"host": "1.1.1.1", "port": 22, "service": "tcp"}, + {"host": "8.8.8.8", "port": 53, "service": "unknown"}, + ] + assert outputs["cve"] == [] + + +def test_parse_vulns_as_list_yields_unknown_severity(): + matches = [ + { + "ip_str": "9.9.9.9", + "hostnames": [], + "vulns": ["CVE-2021-1234"], + } + ] + + outputs = ShodanFindingsParser().parse(_normalize_input_data([]), _results(matches)) + + assert outputs["cve"] == [ + {"id": "CVE-2021-1234", "host": "9.9.9.9", "severity": "Unknown"} + ] + + +def test_parse_deduplicates_and_skips_errors_and_non_ipv4(): + results = { + "targets": ["a", "b"], + "data": [ + {"target": "a", "is_error": True, "response": {}}, + { + "target": "b", + "result": { + "matches": [ + {"ip_str": "1.2.3.4", "hostnames": [], "port": 80}, + {"ip_str": "1.2.3.4", "hostnames": [], "port": 80}, + { + "ip_str": "2001:db8::1", + "hostnames": [], + "port": 8080, + }, + ] + }, + }, + ], + } + + outputs = ShodanFindingsParser().parse(_normalize_input_data([]), results) + + # IPv4 list only holds the valid, de-duplicated IPv4 address. + assert outputs["ipv4"] == ["1.2.3.4"] + # PortsScan keeps one entry per (host, port); the IPv6 host is still a + # valid PortsScan host even though it is excluded from the IPv4 list. + assert outputs["ports_scan"] == [ + {"host": "1.2.3.4", "port": 80, "service": "unknown"}, + {"host": "2001:db8::1", "port": 8080, "service": "unknown"}, + ]