flowsint

Flowsint is a modular OSINT (Open Source Intelligence) investigation and reconnaissance platform. It is designed to provide a stable infrastructure for intelligence gathering, allowing tools to be integrated as pluggable extensions rather than siloed scripts.

Key features include:

• Graph-based visualization: View entity relationships visually.
• Automated Enrichers: Access to 30+ automated enrichers for intelligence gathering.
• Modular Architecture: Clean separation of concerns for extensibility.
• Privacy-first Design: Local data storage and local execution of enrichers.
• Automated Search Flows: Orchestrated investigation workflows.

Flowsint is organized into several autonomous modules:

• flowsint-core: Core utilities, orchestrator, vault, celery tasks, and base classes (handles DB connections, Auth, Logging, and Config).
• flowsint-types: Pydantic models and type definitions (Domain, IP, ASN, Individual, etc.).
• flowsint-enrichers: Enricher modules and scanning logic (Domain, IP, Social, Crypto, etc.).
• flowsint-api: FastAPI server providing REST API endpoints, Auth, Graph integration, and real-time streaming.
• flowsint-app: Frontend application (UI).

Dependency Flow: flowsint-app $\rightarrow$ flowsint-api $\rightarrow$ flowsint-core $\rightarrow$ flowsint-enrichers $\rightarrow$ flowsint-types

In Flowsint, an Enricher is an operation that transforms a source entity (Input A) into one or more target entities (Output B) using a specific method called a Pivot.

The Pivot is the technical process or method used to derive the result (e.g., a DNS resolution, a WHOIS lookup, or an API query).

Conceptual Flow: Source Entity (A) -> Pivot (Method) -> Target Entity (B)

Examples:

• domain name $\rightarrow$ DNS resolution $\rightarrow$ IP address
• IP address $\rightarrow$ WHOIS Lookup $\rightarrow$ owner
• image $\rightarrow$ Reverse Image Search $\rightarrow$ web pages containing that image

The project uses @lukemorales/query-key-factory to manage TanStack Query keys. This provides type-safe query keys with auto-completion, centralized management, and easy invalidation without hardcoded strings.

Key files:

• src/api/query-keys.ts: Main query key definitions.
• src/api/query-keys-examples.ts: Usage examples and custom hooks.

To use the Flowsint platform effectively, it is important to understand the following domain-specific concepts:

• Investigation: A structured process of collecting and analyzing information. It can be exploratory (discovering new elements) or targeted (validating hypotheses). An investigation consists of multiple sketches and one or more analyses.
• Sketch: A visual representation of the current state of the investigation graph, produced by executing enrichers on entities.
• Analysis: The processing and interpretation of collected data to identify trends or confirm/refute hypotheses. These can be quantitative (statistical) or qualitative (contextual).
• Flow: The process of chaining multiple enrichers together, where the output of one enricher serves as the input for the next to expand an investigation.

The Flowsint type system is built on Pydantic models located in the flowsint-types package. Every data type must be a Python class that inherits from FlowsintType and must be decorated with @flowsint_type to be registered in the global type registry. This registration enables automatic validation, serialization, JSON schema generation, auto-discovery, and graph-specific features like automatic label generation.

Types are typically organized in flowsint-types/src/flowsint_types/, where each type (or group of closely related types) resides in its own Python file.

Flowsint is a powerful tool that requires responsible use. Users must adhere to the following guidelines:

Legal Responsibility

• You are solely responsible for your use of the software.
• Use the tool in compliance with all applicable laws and regulations.
• Obtain proper authorization before conducting any security testing or reconnaissance activities.

Infrastructure and Risk

• Local Execution: All enrichers run locally on your machine. This ensures confidentiality but places the burden of risk on the user.
• Service Bans: Rapid, large-scale requests (e.g., performing DNS resolution for 10,000 IPs) can lead to being banned from services or flagged by infrastructure providers.
• Scale Awareness: Be aware of the scale at which gathering can occur and understand the underlying mechanics of the tools used in the enrichers.

Enrichers are the high-level business logic layer in Flowsint. They orchestrate intelligence gathering workflows by taking input data, processing it through tools or APIs, and creating Neo4j graph nodes and relationships.

Unlike Tools (which are low-level wrappers for raw data), Enrichers are high-level workflows that understand types, handle parameters, and manage the graph database structure.

Every enricher follows a two-phase execution model:

1. Scanning Phase: An async method where core logic (API calls, tool execution, data gathering) occurs. Input is automatically validated via Pydantic.
2. Postprocessing Phase: A method where graph nodes and relationships are created in Neo4j based on the scan results and original input.

Replace hardcoded array keys in useQuery with the queryKeys object. For queries requiring parameters, call the corresponding key function with the required arguments.

import { useQuery } from '@tanstack/react-query'
import { queryKeys } from '@/api/query-keys'
import { investigationService } from '@/api/investigation-service'

// Basic query usage
const { data } = useQuery({
  queryKey: queryKeys.investigations.list,
  queryFn: investigationService.get,
})

// Dynamic Keys with Parameters
const { data: detailData } = useQuery({
  queryKey: queryKeys.investigations.detail(investigationId),
  queryFn: () => investigationService.getById(investigationId),
})

To test tools that require Docker, use the @pytest.mark.docker decorator. This allows you to separate tests that require a running Docker daemon from standard unit tests. Place your test files in flowsint-enrichers/tests/tools/ mirroring your tool's directory structure.

# tests/tools/network/test_my_subdomain_tool.py
from tools.network.my_subdomain_tool import MySubdomainTool
import pytest

@pytest.mark.docker
def test_tool_install():
    """Test that the Docker image can be pulled."""
    tool = MySubdomainTool()
    tool.install()
    assert tool.is_installed()

@pytest.mark.docker
def test_tool_launch():
    """Test running the tool against a domain."""
    tool = MySubdomainTool()
    results = tool.launch("example.com")
    assert isinstance(results, list)