DEV Community: vysh

Building a High-Concurrency OSINT Engine in Rust: How I Managed 35+ Async Streams Without Exhausting Sockets

vysh — Sat, 25 Jul 2026 02:15:29 +0000

Description:

For the past few months, I have been building Reconx, an open-source CLI tool designed for mapping external infrastructure and gathering threat intelligence. If you have ever done penetration testing or bug bounty hunting, you know the standard workflow: run five different Go or Python tools, pipe the outputs together, and write bash scripts to parse the mess. I wanted a single, unified engine that could handle dozens of sources concurrently, track state changes, and output clean data. Here is a look under the hood at how I built the network and concurrency architecture in Rust.

The Problem: Socket Exhaustion in Async Rust

When I first started building Reconx, I made a classic mistake. I had over 35 OSINT collectors (querying Shodan, Censys, VirusTotal, etc.), and every time a collector fired, it instantiated a new HTTP client.

When you run that at scale using tokio, you immediately run into ephemeral port exhaustion and get slammed by API rate limits. Target servers drop connections, and the OS runs out of file descriptors.

The Solution: Centralized Connection Pooling

To fix this, I completely refactored how Reconx handles outbound traffic. Instead of letting collectors manage their own network state, I built a centralized HTTP engine inside src/http.rs. I created a single, shared reqwest::Client wrapped in an Arc (Atomic Reference Count). Every single collector now routes its requests through this unified pool.

/// Build a centralized reqwest client with proxy rotation and custom timeouts.
/// Every collector shares this unified engine.
pub fn build_client(timeout_secs: u64, proxy_url: Option<&str>) -> reqwest::Result<reqwest::Client> {
    let mut builder = reqwest::Client::builder()
        .timeout(Duration::from_secs(timeout_secs))
        .user_agent("Mozilla/5.0 (compatible; Reconx/0.1; +https://github.com/redshadow912/Reconx)");

    // Wire up HTTP/SOCKS5 proxy if the user passed the --proxy flag
    if let Some(proxy) = proxy_url {
        if !proxy.is_empty() {
            builder = builder.proxy(Proxy::all(proxy)?);
        }
    }

    builder.build()
}

/// Execute an HTTP request with built-in rate-limiting and exponential backoff
/// Automatically handles HTTP 429 (Too Many Requests) and 5xx Server Errors.
pub async fn request_with_retry(
    client: &reqwest::Client,
    url: &str,
    max_retries: u32,
    rate_limiter: Option<&ApiRateLimiter>,
) -> Result<reqwest::Response, reqwest::Error> {
    let mut retries = 0;

    loop {
        // Enforce per-API quotas (e.g. max 5 requests/sec for Shodan)
        if let Some(limiter) = rate_limiter {
            limiter.until_ready().await;
        }

        let response = client.get(url).send().await?;
        let status = response.status();

        // If successful, or if it's a hard client error (like 404), return immediately
        if status.is_success() || (status.is_client_error() && status.as_u16() != 429) {
            return Ok(response);
        }

        // If rate limited (429) or server error (5xx), apply exponential backoff
        if retries >= max_retries {
            return Ok(response);
        }

        // Wait 500ms -> 1s -> 2s -> 4s before retrying
        let backoff = Duration::from_millis(500 * 2u64.pow(retries));
        tokio::time::sleep(backoff).await;
        retries += 1;
    }
}

This change brought massive benefits:

TCP Connection Reuse: reqwest handles connection pooling under the hood, meaning subsequent requests to the same API reuse the existing socket.

Global Proxy Routing: By passing proxy configurations (HTTP/SOCKS5) into this single client, every single OSINT module instantly gained proxy support without needing to touch their individual code files.

Managing State and Asset Diffing

One of my biggest frustrations with existing tools is running a scan a week later and having to manually figure out what changed.
To solve this, I built an in-memory diffing engine (src/analyzers/diff_engine.rs). It ingests the previous scan's state and compares it against the live run.

use std::collections::HashSet;
use crate::models::Finding;

pub struct DiffEngine;

