How to Integrate External Tools Like DNSx as Flowsint Enrichers

To integrate external tools like DNSx as Flowsint enrichers, create a DockerTool subclass that wraps the DNSx container, build an Enricher subclass that invokes the tool and maps JSON output to Flowsint core types, and place both files under the flowsint-enrichers/src/tools/ and flowsint-enrichers/src/enrichers/ directories so the registry discovers them automatically.

Flowsint, maintained in the reconurge/flowsint repository, separates low-level command execution from high-level graph enrichment to keep reconnaissance pipelines modular. Understanding how to integrate external tools as Flowsint enrichers lets you bring any Dockerized utility—such as the ProjectDiscovery DNSx resolver—into the platform as a first-class component.

Understand the Tool and Enricher Separation

The architecture is split into two distinct layers. The tool layer handles the actual execution of an external program, while the enricher layer orchestrates that execution and translates raw results into persisted graph nodes.

• Tool base class — Defined in flowsint-enrichers/src/tools/base.py, this abstract class establishes the metadata contract and launch signature every tool must implement.
• DockerTool helper — Located in flowsint-enrichers/src/tools/dockertool.py, it provides reusable Docker lifecycle methods including is_installed, install, and a generic launch that runs the container and captures stdout.
• Enricher base class — Imports the tool, iterates over source entities (such as Domain), and converts output dictionaries into strongly typed models from flowsint_core.types.
• Registry — The core registry at flowsint-core/src/flowsint_core/core/services/registry.py auto-discovers enrichers placed under flowsint-enrichers/src/enrichers/, making them available to flow definitions without manual wiring.

Step 1: Create a DockerTool Subclass for DNSx

The first concrete task is a tool wrapper that knows how to pull the projectdiscovery/dnsx image, build the command line, and parse newline-delimited JSON. Save this implementation as flowsint-enrichers/src/tools/network/dnsx.py.

# src/tools/network/dnsx.py

from tools.dockertool import DockerTool
from typing import List, Optional, Any
import json


class DnsxTool(DockerTool):
    """Wrapper for the DNSx Docker image (projectdiscovery/dnsx)."""

    image = "projectdiscovery/dnsx"
    default_tag = "latest"

    @classmethod
    def name(cls) -> str:
        return "dnsx"

    @classmethod
    def category(cls) -> str:
        return "Network"

    @classmethod
    def description(cls) -> str:
        return "Fast DNS resolver/subdomain enumeration (ProjectDiscovery DNSx)."

    @classmethod
    def version(cls) -> str:
        return "2.2.0"

    def launch(
        self,
        target: str,
        json_output: bool = True,
        resolver: Optional[str] = None,
        timeout: int = 30,
    ) -> List[dict]:
        """
        Execute DNSx against ``target`` and return parsed JSON records.

        Args:
            target: Domain/IP or a file containing a list of targets.
            json_output: Ask DNSx to emit newline‑delimited JSON.
            resolver: Custom DNS resolver (e.g. ``8.8.8.8:53``).
            timeout: Maximum container runtime in seconds.

        Returns:
            List of dictionaries, each representing a DNS record.
        """
        # Ensure the image is available

        if not self.is_installed():
            self.install()

        # Build the command line

        cmd = f"-d {target}"
        if json_output:
            cmd += " -json"
        if resolver:
            cmd += f" -r {resolver}"

        # Run the container

        raw = super().launch(command=cmd, timeout=timeout)

        # Parse newline‑delimited JSON

        results: List[dict] = []
        for line in raw.strip().split("\n"):
            if not line:
                continue
            try:
                results.append(json.loads(line))
            except json.JSONDecodeError:
                # Skip malformed lines – tool is defensive

                continue
        return results

Key implementation details:

• DockerTool lifecycle — The wrapper inherits container pull, run, and cleanup behavior, so is_installed() and install() work without extra code.
• Command builder pattern — The cmd string is assembled conditionally based on method arguments, mirroring the pattern documented in docs/developers/managing-tools.mdx.
• Output parser pattern — Raw stdout is split into lines and each line is parsed individually, making the tool resilient to partial or malformed output.

Step 2: Build the Enricher to Map DNSx Output to Graph Entities

