github-starred/flowsint
2
0
Fork 0
mirror of https://github.com/reconurge/flowsint.git synced 2026-07-23 14:24:58 -05:00
Code Issues 980 Packages Projects Releases 14 Wiki Activity

fix(transforms): fixes usage of asnmap + other docker tools

Browse Source 
fix #31
This commit is contained in:
dextmorgn
2025-11-03 18:41:08 +01:00
parent 7ebcbc8acd
commit e574baad93
8 changed files with 603 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files
flowsint-transforms/src/flowsint_transforms/asn/to_cidrs.py
+92 -77
View File
@@ -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
flowsint-transforms/src/flowsint_transforms/cidr/to_ips.py
+86 -43
View File
@@ -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
flowsint-transforms/src/flowsint_transforms/domain/to_asn.py
+110 -25
View File
@@ -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
flowsint-transforms/src/flowsint_transforms/ip/to_asn.py
+70 -18
View File
@@ -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
flowsint-transforms/src/tools/dockertool.py
+8 -3
View File
@@ -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()
flowsint-transforms/src/tools/network/asnmap.py
+40 -3
View File
@@ -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:
flowsint-transforms/src/tools/network/dnsx.py
+88
View File
@@ -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')}"
)
flowsint-transforms/src/tools/network/mapcidr.py
+109
View File
@@ -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')}"
)