impl DiffEngine {
    /// Compare a live scan against the previous historical state 
    /// to extract actionable intelligence (new assets, removed assets, new vulns).
    pub fn diff(previous: &[Finding], current: &[Finding]) -> DiffResult {
        // 1. Transform arrays into HashSets for O(1) lookups
        let prev_subdomains: HashSet<String> = previous.iter().filter_map(|f| {
            if let Finding::Subdomain(s) = f { Some(s.subdomain.clone()) } else { None }
        }).collect();

        let curr_subdomains: HashSet<String> = current.iter().filter_map(|f| {
            if let Finding::Subdomain(s) = f { Some(s.subdomain.clone()) } else { None }
        }).collect();

        let prev_vulns: HashSet<String> = previous.iter().filter_map(|f| {
            if let Finding::Vulnerability(v) = f { 
                Some(format!("{}:{}", v.host, v.cve_id.as_deref().unwrap_or(&v.vulnerability_type))) 
            } else { None }
        }).collect();

        let curr_vulns: HashSet<String> = current.iter().filter_map(|f| {
            if let Finding::Vulnerability(v) = f { 
                Some(format!("{}:{}", v.host, v.cve_id.as_deref().unwrap_or(&v.vulnerability_type))) 
            } else { None }
        }).collect();

        // 2. Compute exact diffs instantly using set mathematics
        let new_subdomains: Vec<String> = curr_subdomains.difference(&prev_subdomains).cloned().collect();
        let removed_subdomains: Vec<String> = prev_subdomains.difference(&curr_subdomains).cloned().collect();
        let new_vulnerabilities: Vec<String> = curr_vulns.difference(&prev_vulns).cloned().collect();

        let total_new = new_subdomains.len() + new_vulnerabilities.len();
        let total_removed = removed_subdomains.len();

        DiffResult {
            new_subdomains,
            removed_subdomains,
            new_vulnerabilities,
            total_new,
            total_removed,
            // ... (other fields omitted for brevity)
        }
    }
}

This engine feeds directly into the takeover_detector.rs and risk_scorer.rs modules, meaning Reconx doesn't just give you a list of subdomains—it tells you exactly what is new and what is vulnerable right now.

Takeaways

Writing a highly concurrent network tool in Rust forces you to think deeply about resource management. Moving from ad-hoc HTTP requests to a centralized, reference-counted connection pool completely stabilized the tool under heavy load.

The project is entirely open-source, and you can check out the full architecture here:
Repository: https://github.com/redshadow912/Reconx

If you are a Rust developer, I would love for you to poke around the codebase. I am particularly interested in feedback on the async stream handling and error mapping.

How I built a runtime "flight recorder" for Node.js AI agents using AsyncLocalStorage

vysh — Wed, 15 Apr 2026 11:21:46 +0000

Introduction

I’ve been building a lot of autonomous AI agents lately. It’s incredibly fun, until you realize a terrifying fact about the current ecosystem: standard AI SDKs offer a false sense of security.

Frameworks like LangChain or AutoGen are phenomenal orchestrators. They let you define explicit tools (like a calculator or databaseSearch) and wrap them in neat execution contexts. But what happens if your agent gets prompt-injected and decides to bypass your tools entirely? What if the LLM hallucination just figures out how to write JavaScript that calls require('node:fs').readFileSync('.env') directly?

Nothing stops it. It's not a bug in the SDK; it's a gap in Node.js itself.

I know the purist answer: "Just migrate to Deno or Bun, they have native --allow-read permissions!" And they are right. If you control your runtime from scratch, you should use them. But for the 90% of us stuck maintaining existing Node.js monorepos, a massive migration isn't an option. We need a pragmatic seatbelt.

So, I built one. Here is how I used AsyncLocalStorage and runtime monkey-patching to build an open-source flight recorder for AI agents.

The Architecture: APM for AI Agents

I realized the problem wasn't exactly new. Companies like Datadog and New Relic have been tracking deeply nested asynchronous executions for years using Application Performance Monitoring (APM). I just needed to apply that exact same architecture to an LLM execution loop.

I broke the problem down into two parts:

Context Isolation: How do I know which agent made the file system call?

Global Interception: How do I actually catch and block the raw Node.js system calls without breaking the rest of the application?

Context Isolation with AsyncLocalStorage

If you haven't used AsyncLocalStorage (ALS) from the node:async_hooks module, it is essentially thread-local storage for asynchronous operations.

When you start an agent run, you wrap it in an ALS context and pass it a "Policy Engine" and a "Receipt." Any function called downstream no matter how many promises it chains through can access that context.

Here’s a simplified sketch of what ReceiptBot does internally (the library hides the AsyncLocalStorage store behind runWithInterceptors).

import { AsyncLocalStorage } from 'node:async_hooks';
import type { PolicyEngine, Receipt } from '@receiptbot/core';

// This holds the state for the current async execution tree
export const context = new AsyncLocalStorage<{ policy: PolicyEngine; receipt: Receipt }>();

