From 5ed9cfbc2a1a111085c447864ccb22a8f636064f Mon Sep 17 00:00:00 2001 From: Valmo Date: Mon, 29 Jun 2026 21:20:54 -0300 Subject: [PATCH] Added check on first frame feed --- src/config.rs | 261 +++++++++++++++++++- src/db.rs | 74 ++++++ src/main.rs | 10 +- src/redis_rtsp.rs | 607 ++++++++++++++++++++++++++++++++++++++++++++++ src/rtsp.rs | 376 ++++++++++++++++++++++++++++ 5 files changed, 1326 insertions(+), 2 deletions(-) create mode 100644 src/redis_rtsp.rs create mode 100644 src/rtsp.rs diff --git a/src/config.rs b/src/config.rs index 7ca91f4..b9a6bfa 100644 --- a/src/config.rs +++ b/src/config.rs @@ -10,12 +10,24 @@ pub enum Mode { Local, Coordinator, Worker, + Rtsp, + RedisCoordinator, + RedisWorker, } #[derive(Debug, Clone)] pub struct Config { pub mode: Mode, pub cidrs: Vec, + pub rtsp_targets: Vec, + pub rtsp_paths: Vec, + pub redis_url: String, + pub redis_queue_name: String, + pub redis_input_csv: String, + pub redis_worker_count: usize, + pub snapshot_dir: PathBuf, + pub sqlite_path: PathBuf, + pub capture_timeout_ms: u64, pub port: u16, pub timeout_ms: u64, pub concurrency: usize, @@ -33,9 +45,36 @@ impl Config { let cli = CliArgs::from_args(&args)?; let cidrs = load_cidrs(cli.cidr_file.as_deref(), cli.cidrs.as_deref())?; + let rtsp_targets = load_csv_targets(cli.targets_file.as_deref(), cli.targets.as_deref())?; + let rtsp_paths = load_rtsp_paths(cli.paths_file.as_deref(), cli.paths.as_deref())?; Ok(Self { mode, cidrs, + rtsp_targets, + rtsp_paths, + redis_url: cli + .redis_url + .unwrap_or_else(|| read_env("REDIS_URL", "redis://127.0.0.1:6379/0")), + redis_queue_name: cli + .redis_queue_name + .unwrap_or_else(|| read_env("RTSP_QUEUE_NAME", "camfinder:rtsp:queue")), + redis_input_csv: cli + .redis_input_csv + .unwrap_or_else(|| read_env("RTSP_INPUT_CSV", "camfinder-open-rtsp.csv")), + redis_worker_count: cli + .redis_worker_count + .or_else(|| parse_usize("RTSP_WORKER_COUNT", 2).ok()) + .unwrap_or(2), + snapshot_dir: cli + .snapshot_dir + .unwrap_or_else(|| PathBuf::from(read_env("RTSP_SNAPSHOT_DIR", "dump/feeds"))), + sqlite_path: cli + .sqlite_path + .unwrap_or_else(|| PathBuf::from(read_env("RTSP_SQLITE_PATH", "dump/rtsp-results.sqlite"))), + capture_timeout_ms: cli + .capture_timeout_ms + .or_else(|| parse_u64("RTSP_CAPTURE_TIMEOUT_MS", 15000).ok()) + .unwrap_or(15000), port: cli .port .or_else(|| parse_u16("SCAN_PORT", 554).ok()) @@ -69,6 +108,17 @@ impl Config { struct CliArgs { cidrs: Option, cidr_file: Option, + targets: Option, + targets_file: Option, + paths: Option, + paths_file: Option, + redis_url: Option, + redis_queue_name: Option, + redis_input_csv: Option, + redis_worker_count: Option, + snapshot_dir: Option, + sqlite_path: Option, + capture_timeout_ms: Option, database_url: Option, coordinator_bind: Option, coordinator_addr: Option, @@ -94,6 +144,54 @@ impl CliArgs { cli.cidr_file = Some(PathBuf::from(next_value(args, &mut index, "--cidr-file")?)); } + "--targets" => { + cli.targets = Some(next_value(args, &mut index, "--targets")?); + } + "--targets-file" => { + cli.targets_file = Some(PathBuf::from(next_value( + args, + &mut index, + "--targets-file", + )?)); + } + "--paths" => { + cli.paths = Some(next_value(args, &mut index, "--paths")?); + } + "--paths-file" => { + cli.paths_file = + Some(PathBuf::from(next_value(args, &mut index, "--paths-file")?)); + } + "--redis-url" => { + cli.redis_url = Some(next_value(args, &mut index, "--redis-url")?); + } + "--queue" | "--queue-name" => { + cli.redis_queue_name = Some(next_value(args, &mut index, arg)?); + } + "--input-csv" | "--csv-file" => { + cli.redis_input_csv = Some(next_value(args, &mut index, arg)?); + } + "--worker-count" => { + cli.redis_worker_count = Some(parse_usize_literal(&next_value( + args, + &mut index, + "--worker-count", + )?)?); + } + "--snapshot-dir" => { + cli.snapshot_dir = + Some(PathBuf::from(next_value(args, &mut index, "--snapshot-dir")?)); + } + "--sqlite-path" => { + cli.sqlite_path = + Some(PathBuf::from(next_value(args, &mut index, "--sqlite-path")?)); + } + "--capture-timeout-ms" => { + cli.capture_timeout_ms = Some(parse_u64_literal(&next_value( + args, + &mut index, + "--capture-timeout-ms", + )?)?); + } "--database" | "--database-url" => { cli.database_url = Some(next_value(args, &mut index, arg)?); } @@ -167,8 +265,11 @@ fn parse_mode_literal(value: &str) -> Result { "local" | "scan" => Ok(Mode::Local), "coordinator" | "coord" | "server" => Ok(Mode::Coordinator), "worker" => Ok(Mode::Worker), + "rtsp" => Ok(Mode::Rtsp), + "redis-coordinator" | "redis-coord" | "redis-server" => Ok(Mode::RedisCoordinator), + "redis-worker" | "redis-client" => Ok(Mode::RedisWorker), other => Err(anyhow!( - "modo inválido: {other}. Use local, coordinator ou worker" + "modo inválido: {other}. Use local, coordinator, worker, rtsp, redis-coordinator ou redis-worker" )), } } @@ -238,6 +339,164 @@ fn parse_cidr_list(input: &str) -> Vec { .collect() } +fn load_csv_targets(file: Option<&Path>, inline: Option<&str>) -> Result> { + if let Some(inline) = inline { + let targets = parse_target_list(inline); + if !targets.is_empty() { + return Ok(targets); + } + } + + if let Some(file) = file { + let content = fs::read_to_string(file) + .with_context(|| format!("failed to read CSV file {}", file.display()))?; + let targets = parse_csv_targets(&content); + if !targets.is_empty() { + return Ok(targets); + } + return Err(anyhow!("arquivo CSV vazio: {}", file.display())); + } + + if let Ok(path) = env::var("RTSP_TARGETS_FILE") { + let path = path.trim(); + if !path.is_empty() { + let content = fs::read_to_string(path) + .with_context(|| format!("failed to read CSV file {path}"))?; + let targets = parse_csv_targets(&content); + if !targets.is_empty() { + return Ok(targets); + } + return Err(anyhow!("arquivo CSV vazio: {path}")); + } + } + + if let Ok(value) = env::var("RTSP_TARGETS") { + let value = value.trim(); + if !value.is_empty() { + let targets = parse_target_list(value); + if !targets.is_empty() { + return Ok(targets); + } + } + } + + Ok(Vec::new()) +} + +fn parse_csv_targets(input: &str) -> Vec { + input + .lines() + .map(str::trim) + .filter(|line| !line.is_empty() && !line.starts_with('#')) + .filter_map(|line| line.split(',').next()) + .map(str::trim) + .map(strip_quotes) + .filter(|entry| !entry.is_empty()) + .filter(|entry| !is_csv_header(entry)) + .map(ToString::to_string) + .collect() +} + +fn parse_target_list(input: &str) -> Vec { + input + .lines() + .flat_map(|line| line.split(|c: char| c == ',' || c.is_whitespace())) + .map(str::trim) + .map(strip_quotes) + .filter(|entry| !entry.is_empty()) + .filter(|entry| !is_csv_header(entry)) + .map(ToString::to_string) + .collect() +} + +fn load_rtsp_paths(file: Option<&Path>, inline: Option<&str>) -> Result> { + if let Some(inline) = inline { + let paths = parse_rtsp_paths(inline); + if !paths.is_empty() { + return Ok(paths); + } + } + + if let Some(file) = file { + let content = fs::read_to_string(file) + .with_context(|| format!("failed to read RTSP path file {}", file.display()))?; + let paths = parse_rtsp_paths(&content); + if !paths.is_empty() { + return Ok(paths); + } + return Err(anyhow!("arquivo de paths RTSP vazio: {}", file.display())); + } + + if let Ok(path) = env::var("RTSP_PATHS_FILE") { + let path = path.trim(); + if !path.is_empty() { + return load_rtsp_paths_from_file(Path::new(path)); + } + } + + if let Ok(value) = env::var("RTSP_PATHS") { + let value = value.trim(); + if !value.is_empty() { + let paths = parse_rtsp_paths(value); + if !paths.is_empty() { + return Ok(paths); + } + } + } + + load_rtsp_paths_from_file(Path::new("rtsp_paths.txt")) +} + +fn load_rtsp_paths_from_file(file: &Path) -> Result> { + let content = fs::read_to_string(file) + .with_context(|| format!("failed to read RTSP path file {}", file.display()))?; + let paths = parse_rtsp_paths(&content); + if !paths.is_empty() { + return Ok(paths); + } + Err(anyhow!("arquivo de paths RTSP vazio: {}", file.display())) +} + +fn parse_rtsp_paths(input: &str) -> Vec { + let mut seen = std::collections::HashSet::new(); + input + .lines() + .flat_map(|line| line.split('#').next()) + .flat_map(|line| line.split(|c: char| c == ',' || c.is_whitespace())) + .map(str::trim) + .map(strip_quotes) + .filter(|entry| !entry.is_empty()) + .filter(|entry| entry.starts_with('/')) + .filter_map(|entry| { + let entry = entry.to_string(); + if seen.insert(entry.clone()) { + Some(entry) + } else { + None + } + }) + .collect() +} + +fn strip_quotes(value: &str) -> &str { + value + .strip_prefix('"') + .and_then(|value| value.strip_suffix('"')) + .or_else(|| { + value + .strip_prefix('\'') + .and_then(|value| value.strip_suffix('\'')) + }) + .unwrap_or(value) +} + +fn is_csv_header(value: &str) -> bool { + matches!( + value.to_ascii_lowercase().as_str(), + "ip" | "host" | "hostname" | "address" | "target" | "camera" + ) +} + fn database_url_from_env() -> String { let value = read_env("DATABASE_URL", ""); if value.is_empty() { diff --git a/src/db.rs b/src/db.rs index 9463c8c..6d5fd50 100644 --- a/src/db.rs +++ b/src/db.rs @@ -51,6 +51,24 @@ impl Database { created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); + + CREATE TABLE IF NOT EXISTS rtsp_probe_results ( + id BIGSERIAL PRIMARY KEY, + target_host TEXT NOT NULL, + target_port INTEGER NOT NULL, + rtsp_path TEXT NOT NULL, + rtsp_url TEXT NOT NULL UNIQUE, + response_code INTEGER, + status TEXT NOT NULL, + error TEXT, + first_seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + last_seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + + CREATE INDEX IF NOT EXISTS rtsp_probe_results_status_idx + ON rtsp_probe_results (status); "#, ) .await @@ -178,6 +196,62 @@ impl Database { i64_to_usize(count, "open endpoint count") } + pub async fn upsert_rtsp_probe_result( + &self, + target_host: &str, + target_port: u16, + rtsp_path: &str, + rtsp_url: &str, + response_code: Option, + status: &str, + error: Option<&str>, + ) -> Result<()> { + let target_port = i32::from(target_port); + let response_code = response_code.map(i32::from); + + self.client + .execute( + r#" + INSERT INTO rtsp_probe_results ( + target_host, + target_port, + rtsp_path, + rtsp_url, + response_code, + status, + error, + first_seen_at, + last_seen_at, + created_at, + updated_at + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, NOW(), NOW(), NOW(), NOW()) + ON CONFLICT (rtsp_url) DO UPDATE SET + target_host = EXCLUDED.target_host, + target_port = EXCLUDED.target_port, + rtsp_path = EXCLUDED.rtsp_path, + response_code = EXCLUDED.response_code, + status = EXCLUDED.status, + error = EXCLUDED.error, + last_seen_at = NOW(), + updated_at = NOW() + "#, + &[ + &target_host, + &target_port, + &rtsp_path, + &rtsp_url, + &response_code, + &status, + &error, + ], + ) + .await + .context("failed to upsert RTSP probe result")?; + + Ok(()) + } + pub async fn summarize_chunks(&self) -> Result<(usize, usize)> { let row = self .client diff --git a/src/main.rs b/src/main.rs index 5a56651..d4fbc3e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,6 +3,8 @@ mod coordinator; mod db; mod ip; mod models; +mod rtsp; +mod redis_rtsp; mod scan; mod terminal; mod worker; @@ -27,7 +29,7 @@ async fn main() -> Result<()> { cidrs: validate_and_filter_cidrs(&config.cidrs)?, ..config }, - Mode::Worker => config, + Mode::Worker | Mode::Rtsp | Mode::RedisCoordinator | Mode::RedisWorker => config, }; let ui = Arc::new(TerminalUi::new(stdout_is_terminal())); @@ -50,6 +52,12 @@ async fn main() -> Result<()> { let database = Arc::new(Database::open(&config.database_url).await?); coordinator::run_coordinator(config, database, ui, interrupted).await } + Mode::Rtsp => { + let database = Arc::new(Database::open(&config.database_url).await?); + rtsp::run_rtsp_scan(config, database, ui, interrupted).await + } + Mode::RedisCoordinator => redis_rtsp::run_redis_coordinator(config, ui, interrupted).await, + Mode::RedisWorker => redis_rtsp::run_redis_worker(config, ui, interrupted).await, Mode::Worker => worker::run_worker(config, ui).await, }; diff --git a/src/redis_rtsp.rs b/src/redis_rtsp.rs new file mode 100644 index 0000000..77e9968 --- /dev/null +++ b/src/redis_rtsp.rs @@ -0,0 +1,607 @@ +use crate::{ + config::Config, + terminal::{ScreenState, TerminalUi}, +}; +use anyhow::{anyhow, Context, Result}; +use std::{ + fs, + path::{Path, PathBuf}, + sync::atomic::{AtomicBool, Ordering}, + sync::Arc, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, +}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + process::Command, + time::{sleep, timeout}, +}; + +const DEFAULT_DONE_SUFFIX: &str = ":done"; +const DEFAULT_SENTINEL: &str = "__CAMFINDER_DONE__"; + +pub async fn run_redis_coordinator( + config: Config, + ui: Arc, + _interrupted: Arc, +) -> Result<()> { + let ips = load_ip_list(&config.redis_input_csv)?; + if ips.is_empty() { + return Err(anyhow!( + "nenhum IP encontrado em {}", + config.redis_input_csv + )); + } + + let done_key = format!("{}{}", config.redis_queue_name, DEFAULT_DONE_SUFFIX); + clear_queue(&config.redis_url, &config.redis_queue_name, &done_key).await?; + enqueue_ips( + &config.redis_url, + &config.redis_queue_name, + &ips, + config.redis_worker_count.max(1), + ) + .await?; + mark_done(&config.redis_url, &done_key).await?; + + if ui.is_enabled() { + ui.print_banner(); + ui.render(&ScreenState { + current_ip: format!("fila Redis: {}", config.redis_queue_name), + tested: ips.len(), + total: ips.len(), + }); + ui.finish(); + } else { + println!( + "Fila Redis alimentada com {} IPs em {}. Workers esperados: {}", + ips.len(), + config.redis_queue_name, + config.redis_worker_count.max(1) + ); + } + + Ok(()) +} + +pub async fn run_redis_worker( + config: Config, + _ui: Arc, + interrupted: Arc, +) -> Result<()> { + ensure_output_dirs(&config.snapshot_dir, &config.sqlite_path)?; + init_sqlite(&config.sqlite_path).await?; + + let done_key = format!("{}{}", config.redis_queue_name, DEFAULT_DONE_SUFFIX); + let started_at = Instant::now(); + let mut processed = 0usize; + let mut captured = 0usize; + let paths = normalize_paths(&config.rtsp_paths); + + println!( + "worker pronto para {} paths em {}", + paths.len(), + config.redis_queue_name + ); + + loop { + if interrupted.load(Ordering::SeqCst) { + break; + } + + match pop_queue_item(&config.redis_url, &config.redis_queue_name).await? { + Some(item) if item == DEFAULT_SENTINEL => break, + Some(ip) => { + for rtsp_path in &paths { + if interrupted.load(Ordering::SeqCst) { + break; + } + + let attempt = build_rtsp_url(&ip, config.port, &rtsp_path); + let checked_at = timestamp_millis(); + let probe = probe_rtsp(&attempt, config.capture_timeout_ms).await; + + let (status, screenshot_path, error) = if probe.is_connectable() { + let output_path = + build_snapshot_path(&config.snapshot_dir, &ip, &rtsp_path, checked_at); + let outcome = capture_feed(&attempt, &output_path, config.capture_timeout_ms) + .await; + + match outcome { + Ok(()) => { + captured += 1; + ( + "captured".to_string(), + Some(output_path.to_string_lossy().to_string()), + None, + ) + } + Err(err) => { + let _ = fs::remove_file(&output_path); + ("failed".to_string(), None, Some(err.to_string())) + } + } + } else { + ( + probe.status, + None, + probe.error.map(|error| error.to_string()), + ) + }; + + insert_sqlite_result( + &config.sqlite_path, + checked_at, + &ip, + &rtsp_path, + &attempt, + status, + screenshot_path.as_deref(), + error.as_deref(), + ) + .await?; + + processed += 1; + println!( + "{} -> {}", + attempt, + screenshot_path.unwrap_or_else(|| "sem imagem".to_string()) + ); + } + } + None => { + if redis_key_exists(&config.redis_url, &done_key).await? { + break; + } + sleep(Duration::from_secs(1)).await; + } + } + } + + let elapsed = started_at.elapsed().as_secs_f64(); + println!( + "worker finalizado em {:.2}s, tentativas {}, capturas {}", + elapsed, processed, captured + ); + + Ok(()) +} + +async fn capture_feed(url: &str, output_path: &Path, timeout_ms: u64) -> Result<()> { + let mut command = Command::new("ffmpeg"); + command + .kill_on_drop(true) + .arg("-hide_banner") + .arg("-loglevel") + .arg("error") + .arg("-fflags") + .arg("nobuffer") + .arg("-flags") + .arg("low_delay") + .arg("-probesize") + .arg("32") + .arg("-analyzeduration") + .arg("0") + .arg("-rtsp_transport") + .arg("tcp") + .arg("-i") + .arg(url) + .arg("-frames:v") + .arg("1") + .arg("-skip_frame") + .arg("nokey") + .arg("-an") + .arg("-sn") + .arg("-dn") + .arg("-y") + .arg(output_path.as_os_str()); + + let output = timeout( + Duration::from_millis(timeout_ms), + async { command.output().await }, + ) + .await + .map_err(|_| anyhow!("ffmpeg timeout for {url}"))??; + + if output.status.success() { + return Ok(()); + } + + let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string(); + if output_path.exists() { + let _ = fs::remove_file(output_path); + } + Err(anyhow!( + "ffmpeg failed for {url}: {}", + if stderr.is_empty() { + "sem stderr".to_string() + } else { + stderr + } + )) +} + +async fn probe_rtsp(url: &str, timeout_ms: u64) -> RtspProbe { + let host = extract_host(url); + let connect_result = timeout( + Duration::from_millis(timeout_ms), + tokio::net::TcpStream::connect(host), + ) + .await; + + let mut stream = match connect_result { + Ok(Ok(stream)) => stream, + Ok(Err(err)) => { + return RtspProbe { + status: "connect_error".to_string(), + error: Some(err.to_string()), + }; + } + Err(_) => { + return RtspProbe { + status: "timeout".to_string(), + error: Some("connect timeout".to_string()), + }; + } + }; + + let request = format!( + "DESCRIBE {url} RTSP/1.0\r\nCSeq: 1\r\nUser-Agent: CamFinder/0.1\r\nAccept: application/sdp\r\n\r\n" + ); + match timeout( + Duration::from_millis(timeout_ms), + stream.write_all(request.as_bytes()), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(err)) => { + return RtspProbe { + status: "write_error".to_string(), + error: Some(err.to_string()), + }; + } + Err(_) => { + return RtspProbe { + status: "timeout".to_string(), + error: Some("write timeout".to_string()), + }; + } + } + + let mut buf = vec![0u8; 1024]; + let read_result = timeout( + Duration::from_millis(timeout_ms), + stream.read(&mut buf), + ) + .await; + + let bytes_read = match read_result { + Ok(Ok(0)) => { + return RtspProbe { + status: "no_response".to_string(), + error: Some("empty response".to_string()), + }; + } + Ok(Ok(n)) => n, + Ok(Err(err)) => { + return RtspProbe { + status: "read_error".to_string(), + error: Some(err.to_string()), + }; + } + Err(_) => { + return RtspProbe { + status: "timeout".to_string(), + error: Some("read timeout".to_string()), + }; + } + }; + + match parse_rtsp_status(&buf[..bytes_read]) { + Ok(Some(200)) => RtspProbe { + status: "ok".to_string(), + error: None, + }, + Ok(Some(401)) | Ok(Some(403)) => RtspProbe { + status: "auth_required".to_string(), + error: None, + }, + Ok(Some(code)) => RtspProbe { + status: format!("rtsp_{code}"), + error: None, + }, + Ok(None) => RtspProbe { + status: "bad_response".to_string(), + error: Some("missing RTSP status line".to_string()), + }, + Err(err) => RtspProbe { + status: "bad_response".to_string(), + error: Some(err.to_string()), + }, + } +} + +#[derive(Debug, Clone)] +struct RtspProbe { + status: String, + error: Option, +} + +impl RtspProbe { + fn is_connectable(&self) -> bool { + matches!( + self.status.as_str(), + "ok" | "auth_required" | "rtsp_200" | "rtsp_401" | "rtsp_403" + ) + } +} + +fn parse_rtsp_status(response: &[u8]) -> Result> { + let response = std::str::from_utf8(response).context("RTSP response is not valid UTF-8")?; + let first_line = response.lines().next().unwrap_or("").trim(); + if !first_line.starts_with("RTSP/1.") { + return Ok(None); + } + + let mut parts = first_line.split_whitespace(); + let _version = parts.next(); + let code = parts + .next() + .ok_or_else(|| anyhow!("RTSP response missing status code"))?; + let code = code + .parse::() + .with_context(|| format!("invalid RTSP status code: {code}"))?; + Ok(Some(code)) +} + +fn extract_host(url: &str) -> String { + let rest = url.strip_prefix("rtsp://").unwrap_or(url); + let authority = rest.split('/').next().unwrap_or(rest); + authority.to_string() +} + +async fn pop_queue_item(redis_url: &str, queue_name: &str) -> Result> { + let output = redis_cli(redis_url, &["--raw", "LPOP", queue_name]).await?; + let value = output.trim(); + if value.is_empty() { + return Ok(None); + } + Ok(Some(value.to_string())) +} + +async fn redis_key_exists(redis_url: &str, key: &str) -> Result { + let output = redis_cli(redis_url, &["--raw", "EXISTS", key]).await?; + Ok(output.trim() == "1") +} + +async fn clear_queue(redis_url: &str, queue_name: &str, done_key: &str) -> Result<()> { + let _ = redis_cli(redis_url, &["DEL", queue_name, done_key]).await?; + Ok(()) +} + +async fn mark_done(redis_url: &str, done_key: &str) -> Result<()> { + let _ = redis_cli(redis_url, &["SET", done_key, "1"]).await?; + Ok(()) +} + +async fn enqueue_ips( + redis_url: &str, + queue_name: &str, + ips: &[String], + worker_count: usize, +) -> Result<()> { + for chunk in ips.chunks(400) { + let mut args = Vec::with_capacity(chunk.len() + 2); + args.push("RPUSH".to_string()); + args.push(queue_name.to_string()); + args.extend(chunk.iter().cloned()); + let args_ref = args.iter().map(String::as_str).collect::>(); + let _ = redis_cli(redis_url, &args_ref).await?; + } + + let mut sentinels = Vec::with_capacity(worker_count + 2); + sentinels.push("RPUSH".to_string()); + sentinels.push(queue_name.to_string()); + sentinels.extend(std::iter::repeat_with(|| DEFAULT_SENTINEL.to_string()).take(worker_count)); + let sentinel_args = sentinels.iter().map(String::as_str).collect::>(); + let _ = redis_cli(redis_url, &sentinel_args).await?; + Ok(()) +} + +async fn redis_cli(redis_url: &str, args: &[&str]) -> Result { + let mut command = Command::new("redis-cli"); + command.kill_on_drop(true).arg("-u").arg(redis_url); + for arg in args { + command.arg(arg); + } + + let output = timeout(Duration::from_secs(15), async { command.output().await }) + .await + .context("redis-cli timeout")??; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string(); + return Err(anyhow!( + "redis-cli failed{}", + if stderr.is_empty() { + String::new() + } else { + format!(": {stderr}") + } + )); + } + + Ok(String::from_utf8_lossy(&output.stdout).to_string()) +} + +async fn init_sqlite(sqlite_path: &Path) -> Result<()> { + let sql = r#" + CREATE TABLE IF NOT EXISTS rtsp_feed_results ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + checked_at INTEGER NOT NULL, + target_ip TEXT NOT NULL, + rtsp_path TEXT NOT NULL, + rtsp_url TEXT NOT NULL, + screenshot_path TEXT, + status TEXT NOT NULL, + error TEXT + ); + "#; + sqlite_exec(sqlite_path, sql).await +} + +async fn insert_sqlite_result( + sqlite_path: &Path, + checked_at: i64, + target_ip: &str, + rtsp_path: &str, + rtsp_url: &str, + status: String, + screenshot_path: Option<&str>, + error: Option<&str>, +) -> Result<()> { + let sql = format!( + "INSERT INTO rtsp_feed_results (checked_at, target_ip, rtsp_path, rtsp_url, screenshot_path, status, error) VALUES ({checked_at}, {target_ip}, {rtsp_path}, {rtsp_url}, {screenshot_path}, {status}, {error});", + checked_at = checked_at, + target_ip = sql_quote(target_ip), + rtsp_path = sql_quote(rtsp_path), + rtsp_url = sql_quote(rtsp_url), + screenshot_path = sql_nullable_quote(screenshot_path), + status = sql_quote(&status), + error = sql_nullable_quote(error), + ); + sqlite_exec(sqlite_path, &sql).await +} + +async fn sqlite_exec(sqlite_path: &Path, sql: &str) -> Result<()> { + let output = timeout( + Duration::from_secs(20), + async { + let mut command = Command::new("sqlite3"); + command.kill_on_drop(true).arg(sqlite_path).arg(sql); + command.output().await + }, + ) + .await + .context("sqlite3 timeout")??; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string(); + return Err(anyhow!( + "sqlite3 failed{}", + if stderr.is_empty() { + String::new() + } else { + format!(": {stderr}") + } + )); + } + + Ok(()) +} + +fn load_ip_list(input: &str) -> Result> { + let content = fs::read_to_string(input) + .with_context(|| format!("failed to read input CSV {input}"))?; + let mut ips = Vec::new(); + + for line in content.lines() { + let line = line.trim(); + if line.is_empty() || line.starts_with('#') { + continue; + } + + let Some(first) = line.split(',').next() else { + continue; + }; + let value = strip_quotes(first.trim()); + if value.is_empty() || is_header(value) { + continue; + } + ips.push(value.to_string()); + } + + Ok(ips) +} + +fn normalize_paths(paths: &[String]) -> Vec { + paths + .iter() + .map(|path| path.trim()) + .filter(|path| !path.is_empty()) + .map(|path| { + if path.starts_with('/') { + path.to_string() + } else { + format!("/{path}") + } + }) + .collect() +} + +fn build_rtsp_url(ip: &str, port: u16, path: &str) -> String { + format!("rtsp://{ip}:{port}{path}") +} + +fn build_snapshot_path(base_dir: &Path, ip: &str, rtsp_path: &str, checked_at: i64) -> PathBuf { + let ip_component = sanitize_component(ip); + let path_component = sanitize_component(rtsp_path); + base_dir.join(format!("{ip_component}_{checked_at}_{path_component}.png")) +} + +fn ensure_output_dirs(snapshot_dir: &Path, sqlite_path: &Path) -> Result<()> { + fs::create_dir_all(snapshot_dir) + .with_context(|| format!("failed to create snapshot dir {}", snapshot_dir.display()))?; + if let Some(parent) = sqlite_path.parent() { + fs::create_dir_all(parent) + .with_context(|| format!("failed to create sqlite dir {}", parent.display()))?; + } + Ok(()) +} + +fn sql_quote(value: &str) -> String { + format!("'{}'", value.replace('\'', "''")) +} + +fn sql_nullable_quote(value: Option<&str>) -> String { + match value { + Some(value) if !value.is_empty() => sql_quote(value), + _ => "NULL".to_string(), + } +} + +fn strip_quotes(value: &str) -> &str { + value + .strip_prefix('"') + .and_then(|value| value.strip_suffix('"')) + .or_else(|| { + value + .strip_prefix('\'') + .and_then(|value| value.strip_suffix('\'')) + }) + .unwrap_or(value) +} + +fn is_header(value: &str) -> bool { + matches!( + value.to_ascii_lowercase().as_str(), + "ip" | "host" | "hostname" | "address" | "target" | "camera" + ) +} + +fn sanitize_component(value: &str) -> String { + value + .chars() + .map(|ch| match ch { + 'a'..='z' | 'A'..='Z' | '0'..='9' | '.' | '-' | '_' => ch, + _ => '_', + }) + .collect() +} + +fn timestamp_millis() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_millis() as i64) + .unwrap_or_default() +} diff --git a/src/rtsp.rs b/src/rtsp.rs new file mode 100644 index 0000000..70028d4 --- /dev/null +++ b/src/rtsp.rs @@ -0,0 +1,376 @@ +use crate::{ + config::Config, + db::Database, + terminal::{ScreenState, TerminalUi}, +}; +use anyhow::{anyhow, Context, Result}; +use std::{ + sync::{ + atomic::{AtomicBool, AtomicUsize, Ordering}, + Arc, + }, + time::Instant, +}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpStream, + time::timeout, +}; + +#[derive(Debug, Clone)] +struct RtspAttempt { + host: String, + path: String, + url: String, +} + +#[derive(Debug, Clone)] +struct RtspOutcome { + status: String, + response_code: Option, + error: Option, +} + +pub async fn run_rtsp_scan( + config: Config, + database: Arc, + ui: Arc, + interrupted: Arc, +) -> Result<()> { + let targets = normalize_targets(&config.rtsp_targets); + let paths = normalize_paths(&config.rtsp_paths); + + if targets.is_empty() { + return Err(anyhow!( + "nenhum alvo RTSP informado. Use RTSP_TARGETS_FILE, RTSP_TARGETS ou --targets-file/--targets" + )); + } + + if paths.is_empty() { + return Err(anyhow!("nenhum path RTSP válido foi informado")); + } + + let attempts = build_attempts(&targets, &paths, config.port); + let total = attempts.len(); + let start = Instant::now(); + + if ui.is_enabled() { + ui.print_banner(); + ui.render(&ScreenState { + current_ip: attempts + .first() + .map(|attempt| attempt.url.clone()) + .unwrap_or_else(|| "concluído".to_string()), + tested: 0, + total, + }); + } else { + println!( + "RTSP scan pronto para {} alvos e {} paths ({} tentativas totais), porta {}, concorrência {}.", + targets.len(), + paths.len(), + total, + config.port, + config.concurrency + ); + } + + let next_attempt = Arc::new(AtomicUsize::new(0)); + let tested_this_run = Arc::new(AtomicUsize::new(0)); + let successful_this_run = Arc::new(AtomicUsize::new(0)); + let auth_required_this_run = Arc::new(AtomicUsize::new(0)); + let mut handles = Vec::new(); + + for _ in 0..config.concurrency.max(1) { + let attempts = attempts.clone(); + let next_attempt = Arc::clone(&next_attempt); + let tested_this_run = Arc::clone(&tested_this_run); + let successful_this_run = Arc::clone(&successful_this_run); + let auth_required_this_run = Arc::clone(&auth_required_this_run); + let database = Arc::clone(&database); + let ui = Arc::clone(&ui); + let interrupted = Arc::clone(&interrupted); + let config = config.clone(); + + handles.push(tokio::spawn(async move { + loop { + if interrupted.load(Ordering::SeqCst) { + break; + } + + let index = next_attempt.fetch_add(1, Ordering::SeqCst); + let Some(attempt) = attempts.get(index).cloned() else { + break; + }; + + let outcome = probe_rtsp(&attempt, config.port, config.timeout_ms).await; + database + .upsert_rtsp_probe_result( + &attempt.host, + config.port, + &attempt.path, + &attempt.url, + outcome.response_code, + &outcome.status, + outcome.error.as_deref(), + ) + .await?; + + let tested_now = tested_this_run.fetch_add(1, Ordering::SeqCst) + 1; + if outcome.status == "ok" { + successful_this_run.fetch_add(1, Ordering::SeqCst); + } else if outcome.status == "auth_required" { + auth_required_this_run.fetch_add(1, Ordering::SeqCst); + } + + if should_render_progress(tested_now, total) || tested_now == total { + ui.render(&ScreenState { + current_ip: attempt.url, + tested: tested_now, + total, + }); + } + } + + Ok::<(), anyhow::Error>(()) + })); + } + + for handle in handles { + handle.await??; + } + + let duration = start.elapsed().as_secs_f64(); + let tested = tested_this_run.load(Ordering::SeqCst); + let success = successful_this_run.load(Ordering::SeqCst); + let auth_required = auth_required_this_run.load(Ordering::SeqCst); + + if interrupted.load(Ordering::SeqCst) { + if ui.is_enabled() { + ui.render(&ScreenState { + current_ip: "interrompido".to_string(), + tested, + total, + }); + ui.finish(); + } else { + println!("RTSP scan interrompido após {:.2}s.", duration); + println!("Tentativas executadas: {}", tested); + println!("Streams confirmados: {}", success); + println!("Respostas com autenticação: {}", auth_required); + } + return Ok(()); + } + + if ui.is_enabled() { + ui.render(&ScreenState { + current_ip: "concluído".to_string(), + tested: total, + total, + }); + ui.finish(); + } else { + println!("RTSP scan concluído em {:.2}s", duration); + println!("Tentativas executadas: {}", tested); + println!("Streams confirmados: {}", success); + println!("Respostas com autenticação: {}", auth_required); + } + + Ok(()) +} + +fn build_attempts(targets: &[String], paths: &[String], port: u16) -> Vec { + let mut attempts = Vec::with_capacity(targets.len() * paths.len()); + for host in targets { + for path in paths { + attempts.push(RtspAttempt { + host: host.clone(), + path: path.clone(), + url: format!("rtsp://{}:{}{}", host, port, path), + }); + } + } + attempts +} + +async fn probe_rtsp(attempt: &RtspAttempt, port: u16, timeout_ms: u64) -> RtspOutcome { + let connect_result = timeout( + std::time::Duration::from_millis(timeout_ms), + TcpStream::connect(format!("{}:{}", attempt.host, port)), + ) + .await; + + let mut stream = match connect_result { + Ok(Ok(stream)) => stream, + Ok(Err(err)) => { + return RtspOutcome { + status: "connect_error".to_string(), + response_code: None, + error: Some(err.to_string()), + }; + } + Err(_) => { + return RtspOutcome { + status: "timeout".to_string(), + response_code: None, + error: Some("connect timeout".to_string()), + }; + } + }; + + let request = build_describe_request(&attempt.url); + let write_result = timeout( + std::time::Duration::from_millis(timeout_ms), + stream.write_all(request.as_bytes()), + ) + .await; + match write_result { + Ok(Ok(())) => {} + Ok(Err(err)) => { + return RtspOutcome { + status: "write_error".to_string(), + response_code: None, + error: Some(err.to_string()), + }; + } + Err(_) => { + return RtspOutcome { + status: "timeout".to_string(), + response_code: None, + error: Some("write timeout".to_string()), + }; + } + } + + let mut buf = vec![0u8; 2048]; + let read_result = timeout( + std::time::Duration::from_millis(timeout_ms), + stream.read(&mut buf), + ) + .await; + + let bytes_read = match read_result { + Ok(Ok(0)) => { + return RtspOutcome { + status: "no_response".to_string(), + response_code: None, + error: Some("empty response".to_string()), + }; + } + Ok(Ok(n)) => n, + Ok(Err(err)) => { + return RtspOutcome { + status: "read_error".to_string(), + response_code: None, + error: Some(err.to_string()), + }; + } + Err(_) => { + return RtspOutcome { + status: "timeout".to_string(), + response_code: None, + error: Some("read timeout".to_string()), + }; + } + }; + + match parse_rtsp_status(&buf[..bytes_read]) { + Ok(Some(code)) if code == 200 => RtspOutcome { + status: "ok".to_string(), + response_code: Some(code), + error: None, + }, + Ok(Some(code)) if code == 401 || code == 403 => RtspOutcome { + status: "auth_required".to_string(), + response_code: Some(code), + error: None, + }, + Ok(Some(code)) if code == 404 => RtspOutcome { + status: "not_found".to_string(), + response_code: Some(code), + error: None, + }, + Ok(Some(code)) => RtspOutcome { + status: "rtsp_response".to_string(), + response_code: Some(code), + error: None, + }, + Ok(None) => RtspOutcome { + status: "bad_response".to_string(), + response_code: None, + error: Some("missing RTSP status line".to_string()), + }, + Err(err) => RtspOutcome { + status: "bad_response".to_string(), + response_code: None, + error: Some(err.to_string()), + }, + } +} + +fn build_describe_request(url: &str) -> String { + format!( + "DESCRIBE {url} RTSP/1.0\r\nCSeq: 1\r\nUser-Agent: CamFinder/0.1\r\nAccept: application/sdp\r\n\r\n" + ) +} + +fn parse_rtsp_status(response: &[u8]) -> Result> { + let response = std::str::from_utf8(response).context("RTSP response is not valid UTF-8")?; + let first_line = response.lines().next().unwrap_or("").trim(); + if !first_line.starts_with("RTSP/1.") { + return Ok(None); + } + + let mut parts = first_line.split_whitespace(); + let _version = parts.next(); + let code = parts + .next() + .ok_or_else(|| anyhow!("RTSP response missing status code"))?; + let code = code + .parse::() + .with_context(|| format!("invalid RTSP status code: {code}"))?; + Ok(Some(code)) +} + +fn normalize_targets(targets: &[String]) -> Vec { + targets + .iter() + .map(|target| target.trim()) + .filter(|target| !target.is_empty()) + .map(strip_rtsp_target) + .filter(|target| !target.is_empty()) + .map(ToString::to_string) + .collect() +} + +fn normalize_paths(paths: &[String]) -> Vec { + paths + .iter() + .map(|path| path.trim()) + .filter(|path| !path.is_empty()) + .map(|path| { + if path.starts_with('/') { + path.to_string() + } else { + format!("/{path}") + } + }) + .collect() +} + +fn strip_rtsp_target(value: &str) -> &str { + value + .strip_prefix("rtsp://") + .unwrap_or(value) + .split('/') + .next() + .unwrap_or(value) +} + +fn should_render_progress(tested: usize, total: usize) -> bool { + if total == 0 { + return false; + } + + tested == total || tested % 25 == 0 || tested * 4 >= total +}