Next, create an enricher that consumes the tool and converts DNS records into Flowsint core types. Save this file as flowsint-enrichers/src/enrichers/network/dnsx_enricher.py.

# src/enrichers/network/dnsx_enricher.py

from enrichers.base import Enricher
from tools.network.dnsx import DnsxTool
from flowsint_core.types import Domain, Ip, Cname, DnsRecord  # Pydantic models

from typing import List


class DnsxEnricher(Enricher):
    """Enricher that resolves a domain (or a list) via DNSx."""

    async def scan(self, domains: List[Domain]) -> List[DnsRecord]:
        tool = DnsxTool()
        records: List[DnsRecord] = []

        for domain in domains:
            raw = tool.launch(target=domain.name, json_output=True)

            for entry in raw:
                # DNSx returns a dict with keys like "type", "name", "value"

                rec_type = entry.get("type")
                name = entry.get("name")
                value = entry.get("value")

                if rec_type == "A":
                    records.append(Ip(address=value, source=domain))
                elif rec_type == "CNAME":
                    records.append(Cname(name=name, target=value, source=domain))
                else:
                    # Generic record type – store as DnsRecord for later enrichment

                    records.append(
                        DnsRecord(
                            type=rec_type,
                            name=name,
                            value=value,
                            source=domain,
                        )
                    )
        return records

What happens during enrichment:

1. The scan method receives a list of Domain objects.
2. For each domain, it calls DnsxTool.launch with json_output=True.
3. Returned JSON rows are mapped onto Flowsint core types (Ip, Cname, DnsRecord) according to the DNS record type.
4. The framework automatically persists those objects and creates graph edges—for example, DOMAIN → RESOLVES_TO → IP—in Neo4j via the core service API.

Step 3: Register the Enricher and Reference It in a Flow

Flowsint uses entry-point style discovery for enrichers. If you place dnsx_enricher.py under flowsint-enrichers/src/enrichers/network/, the registry picks it up automatically at startup. For explicit control, you can add it to the ENRICHERS mapping in flowsint-enrichers/src/enrichers/__init__.py:

from .network.dnsx_enricher import DnsxEnricher

ENRICHERS = {
    "dnsx": DnsxEnricher,
    # …other enrichers…

}

Once registered, users can reference the enricher by name inside a flow definition:

steps:
  - name: dnsx
    type: enricher
    inputs:
      - domain: mytarget.com

The engine resolves "dnsx" through the registry, executes the DnsxEnricher.scan coroutine, and merges the results into the active investigation graph.

Summary

• Tool layer — Subclass DockerTool (from flowsint-enrichers/src/tools/dockertool.py) to wrap external Docker images, implement launch, and parse raw output.
• Enricher layer — Subclass Enricher (from enrichers.base), invoke your tool per source entity, and map JSON records to Flowsint types such as Ip, Cname, and DnsRecord.
• Registration — Drop files under flowsint-enrichers/src/enrichers/ for auto-discovery, or map them explicitly in the package __init__.py.
• Flow usage — Reference the enricher by its registered name in YAML flow definitions; the engine handles scheduling and Neo4j persistence automatically.

Frequently Asked Questions

What is the difference between a Tool and an Enricher in Flowsint?

A Tool is a low-level wrapper that knows how to execute an external binary or container and return raw data. An Enricher is a high-level orchestrator that accepts source graph entities, calls one or more tools, validates the output, and converts it into typed nodes that Flowsint persists in Neo4j.

Where does the registry discover new enrichers?

The registry defined in flowsint-core/src/flowsint_core/core/services/registry.py walks the flowsint-enrichers/src/enrichers/ package at startup. Any Enricher subclass placed inside that tree is loaded automatically via entry-point discovery.

How does Flowsint persist enricher output?

When an enricher returns typed objects—such as Ip or Cname—the core service API automatically writes them to Neo4j and creates appropriate graph edges. You do not need to write Cypher queries inside the enricher; the framework handles persistence once the typed models are returned from scan.

Can the DockerTool pattern be reused for other command-line utilities?

Yes. The DockerTool base class in flowsint-enrichers/src/tools/dockertool.py is generic. You can subclass it for any utility distributed as a Docker image—such as other ProjectDiscovery tools—by setting the image and default_tag class attributes and implementing a launch method that builds the tool-specific command line and parses its output.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}