// Simplified internal sketch (ReceiptBot’s public API exposes runWithInterceptors, not the ALS store)
export async function runWithInterceptors(policy: PolicyEngine, receipt: Receipt, agentFn: () => Promise<any>) {
  // (Global monkey-patches are applied here)
  return context.run({ policy, receipt }, async () => {
    return await agentFn();
  });
}

Now, even if a rogue dependency nested five layers deep tries to read a file, ReceiptBot can look up the current ALS store and know which policy applies.

Runtime Monkey-Patching Node Core

To stop the agent from reading secrets or making rogue network requests, I needed a global interceptor. Using module.createRequire, the tool monkey-patches Node's core modules (fs, http, child_process, net, tls) at runtime.

During initialization, it replaces the original functions with wrappers. Here is a simplified look at how the fs.readFileSync patch works:

import { PolicyViolationError } from '@receiptbot/core';
const originalReadFileSync = fs.readFileSync;

fs.readFileSync = function (...args) {
  const ctx = context.getStore(); // Check if we are inside an agent run

  if (ctx) {
    // ReceiptBot records the attempt FIRST; policy evaluation happens inside addEvent()
    const event = ctx.receipt.addEvent({
      type: 'tool.fs',
      action: `fs.readFileSync("${String(args[0])}")`,
      payload: { op: 'readFile', path: String(args[0]) },
    });

    // If the policy engine flagged it, kill the execution
    if (event.status === 'BLOCKED_BY_POLICY') {
      throw new PolicyViolationError('tool.fs', event.action, event.policyTrigger ?? 'Policy violation');
    }
  }

  // Execute the original function if allowed
  return originalReadFileSync.apply(this, args);
};

Hard-Stops, Cost Caps, and Redaction

Security isn't just about file access; it's about your API budget. A common failure mode for autonomous agents is getting stuck in a while(true) loop of hallucination, racking up a massive OpenAI API bill overnight.

While the network interceptor (http/fetch/net) is great for enforcing URL domain blocklists, calculating tokens natively at the network layer is messy. Instead, the Policy Engine allows you to enforce a hard budget cap:

const policy = new PolicyEngine()
  .denyPathGlobs(['**/.env'])
  .maxCost(1.00); // Hard stop at $1.00

When an LLM API call happens (either via a framework adapter or manually emitted as an llm.call event), it includes a costImpactUsd property. The Policy Engine validates the running total on every one of these events. The moment the next call would push the total over $1.00, it throws a PolicyViolationError and kills the execution path.

Finally, before any logs are written to the JSON receipt, the tool runs a redaction pass. It uses regex patterns to catch AWS keys, Stripe tokens, and OpenAI keys, replacing them with labeled markers like [REDACTED_OPENAI_API_KEY] so your audit logs don't become a new security vulnerability.

The Result: ReceiptBot

I packaged this architecture into an open-source tool called ReceiptBot.

It requires zero external infrastructure. It just sits quietly in your Node codebase, intercepts rogue system calls, and spits out a highly detailed, redacted JSON "receipt" of exactly what the agent did.

redshadow912 / ReceiptBot

🧾 ReceiptBot

A Flight Recorder and Seatbelt for Node.js AI Agents.

Monkey-patching isn't a hard OS sandbox — ReceiptBot is not trying to be one. It's your in-process flight recorder: a structured audit trail of every I/O operation, a cost governor that cuts off runaway LLM loops, and a secret scrubber that runs before any log is written. All of it drops into your existing Node.js project in one function call.

View on GitHub · Quickstart · Architecture · Full API Reference

What is ReceiptBot?

ReceiptBot is a runtime governance library for Node.js that wraps your AI agent's async execution context with:

A Policy Engine — rules you define that block dangerous operations before they happen
A Flight Recorder — an immutable, structured audit trail (a "receipt") of every action taken
A Global Interceptor — monkey-patches raw Node.js core modules so even rogue third-party library calls are caught

It does…

View on GitHub

I want to be fully transparent: this is not a perfect OS-level sandbox like a V8 Isolate. There are always edge cases with monkey-patching in JavaScript. But it covers the most common, dangerous escape hatches (direct node:fs, http, child_process, fetch) within the same process.

If you are building with LangChain, AutoGen, or just raw LLM calls in Node.js, and you want a pragmatic "seatbelt" to keep your .env files safe and your budget capped, I’d love for you to check it out.

I would love any brutal architectural feedback you have!