Compare commits

...

5 Commits

Author SHA1 Message Date
valmojr 2ccdfaf1c7 added dump folders and csv results to ignore list 2026-06-29 23:55:49 -03:00
valmojr d18ce214fd Added RTSP parser to dockerfiles 2026-06-29 23:55:08 -03:00
valmojr 99365c2c6e Added main RTSP pathes 2026-06-29 23:54:48 -03:00
valmojr cabcc65954 Added RTSP parser 2026-06-29 23:54:34 -03:00
valmojr 5ed9cfbc2a Added check on first frame feed 2026-06-29 21:20:54 -03:00
11 changed files with 1537 additions and 6 deletions
+4 -1
View File
@@ -5,4 +5,7 @@ dev.db*
prisma/dev.db
prisma/dev.db-journal
dist
target
target
*.csv
*.sql
dump
+1 -1
View File
@@ -6,5 +6,5 @@ edition = "2021"
[dependencies]
anyhow = "1"
dotenvy = "0.15"
tokio = { version = "1", features = ["io-util", "macros", "net", "rt-multi-thread", "signal", "time"] }
tokio = { version = "1", features = ["io-util", "macros", "net", "process", "rt-multi-thread", "signal", "time"] }
tokio-postgres = "0.7"
+2 -1
View File
@@ -8,11 +8,12 @@ RUN cargo build --release
FROM debian:bookworm-slim
RUN apt-get update \
&& apt-get install -y --no-install-recommends ca-certificates \
&& apt-get install -y --no-install-recommends ca-certificates ffmpeg redis-tools sqlite3 \
&& rm -rf /var/lib/apt/lists/*
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
ENTRYPOINT ["camfinder"]
+37 -1
View File
@@ -2,11 +2,12 @@
Scanner TCP em Rust para descobrir hosts em CIDRs internos, testar a porta `554` e salvar progresso e endpoints abertos em PostgreSQL.
O projeto tem três modos:
O projeto tem quatro modos:
- `local`: roda o scan inteiro no processo atual.
- `coordinator`: recebe conexões de workers, distribui blocos e grava resultados no banco.
- `worker`: conecta ao coordenador, recebe blocos e testa os hosts.
- `rtsp`: lê destinos de um CSV e testa URLs RTSP com paths predefinidos.
O progresso é salvo por bloco, então o scan pode ser retomado sem repetir o que já foi concluído.
@@ -15,6 +16,7 @@ O progresso é salvo por bloco, então o scan pode ser retomado sem repetir o qu
1. Copie `.env.example` para `.env`.
2. Configure `DATABASE_URL` apontando para um Postgres acessível.
3. Rode `cargo run -- local`, `cargo run -- coordinator` ou `cargo run -- worker`.
4. Para RTSP, rode `cargo run -- rtsp --targets-file targets.csv`.
## Docker Compose
@@ -24,6 +26,12 @@ O `docker-compose.yml` já sobe:
- `coordinator` exposto no host pela porta `666`.
- `worker` conectado ao coordenador na rede interna do compose.
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.
Para escalar workers:
- `docker compose up --scale worker=3`
@@ -51,6 +59,34 @@ Para workers de outras máquinas, use `vps.valmo.dev:666` como endereço do coor
- `COORDINATOR_BIND`
- `COORDINATOR_ADDR`
- `WORKER_NAME`
- `RTSP_TARGETS`
- `RTSP_TARGETS_FILE`
- `RTSP_PATHS`
- `RTSP_PATHS_FILE`
- `REDIS_URL`
- `RTSP_QUEUE_NAME`
- `RTSP_INPUT_CSV`
- `RTSP_WORKER_COUNT`
- `RTSP_SNAPSHOT_DIR`
- `RTSP_SQLITE_PATH`
- `RTSP_CAPTURE_TIMEOUT_MS`
## Modo RTSP
O modo `rtsp` monta URLs no formato `rtsp://<host>:554/<path>` e testa a conexão
via `DESCRIBE`, sem tentar usuário, senha ou variações de autenticação.
Por padrão, ele lê os paths de [rtsp_paths.txt](/home/valmo/Documents/CamFinder/rtsp_paths.txt).
Exemplo:
```bash
cargo run -- rtsp --targets-file targets.csv
```
No fluxo `docker-compose.redis-rtsp.yml`, os workers salvam os frames em
`dump/feeds/<ip>_<timestamp>.png` e gravam os resultados em
`dump/rtsp-results.sqlite`.
## Restrições
+53
View File
@@ -0,0 +1,53 @@
services:
redis:
image: redis:7-alpine
command: ["redis-server", "--appendonly", "yes"]
volumes:
- camfinder-redis:/data
coordinator:
build: .
command: ["redis-coordinator"]
environment:
REDIS_URL: redis://redis:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_INPUT_CSV: /app/camfinder-open-rtsp.csv
RTSP_WORKER_COUNT: 2
depends_on:
- redis
volumes:
- ./camfinder-open-rtsp.csv:/app/camfinder-open-rtsp.csv:ro
- ./dump:/app/dump
worker-1:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://redis:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_SQLITE_PATH: /app/dump/rtsp-results.sqlite
RTSP_CAPTURE_TIMEOUT_MS: 15000
depends_on:
- redis
- coordinator
volumes:
- ./dump:/app/dump
worker-2:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://redis:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_SQLITE_PATH: /app/dump/rtsp-results.sqlite
RTSP_CAPTURE_TIMEOUT_MS: 15000
depends_on:
- redis
- coordinator
volumes:
- ./dump:/app/dump
volumes:
camfinder-redis:
+114
View File
@@ -0,0 +1,114 @@
/
/1
/1.AMP
/1/stream1
/11
/12
/CAM_ID.password.mp2
/GetData.cgi
/LowResolutionVideo
/MediaInput/h264
/MediaInput/mpeg4
/ONVIF/MediaInput
/Streaming/Channels/1
/Streaming/Channels/101
/Streaming/Channels/102
/Streaming/Channels/2
/VideoInput/1/h264/1
/VideoInput/1/mpeg4/1
/access_code
/access_name_for_stream_1_to_5
/axis-media/media.amp
/axis-media/media.amp?videocodec=h264
/av0_0
/av2
/avn=2
/cam
/cam/realmonitor
/cam/realmonitor?channel=1&subtype=0
/cam/realmonitor?channel=1&subtype=1
/cam0_0
/cam0_1
/cam1/h264
/cam1/h264/multicast
/cam1/mjpeg
/cam1/mpeg4
/cam1/onvif-h264
/camera.stm
/ch0
/ch0_0.h264
/ch001.sdp
/ch01.264
/ch0_unicast_firststream
/ch0_unicast_secondstream
/ch1-s1
/channel1
/h264
/h264/media.amp
/h264_vga.sdp
/image.mpg
/img/media.sav
/img/video.asf
/img/video.sav
/ioImage/1
/ipcam.sdp
/ipcam_h264.sdp
/live
/live.sdp
/live/av0
/live/ch00_0
/live/h264
/live/mpeg4
/live3.sdp
/live_mpeg4.sdp
/livestream
/livestream/
/media
/media.amp
/media/media.amp
/media/video1
/medias2
/mjpeg/media.smp
/mp4
/mpeg4
/mpeg4/1/media.amp
/mpeg4/media.amp
/mpeg4/media.smp
/mpeg4unicast
/mpg4/rtsp.amp
/multicaststream
/now.mp4
/nph-h264.cgi
/nphMpeg4/g726-640x
/nphMpeg4/g726-640x48
/nphMpeg4/g726-640x480
/nphMpeg4/nil-320x240
/onvif-media/media.amp
/onvif1
/play1.sdp
/play2.sdp
/rtpvideo1.sdp
/rtsp_live0
/rtsp_live1
/rtsp_live2
/rtsp_tunnel
/rtsph264
/rtsph2641080p
/stream
/stream1
/ucast/11
/user.pin.mp2
/user_defined
/video
/video.3gp
/video.h264
/video.mjpg
/video.mp4
/video.pro1
/video.pro2
/video.pro3
/video1
/video1+audio1
/videoMain
/vis
/wfov
+260 -1
View File
@@ -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<String>,
pub rtsp_targets: Vec<String>,
pub rtsp_paths: Vec<String>,
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<String>,
cidr_file: Option<PathBuf>,
targets: Option<String>,
targets_file: Option<PathBuf>,
paths: Option<String>,
paths_file: Option<PathBuf>,
redis_url: Option<String>,
redis_queue_name: Option<String>,
redis_input_csv: Option<String>,
redis_worker_count: Option<usize>,
snapshot_dir: Option<PathBuf>,
sqlite_path: Option<PathBuf>,
capture_timeout_ms: Option<u64>,
database_url: Option<String>,
coordinator_bind: Option<SocketAddr>,
coordinator_addr: Option<String>,
@@ -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<Mode> {
"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<String> {
.collect()
}
fn load_csv_targets(file: Option<&Path>, inline: Option<&str>) -> Result<Vec<String>> {
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<String> {
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<String> {
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<Vec<String>> {
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<Vec<String>> {
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<String> {
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() {
+74
View File
@@ -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<u16>,
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
+9 -1
View File
@@ -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,
};
+607
View File
@@ -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<TerminalUi>,
_interrupted: Arc<AtomicBool>,
) -> 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<TerminalUi>,
interrupted: Arc<AtomicBool>,
) -> 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<String>,
}
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<Option<u16>> {
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::<u16>()
.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<Option<String>> {
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<bool> {
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::<Vec<_>>();
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::<Vec<_>>();
let _ = redis_cli(redis_url, &sentinel_args).await?;
Ok(())
}
async fn redis_cli(redis_url: &str, args: &[&str]) -> Result<String> {
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<Vec<String>> {
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<String> {
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()
}
+376
View File
@@ -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<u16>,
error: Option<String>,
}
pub async fn run_rtsp_scan(
config: Config,
database: Arc<Database>,
ui: Arc<TerminalUi>,
interrupted: Arc<AtomicBool>,
) -> 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<RtspAttempt> {
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<Option<u16>> {
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::<u16>()
.with_context(|| format!("invalid RTSP status code: {code}"))?;
Ok(Some(code))
}
fn normalize_targets(targets: &[String]) -> Vec<String> {
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<String> {
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
}