From e574baad939be2402aa48923032c9cdc1c1fa664 Mon Sep 17 00:00:00 2001 From: dextmorgn Date: Mon, 3 Nov 2025 18:41:08 +0100 Subject: [PATCH] fix(transforms): fixes usage of asnmap + other docker tools fix #31 --- .../src/flowsint_transforms/asn/to_cidrs.py | 169 ++++++++++-------- .../src/flowsint_transforms/cidr/to_ips.py | 129 ++++++++----- .../src/flowsint_transforms/domain/to_asn.py | 135 +++++++++++--- .../src/flowsint_transforms/ip/to_asn.py | 88 +++++++-- flowsint-transforms/src/tools/dockertool.py | 11 +- .../src/tools/network/asnmap.py | 43 ++++- flowsint-transforms/src/tools/network/dnsx.py | 88 +++++++++ .../src/tools/network/mapcidr.py | 109 +++++++++++ 8 files changed, 603 insertions(+), 169 deletions(-) create mode 100644 flowsint-transforms/src/tools/network/dnsx.py create mode 100644 flowsint-transforms/src/tools/network/mapcidr.py diff --git a/flowsint-transforms/src/flowsint_transforms/asn/to_cidrs.py b/flowsint-transforms/src/flowsint_transforms/asn/to_cidrs.py index 724e174d..251613fb 100644 --- a/flowsint-transforms/src/flowsint_transforms/asn/to_cidrs.py +++ b/flowsint-transforms/src/flowsint_transforms/asn/to_cidrs.py @@ -1,20 +1,55 @@ import json -import subprocess -from typing import List, Dict, Any, Union +import os +from typing import List, Dict, Any, Union, Optional from flowsint_core.core.transform_base import Transform +from flowsint_core.core.graph_db import Neo4jConnection from flowsint_types.cidr import CIDR from flowsint_types.asn import ASN from flowsint_core.utils import is_valid_asn, parse_asn from flowsint_core.core.logger import Logger +from tools.network.asnmap import AsnmapTool class AsnToCidrsTransform(Transform): - """Takes an ASN and returns its corresponding CIDRs.""" + """[ASNMAP] Takes an ASN and returns its corresponding CIDRs.""" # Define types as class attributes - base class handles schema generation automatically InputType = List[ASN] OutputType = List[CIDR] + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + neo4j_conn: Optional[Neo4jConnection] = None, + vault=None, + params: Optional[Dict[str, Any]] = None, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + neo4j_conn=neo4j_conn, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + ) + + @classmethod + def required_params(cls) -> bool: + return True + + @classmethod + def get_params_schema(cls) -> List[Dict[str, Any]]: + """Declare required parameters for this transform""" + return [ + { + "name": "PDCP_API_KEY", + "type": "vaultSecret", + "description": "The ProjectDiscovery Cloud Platform API key for asnmap.", + "required": True, + }, + ] + @classmethod def name(cls) -> str: return "asn_to_cidrs" @@ -25,7 +60,7 @@ class AsnToCidrsTransform(Transform): @classmethod def key(cls) -> str: - return "network" + return "number" def preprocess( self, data: Union[List[str], List[int], List[dict], InputType] @@ -51,79 +86,54 @@ class AsnToCidrsTransform(Transform): """Find CIDR from ASN using asnmap.""" cidrs: OutputType = [] self._asn_to_cidrs_map = [] # Store mapping for postprocess + asnmap = AsnmapTool() + + # Retrieve API key from vault or environment + api_key = self.get_secret("PDCP_API_KEY", os.getenv("PDCP_API_KEY")) for asn in data: - asn_cidrs = [] - cidr_data = self.__get_cidrs_from_asn(asn.number) - if cidr_data and "as_range" in cidr_data and cidr_data["as_range"]: - # Add all CIDRs for this ASN - for cidr_str in cidr_data["as_range"]: - try: - cidr = CIDR(network=cidr_str) - cidrs.append(cidr) - asn_cidrs.append(cidr) - except Exception as e: - Logger.error( - self.sketch_id, - {"message": f"Failed to parse CIDR {cidr_str}: {str(e)}"}, - ) - else: - Logger.warn( - self.sketch_id, {"message": f"No CIDRs found for ASN {asn.number}"} - ) - - if asn_cidrs: # Only add to mapping if we found valid CIDRs - self._asn_to_cidrs_map.append((asn, asn_cidrs)) - return cidrs - - def __get_cidrs_from_asn(self, asn: int) -> Dict[str, Any]: - try: - command = f"echo {asn} | asnmap -silent -json | jq -s '.'" - result = subprocess.run( - command, shell=True, capture_output=True, text=True, timeout=60 - ) - if not result.stdout.strip(): - Logger.info(self.sketch_id, {"message": f"No CIDRs found for {asn}."}) - return None try: - # Parse the JSON array - data_array = json.loads(result.stdout) - if not data_array: - return None + asn_cidrs = [] + # Use asnmap tool to get CIDR info, passing the API key + # asnmap expects ASN with "AS" prefix + cidr_data = asnmap.launch(f"AS{asn.number}", type="asn", api_key=api_key) - combined_data = { - "as_range": [], - "as_name": None, - "as_country": None, - "as_number": None, - } + if cidr_data and "as_range" in cidr_data and cidr_data["as_range"]: + # Add all CIDRs for this ASN + for cidr_str in cidr_data["as_range"]: + try: + cidr = CIDR(network=cidr_str) + cidrs.append(cidr) + asn_cidrs.append(cidr) + except Exception as e: + Logger.error( + self.sketch_id, + {"message": f"Failed to parse CIDR {cidr_str}: {str(e)}"}, + ) - for data in data_array: - if "as_range" in data: - combined_data["as_range"].extend(data["as_range"]) - if data.get("as_name") and not combined_data["as_name"]: - combined_data["as_name"] = data["as_name"] - if data.get("as_country") and not combined_data["as_country"]: - combined_data["as_country"] = data["as_country"] - if data.get("as_number") and not combined_data["as_number"]: - combined_data["as_number"] = data["as_number"] + Logger.info( + self.sketch_id, + { + "message": f"[ASNMAP] Found {len(asn_cidrs)} CIDRs for AS{asn.number}" + }, + ) + else: + Logger.warn( + self.sketch_id, + {"message": f"[ASNMAP] No CIDRs found for AS{asn.number}"}, + ) - return combined_data if combined_data["as_range"] else None + if asn_cidrs: # Only add to mapping if we found valid CIDRs + self._asn_to_cidrs_map.append((asn, asn_cidrs)) - except json.JSONDecodeError: + except Exception as e: Logger.error( self.sketch_id, - { - "message": f"Failed to parse JSON from asnmap output: {result.stdout}" - }, + {"message": f"Error getting CIDRs for ASN {asn.number}: {e}"}, ) - return None + continue - except Exception as e: - Logger.error( - self.sketch_id, {"message": f"asnmap exception for {asn}: {str(e)}"} - ) - return None + return cidrs def postprocess(self, results: OutputType, original_input: InputType) -> OutputType: # Create Neo4j relationships between ASNs and their corresponding CIDRs @@ -139,10 +149,8 @@ class AsnToCidrsTransform(Transform): "asn", "number", asn.number, - name=asn.name or "Unknown", - country=asn.country or "Unknown", label=f"AS{asn.number}", - caption=f"AS{asn.number} - {asn.name or 'Unknown'}", + caption=f"AS{asn.number}", type="asn", ) @@ -150,6 +158,7 @@ class AsnToCidrsTransform(Transform): "cidr", "network", str(cidr.network), + label=str(cidr.network), caption=str(cidr.network), type="cidr", ) @@ -163,41 +172,47 @@ class AsnToCidrsTransform(Transform): str(cidr.network), "ANNOUNCES", ) + + self.log_graph_message( + f"AS{asn.number} announces CIDR {cidr.network}" + ) else: # Fallback: original behavior (one-to-one zip) for asn, cidr in zip(original_input, results): if str(cidr.network) == "0.0.0.0/0": continue # Skip default CIDR for unknown ASN - self.log_graph_message(f"ASN {asn.number} -> {cidr.network}") if self.neo4j_conn: self.create_node( - "ASN", + "asn", "number", asn.number, - name=asn.name or "Unknown", - country=asn.country or "Unknown", label=f"AS{asn.number}", - caption=f"AS{asn.number} - {asn.name or 'Unknown'}", + caption=f"AS{asn.number}", type="asn", ) self.create_node( - "CIDR", + "cidr", "network", str(cidr.network), + label=str(cidr.network), caption=str(cidr.network), type="cidr", ) self.create_relationship( - "ASN", + "asn", "number", asn.number, - "CIDR", + "cidr", "network", str(cidr.network), "ANNOUNCES", ) + + self.log_graph_message( + f"AS{asn.number} announces CIDR {cidr.network}" + ) return results diff --git a/flowsint-transforms/src/flowsint_transforms/cidr/to_ips.py b/flowsint-transforms/src/flowsint_transforms/cidr/to_ips.py index 1666b135..8e0dc970 100644 --- a/flowsint-transforms/src/flowsint_transforms/cidr/to_ips.py +++ b/flowsint-transforms/src/flowsint_transforms/cidr/to_ips.py @@ -1,22 +1,57 @@ -import subprocess -from typing import List, Union +import os +from typing import Any, Dict, List, Optional, Union from flowsint_core.core.transform_base import Transform +from flowsint_core.core.graph_db import Neo4jConnection from flowsint_types.cidr import CIDR from flowsint_types.ip import Ip from flowsint_core.core.logger import Logger +from tools.network.mapcidr import MapcidrTool class CidrToIpsTransform(Transform): - """Takes a CIDR and returns its corresponding IP addresses.""" + """[MAPCIDR] Takes a CIDR and returns its corresponding IP addresses.""" # Define types as class attributes - base class handles schema generation automatically InputType = List[CIDR] OutputType = List[Ip] + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + neo4j_conn: Optional[Neo4jConnection] = None, + vault=None, + params: Optional[Dict[str, Any]] = None, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + neo4j_conn=neo4j_conn, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + ) + + @classmethod + def required_params(cls) -> bool: + return False + + @classmethod + def get_params_schema(cls) -> List[Dict[str, Any]]: + """Declare optional parameters for this transform""" + return [ + { + "name": "PDCP_API_KEY", + "type": "vaultSecret", + "description": "Optional ProjectDiscovery Cloud Platform API key for mapcidr.", + "required": False, + }, + ] + @classmethod def name(cls) -> str: return "cidr_to_ips" - + @classmethod def key(cls) -> str: return "network" @@ -44,48 +79,50 @@ class CidrToIpsTransform(Transform): return cleaned async def scan(self, data: InputType) -> OutputType: - """Find IP addresses from CIDR using dnsx.""" + """Find IP addresses from CIDR using mapcidr.""" ips: OutputType = [] + mapcidr = MapcidrTool() + + # Retrieve API key from vault or environment (optional) + api_key = self.get_secret("PDCP_API_KEY", os.getenv("PDCP_API_KEY")) + for cidr in data: - ip_addresses = self.__get_ips_from_cidr(cidr.network) - if ip_addresses: - for ip_str in ip_addresses: - try: - ip = Ip(address=ip_str.strip()) - ips.append(ip) - except Exception as e: - Logger.error( - self.sketch_id, - {"message": f"Failed to parse IP {ip_str}: {str(e)}"}, - ) - else: - Logger.warn( - self.sketch_id, {"message": f"No IPs found for CIDR {cidr.network}"} + try: + # Use mapcidr tool to get IPs from CIDR, passing the API key + ip_addresses = mapcidr.launch(cidr.network, api_key=api_key) + + if ip_addresses: + for ip_str in ip_addresses: + try: + ip = Ip(address=ip_str.strip()) + ips.append(ip) + except Exception as e: + Logger.error( + self.sketch_id, + {"message": f"Failed to parse IP {ip_str}: {str(e)}"}, + ) + + Logger.info( + self.sketch_id, + { + "message": f"[MAPCIDR] Found {len(ip_addresses)} IPs for CIDR {cidr.network}" + }, + ) + else: + Logger.warn( + self.sketch_id, + {"message": f"[MAPCIDR] No IPs found for CIDR {cidr.network}"}, + ) + + except Exception as e: + Logger.error( + self.sketch_id, + {"message": f"Error getting IPs for CIDR {cidr.network}: {e}"}, ) + continue + return ips - def __get_ips_from_cidr(self, cidr: str) -> List[str]: - try: - command = f"echo {cidr} | dnsx -ptr" - result = subprocess.run( - command, shell=True, capture_output=True, text=True, timeout=60 - ) - if not result.stdout.strip(): - Logger.info(self.sketch_id, {"message": f"No IPs found for {cidr}."}) - return [] - - # Split the output by newlines and filter out empty lines - ip_addresses = [ - ip.strip() for ip in result.stdout.split("\n") if ip.strip() - ] - return ip_addresses - - except Exception as e: - Logger.error( - self.sketch_id, {"message": f"dnsx exception for {cidr}: {str(e)}"} - ) - return [] - def postprocess(self, results: OutputType, original_input: InputType) -> OutputType: # Create Neo4j relationships between CIDRs and their corresponding IPs for cidr, ip in zip(original_input, results): @@ -95,13 +132,19 @@ class CidrToIpsTransform(Transform): "cidr", "network", str(cidr.network), + label=str(cidr.network), caption=str(cidr.network), type="cidr", ) # Create IP node self.create_node( - "ip", "address", ip.address, caption=ip.address, type="ip" + "ip", + "address", + ip.address, + label=ip.address, + caption=ip.address, + type="ip", ) # Create relationship @@ -115,7 +158,7 @@ class CidrToIpsTransform(Transform): "CONTAINS", ) - self.log_graph_message(f"Found {len(results)} IPs for CIDR {cidr.network}") + self.log_graph_message(f"CIDR {cidr.network} contains IP {ip.address}") return results diff --git a/flowsint-transforms/src/flowsint_transforms/domain/to_asn.py b/flowsint-transforms/src/flowsint_transforms/domain/to_asn.py index ec8684d4..83c6917e 100644 --- a/flowsint-transforms/src/flowsint_transforms/domain/to_asn.py +++ b/flowsint-transforms/src/flowsint_transforms/domain/to_asn.py @@ -1,21 +1,56 @@ import json +import os import socket -import subprocess -from typing import List, Union +from typing import Any, Dict, List, Optional, Union from flowsint_core.core.transform_base import Transform +from flowsint_core.core.graph_db import Neo4jConnection from flowsint_types.domain import Domain from flowsint_types.asn import ASN from flowsint_core.utils import is_valid_domain from flowsint_core.core.logger import Logger +from tools.network.asnmap import AsnmapTool class DomainToAsnTransform(Transform): - """Takes a domain and returns its corresponding ASN.""" + """[ASNMAP] Takes a domain and returns its corresponding ASN.""" # Define types as class attributes - base class handles schema generation automatically InputType = List[Domain] OutputType = List[ASN] + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + neo4j_conn: Optional[Neo4jConnection] = None, + vault=None, + params: Optional[Dict[str, Any]] = None, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + neo4j_conn=neo4j_conn, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + ) + + @classmethod + def required_params(cls) -> bool: + return True + + @classmethod + def get_params_schema(cls) -> List[Dict[str, Any]]: + """Declare required parameters for this transform""" + return [ + { + "name": "PDCP_API_KEY", + "type": "vaultSecret", + "description": "The ProjectDiscovery Cloud Platform API key for asnmap.", + "required": True, + }, + ] + @classmethod def name(cls) -> str: return "domain_to_asn" @@ -26,7 +61,7 @@ class DomainToAsnTransform(Transform): @classmethod def key(cls) -> str: - return "Domain" + return "domain" def preprocess(self, data: Union[List[str], List[dict], InputType]) -> InputType: cleaned: InputType = [] @@ -46,31 +81,43 @@ class DomainToAsnTransform(Transform): async def scan(self, data: InputType) -> OutputType: results: OutputType = [] + asnmap = AsnmapTool() + + # Retrieve API key from vault or environment + api_key = self.get_secret("PDCP_API_KEY", os.getenv("PDCP_API_KEY")) + for domain in data: try: - # First resolve domain to IP - ip = socket.gethostbyname(domain.domain) + # Use asnmap tool to get ASN info from domain, passing the API key + asn_data = asnmap.launch(domain.domain, type="domain", api_key=api_key) - # Use asnmap to get ASN info - result = subprocess.run( - ["asnmap", "-a", ip, "-json"], - capture_output=True, - text=True, - timeout=30, - ) + if asn_data and "as_number" in asn_data: + # Parse ASN number from string like "AS16276" to integer 16276 + asn_string = asn_data["as_number"] + asn_number = int(asn_string.replace("AS", "").replace("as", "")) - if result.returncode == 0: - output = result.stdout.strip() - if output: - asn_data = json.loads(output) - if asn_data and "as_number" in asn_data: - asn = ASN( - asn=str(asn_data["as_number"]), - name=asn_data.get("as_name", ""), - org=asn_data.get("as_org", ""), - country=asn_data.get("as_country", ""), - ) - results.append(asn) + # Create ASN object with correct field mapping + asn = ASN( + number=asn_number, + name=asn_data.get("as_name", ""), + country=asn_data.get("as_country", ""), + description=asn_data.get("as_name", ""), + ) + results.append(asn) + + Logger.info( + self.sketch_id, + { + "message": f"[ASNMAP] Found AS{asn.number} ({asn.name}) for domain {domain.domain}" + }, + ) + else: + Logger.warn( + self.sketch_id, + { + "message": f"[ASNMAP] No ASN data or missing 'as_number' field for domain {domain.domain}" + }, + ) except Exception as e: Logger.error( @@ -84,6 +131,44 @@ class DomainToAsnTransform(Transform): def postprocess( self, results: OutputType, input_data: InputType = None ) -> OutputType: + # Create Neo4j relationships between domains and their corresponding ASNs + if input_data and self.neo4j_conn: + for domain, asn in zip(input_data, results): + # Create domain node + self.create_node( + "domain", + "domain", + domain.domain, + label=domain.domain, + caption=domain.domain, + type="domain", + ) + + # Create ASN node + self.create_node( + "asn", + "number", + asn.number, + label=f"AS{asn.number}", + caption=f"AS{asn.number}", + type="asn", + ) + + # Create relationship + self.create_relationship( + "domain", + "domain", + domain.domain, + "asn", + "number", + asn.number, + "HOSTED_IN", + ) + + self.log_graph_message( + f"Domain {domain.domain} is hosted in AS{asn.number} ({asn.name})" + ) + return results diff --git a/flowsint-transforms/src/flowsint_transforms/ip/to_asn.py b/flowsint-transforms/src/flowsint_transforms/ip/to_asn.py index 49e07bf7..90214072 100644 --- a/flowsint-transforms/src/flowsint_transforms/ip/to_asn.py +++ b/flowsint-transforms/src/flowsint_transforms/ip/to_asn.py @@ -1,6 +1,8 @@ import json -from typing import List, Union +import os +from typing import Any, Dict, List, Optional, Union from flowsint_core.core.transform_base import Transform +from flowsint_core.core.graph_db import Neo4jConnection from flowsint_types.ip import Ip from flowsint_types.asn import ASN from flowsint_core.utils import is_valid_ip @@ -15,6 +17,39 @@ class IpToAsnTransform(Transform): InputType = List[Ip] OutputType = List[ASN] + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + neo4j_conn: Optional[Neo4jConnection] = None, + vault=None, + params: Optional[Dict[str, Any]] = None, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + neo4j_conn=neo4j_conn, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + ) + + @classmethod + def required_params(cls) -> bool: + return True + + @classmethod + def get_params_schema(cls) -> List[Dict[str, Any]]: + """Declare required parameters for this transform""" + return [ + { + "name": "PDCP_API_KEY", + "type": "vaultSecret", + "description": "The ProjectDiscovery Cloud Platform API key for asnmap.", + "required": True, + }, + ] + @classmethod def name(cls) -> str: return "ip_to_asn" @@ -47,20 +82,38 @@ class IpToAsnTransform(Transform): results: OutputType = [] asnmap = AsnmapTool() + # Retrieve API key from vault or environment + api_key = self.get_secret("PDCP_API_KEY", os.getenv("PDCP_API_KEY")) + for ip in data: try: - # Use asnmap tool to get ASN info - asn_data = asnmap.launch(ip.address, type="ip") - + # Use asnmap tool to get ASN info, passing the API key + asn_data = asnmap.launch(ip.address, type="ip", api_key=api_key) if asn_data and "as_number" in asn_data: + # Parse ASN number from string like "AS16276" to integer 16276 + asn_string = asn_data["as_number"] + asn_number = int(asn_string.replace("AS", "").replace("as", "")) + # Create ASN object with correct field mapping asn = ASN( - asn=str(asn_data["as_number"]), + number=asn_number, name=asn_data.get("as_name", ""), - org=asn_data.get("as_org", ""), country=asn_data.get("as_country", ""), + description=asn_data.get("as_name", ""), ) results.append(asn) - + Logger.info( + self.sketch_id, + { + "message": f"[ASNMAP] Found AS{asn.number} ({asn.name}) for IP {ip.address}" + }, + ) + else: + Logger.warn( + self.sketch_id, + { + "message": f"[ASNMAP] No ASN data or missing 'as_number' field for IP {ip.address}. Data keys: {list(asn_data.keys()) if asn_data else 'None'}" + }, + ) except Exception as e: Logger.error( self.sketch_id, @@ -77,22 +130,21 @@ class IpToAsnTransform(Transform): if input_data and self.neo4j_conn: for ip, asn in zip(input_data, results): # Create IP node - self.create_node( - "ip", "address", ip.address, caption=ip.address, type="ip" - ) - + self.create_node("ip", "address", ip.address, label=ip.address, type="ip", **ip.__dict__) # Create ASN node - self.create_node( - "asn", "network", asn.asn, caption=f"AS{asn.asn}", type="asn" - ) - + self.create_node("asn", "number", asn.number, label=f"AS{asn.number}", type="asn", **asn.__dict__) # Create relationship self.create_relationship( - "ip", "address", ip.address, "asn", "network", asn.asn, "BELONGS_TO" + "ip", + "address", + ip.address, + "asn", + "number", + asn.number, + "BELONGS_TO", ) - self.log_graph_message( - f"IP {ip.address} belongs to AS{asn.asn} ({asn.name})" + f"IP {ip.address} belongs to AS{asn.number} ({asn.name})" ) return results diff --git a/flowsint-transforms/src/tools/dockertool.py b/flowsint-transforms/src/tools/dockertool.py index a696fcec..bf906668 100644 --- a/flowsint-transforms/src/tools/dockertool.py +++ b/flowsint-transforms/src/tools/dockertool.py @@ -46,8 +46,13 @@ class DockerTool(Tool): except ImageNotFound: return False - def launch(self, command: str, volumes: dict = None, timeout: int = 30): + def launch(self, command: str, volumes: dict = None, timeout: int = 30, environment: dict = None): self.install() + # Merge default environment with custom environment + env = {"TERM": "dumb"} # Set terminal type to avoid TTY issues + if environment: + env.update(environment) + try: result = self.client.containers.run( self.image, @@ -60,7 +65,7 @@ class DockerTool(Tool): tty=False, network_mode="bridge", stdin_open=False, # Ensure stdin is not open - environment={"TERM": "dumb"}, # Set terminal type to avoid TTY issues + environment=env, ) return result.decode() except ImageNotFound: @@ -89,7 +94,7 @@ class DockerTool(Tool): tty=False, network_mode="bridge", stdin_open=False, - environment={"TERM": "dumb"}, + environment=env, ) # If we get here, the command actually worked return test_result.decode() diff --git a/flowsint-transforms/src/tools/network/asnmap.py b/flowsint-transforms/src/tools/network/asnmap.py index 757445c2..df581273 100644 --- a/flowsint-transforms/src/tools/network/asnmap.py +++ b/flowsint-transforms/src/tools/network/asnmap.py @@ -51,7 +51,7 @@ class AsnmapTool(DockerTool): return super().is_installed() def launch( - self, item: str, type: Literal["domain", "organization", "ip", "asn"] = "domain" + self, item: str, type: Literal["domain", "organization", "ip", "asn"] = "domain", api_key: str = None ) -> Any: flags = {"domain": "-d", "org": "-org", "ip": "-i", "asn": "-a"} if type not in flags: @@ -59,11 +59,48 @@ class AsnmapTool(DockerTool): f"Invalid type: '{type}'. Valid types are: {list(flags.keys())}" ) flag = flags[type] + + # Prepare environment variables + env = {} + if api_key: + env["PDCP_API_KEY"] = api_key + try: # Use the -target argument as asnmap expects - result = super().launch(f"{flag} {item} -silent -json") + result = super().launch(f"{flag} {item} -silent -json", environment=env) if result and result != "": - return json.loads(result) + # asnmap returns newline-delimited JSON (one JSON object per line) + lines = result.strip().split('\n') + + if len(lines) == 1: + # Single JSON object + return json.loads(lines[0]) + else: + # Multiple JSON objects - combine them + combined_data = { + "as_range": [], + "as_name": None, + "as_country": None, + "as_number": None, + } + + for line in lines: + if not line.strip(): + continue + try: + data = json.loads(line) + if "as_range" in data: + combined_data["as_range"].extend(data["as_range"]) + if data.get("as_name") and not combined_data["as_name"]: + combined_data["as_name"] = data["as_name"] + if data.get("as_country") and not combined_data["as_country"]: + combined_data["as_country"] = data["as_country"] + if data.get("as_number") and not combined_data["as_number"]: + combined_data["as_number"] = data["as_number"] + except json.JSONDecodeError: + continue + + return combined_data if combined_data["as_range"] or combined_data["as_number"] else {} else: return {} except json.JSONDecodeError as e: diff --git a/flowsint-transforms/src/tools/network/dnsx.py b/flowsint-transforms/src/tools/network/dnsx.py new file mode 100644 index 00000000..d493db02 --- /dev/null +++ b/flowsint-transforms/src/tools/network/dnsx.py @@ -0,0 +1,88 @@ +import json +from typing import Any, List +from ..dockertool import DockerTool + + +class DnsxTool(DockerTool): + image = "projectdiscovery/dnsx" + default_tag = "latest" + + def __init__(self): + super().__init__(self.image, self.default_tag) + + @classmethod + def name(cls) -> str: + return "dnsx" + + @classmethod + def description(cls) -> str: + return "Fast and multi-purpose DNS toolkit for running various DNS queries." + + @classmethod + def category(cls) -> str: + return "DNS resolution" + + def install(self) -> None: + super().install() + + def version(self) -> str: + try: + output = self.client.containers.run( + image=self.image, + command="--version", + remove=True, + stderr=True, + stdout=True, + ) + output_str = output.decode() + import re + + match = re.search(r"(v[\d\.]+)", output_str) + version = match.group(1) if match else "unknown" + return version + except Exception as e: + return f"unknown (error: {str(e)})" + + def update(self) -> None: + # Pull the latest image + self.install() + + def is_installed(self) -> bool: + return super().is_installed() + + def launch(self, cidr: str, ptr: bool = False, api_key: str = None) -> List[str]: + """ + Run dnsx to resolve IPs from CIDR. + + Args: + cidr: CIDR block to resolve (e.g., "192.168.1.0/24") + ptr: Whether to include PTR records + api_key: Optional ProjectDiscovery Cloud Platform API key + + Returns: + List of IP addresses + """ + # Build command + flags = ["-silent"] + if ptr: + flags.append("-ptr") + + command = f"-cidr {cidr} {' '.join(flags)}" + + # Prepare environment variables + env = {} + if api_key: + env["PDCP_API_KEY"] = api_key + + try: + result = super().launch(command, environment=env) + if result and result.strip(): + # Split by newlines and filter out empty lines + ips = [line.strip() for line in result.split("\n") if line.strip()] + return ips + else: + return [] + except Exception as e: + raise RuntimeError( + f"Error running dnsx: {str(e)}. Output: {getattr(e, 'output', 'No output')}" + ) diff --git a/flowsint-transforms/src/tools/network/mapcidr.py b/flowsint-transforms/src/tools/network/mapcidr.py new file mode 100644 index 00000000..e019edce --- /dev/null +++ b/flowsint-transforms/src/tools/network/mapcidr.py @@ -0,0 +1,109 @@ +from typing import List +from ..dockertool import DockerTool + + +class MapcidrTool(DockerTool): + image = "projectdiscovery/mapcidr" + default_tag = "latest" + + def __init__(self): + super().__init__(self.image, self.default_tag) + + @classmethod + def name(cls) -> str: + return "mapcidr" + + @classmethod + def description(cls) -> str: + return "Small utility program to perform multiple operations for a given subnet/CIDR ranges." + + @classmethod + def category(cls) -> str: + return "Network utilities" + + def install(self) -> None: + super().install() + + def version(self) -> str: + try: + output = self.client.containers.run( + image=self.image, + command="--version", + remove=True, + stderr=True, + stdout=True, + ) + output_str = output.decode() + import re + + match = re.search(r"(v[\d\.]+)", output_str) + version = match.group(1) if match else "unknown" + return version + except Exception as e: + return f"unknown (error: {str(e)})" + + def update(self) -> None: + # Pull the latest image + self.install() + + def is_installed(self) -> bool: + return super().is_installed() + + def launch( + self, + cidr: str, + slice_by: int = None, + aggregate: bool = False, + shuffle_ips: bool = False, + shuffle_ports: bool = False, + count: bool = False, + api_key: str = None, + ) -> List[str]: + """ + Run mapcidr to expand CIDR ranges into IPs. + + Args: + cidr: CIDR block to expand (e.g., "192.168.1.0/24") + slice_by: Slice CIDR by number of hosts + aggregate: Aggregate IPs/CIDRs into smallest subnet possible + shuffle_ips: Shuffle IPs in output + shuffle_ports: Shuffle ports in output + count: Count number of IPs in CIDR + api_key: Optional ProjectDiscovery Cloud Platform API key + + Returns: + List of IP addresses + """ + # Build command + flags = ["-silent"] + + if slice_by: + flags.append(f"-slice-by {slice_by}") + if aggregate: + flags.append("-aggregate") + if shuffle_ips: + flags.append("-shuffle-ip") + if shuffle_ports: + flags.append("-shuffle-port") + if count: + flags.append("-count") + + command = f"-cidr {cidr} {' '.join(flags)}" + + # Prepare environment variables + env = {} + if api_key: + env["PDCP_API_KEY"] = api_key + + try: + result = super().launch(command, environment=env) + if result and result.strip(): + # Split by newlines and filter out empty lines + ips = [line.strip() for line in result.split("\n") if line.strip()] + return ips + else: + return [] + except Exception as e: + raise RuntimeError( + f"Error running mapcidr: {str(e)}. Output: {getattr(e, 'output', 'No output')}" + )