Compare commits

...

31 Commits

Author SHA1 Message Date
valmojr c38ec59815 added redis port to host 2026-06-30 22:23:09 -03:00
valmojr 9a5f80974a fixed typos for redis 2026-06-30 18:39:38 -03:00
valmojr e81211c88e Refactored redis component to use a centralized postgres database with coordinator instead of descentralized sqlite shitty databases 2026-06-30 18:07:47 -03:00
valmojr bae8712f3b added handler that is going to jump pathes if the returned status is 5xx or 404 2026-06-30 17:44:16 -03:00
valmojr fc3b2ef309 fixed sqlite lock 2026-06-30 02:39:24 -03:00
valmojr 8416b92a8d added params to bin 2026-06-30 02:19:49 -03:00
valmojr 5761deb298 ultima chance 2026-06-30 02:16:23 -03:00
valmojr 1fbe319a10 sei la2 2026-06-30 02:07:59 -03:00
valmojr b1572c029e sei lá 2026-06-30 01:49:11 -03:00
valmojr 52c93d2cd5 added root to root 2026-06-30 01:35:31 -03:00
valmojr fd410f8109 added pro max redis compose 2026-06-30 01:26:11 -03:00
valmojr 32ccdbec83 Added csv to docker image 2026-06-30 01:13:48 -03:00
valmojr 29dfcf169c Added csv 2026-06-30 00:56:08 -03:00
valmojr 0fc1473a6f Improved scale on redis workers 2026-06-30 00:36:31 -03:00
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
Valmo Trindade d83acce887 fixed coord timeouts 2026-06-29 10:35:50 +00:00
valmojr 189b496ad9 fixed inet typos 2026-06-29 07:13:21 -03:00
valmojr e886d643a8 Fixed worker UI 2026-06-29 07:08:25 -03:00
valmojr 60df71843a added hostname to connectors 2026-06-29 06:58:02 -03:00
valmojr 8f791a700b 666 2026-06-29 06:39:52 -03:00
valmojr fe73a95b15 removido DB URL de workers 2026-06-29 06:32:40 -03:00
valmojr 811b59b5f4 Fixed CIDR parsing on workers 2026-06-29 06:26:19 -03:00
Valmo Trindade da0b9b71bb fixed dockerfiçe 2026-06-29 09:27:23 +00:00
valmojr 9838e1b612 Changed default CIDR 2026-06-29 06:12:21 -03:00
valmojr 75fb09e767 Added one more worker to the compose 2026-06-29 06:05:02 -03:00
valmojr 9dce723945 Added test CIDRS 2026-06-29 05:08:27 -03:00
valmojr 0f1c867771 Unlimited CIDR parsing 2026-06-29 04:53:53 -03:00
21 changed files with 11939 additions and 119 deletions
+2
View File
@@ -6,3 +6,5 @@ prisma/dev.db
prisma/dev.db-journal prisma/dev.db-journal
dist dist
target target
*.sql
dump
Generated
+30
View File
@@ -64,6 +64,7 @@ version = "0.1.0"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"dotenvy", "dotenvy",
"terminal_size",
"tokio", "tokio",
"tokio-postgres", "tokio-postgres",
] ]
@@ -255,6 +256,12 @@ dependencies = [
"libc", "libc",
] ]
[[package]]
name = "linux-raw-sys"
version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53"
[[package]] [[package]]
name = "lock_api" name = "lock_api"
version = "0.4.14" version = "0.4.14"
@@ -454,6 +461,19 @@ dependencies = [
"bitflags", "bitflags",
] ]
[[package]]
name = "rustix"
version = "1.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190"
dependencies = [
"bitflags",
"errno",
"libc",
"linux-raw-sys",
"windows-sys",
]
[[package]] [[package]]
name = "rustversion" name = "rustversion"
version = "1.0.22" version = "1.0.22"
@@ -566,6 +586,16 @@ dependencies = [
"unicode-ident", "unicode-ident",
] ]
[[package]]
name = "terminal_size"
version = "0.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "230a1b821ccbd75b185820a1f1ff7b14d21da1e442e22c0863ea5f08771a8874"
dependencies = [
"rustix",
"windows-sys",
]
[[package]] [[package]]
name = "tinyvec" name = "tinyvec"
version = "1.11.0" version = "1.11.0"
+2 -1
View File
@@ -6,5 +6,6 @@ edition = "2021"
[dependencies] [dependencies]
anyhow = "1" anyhow = "1"
dotenvy = "0.15" 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" tokio-postgres = "0.7"
terminal_size = "0.4"
+10 -2
View File
@@ -1,17 +1,25 @@
FROM rust:1.78-bookworm AS builder FROM rust:1.96.0-bookworm AS builder
WORKDIR /app WORKDIR /app
COPY Cargo.toml Cargo.lock ./ COPY Cargo.toml Cargo.lock ./
COPY src ./src COPY src ./src
COPY cidrs.txt ./cidrs.txt
COPY camfinder-open-rtsp.csv ./camfinder-open-rtsp.csv
COPY rtsp_paths.txt ./rtsp_paths.txt
COPY main_ids.json ./main_ids.json
RUN cargo build --release RUN cargo build --release
FROM debian:bookworm-slim FROM debian:bookworm-slim
RUN apt-get update \ 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 \
&& rm -rf /var/lib/apt/lists/* && rm -rf /var/lib/apt/lists/*
WORKDIR /app WORKDIR /app
COPY --from=builder /app/target/release/camfinder /usr/local/bin/camfinder COPY --from=builder /app/target/release/camfinder /usr/local/bin/camfinder
COPY cidrs.txt /app/cidrs.txt
COPY camfinder-open-rtsp.csv /app/camfinder-open-rtsp.csv
COPY rtsp_paths.txt /app/rtsp_paths.txt
COPY main_ids.json /app/main_ids.json
ENTRYPOINT ["camfinder"] ENTRYPOINT ["camfinder"]
+43 -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. 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. - `local`: roda o scan inteiro no processo atual.
- `coordinator`: recebe conexões de workers, distribui blocos e grava resultados no banco. - `coordinator`: recebe conexões de workers, distribui blocos e grava resultados no banco.
- `worker`: conecta ao coordenador, recebe blocos e testa os hosts. - `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. 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`. 1. Copie `.env.example` para `.env`.
2. Configure `DATABASE_URL` apontando para um Postgres acessível. 2. Configure `DATABASE_URL` apontando para um Postgres acessível.
3. Rode `cargo run -- local`, `cargo run -- coordinator` ou `cargo run -- worker`. 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 ## Docker Compose
@@ -24,10 +26,20 @@ O `docker-compose.yml` já sobe:
- `coordinator` exposto no host pela porta `666`. - `coordinator` exposto no host pela porta `666`.
- `worker` conectado ao coordenador na rede interna do compose. - `worker` conectado ao coordenador na rede interna do compose.
Para o fluxo RTSP com Redis, use `docker-compose.redis-rtsp.yml`:
- `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. No modo simples/local, monte o repositório inteiro em `/app`.
- `worker-1` e `worker-2` para consumir blocos, testar paths RTSP e salvar screenshots e SQLite.
Para escalar workers: Para escalar workers:
- `docker compose up --scale worker=3` - `docker compose up --scale worker=3`
Para um worker em outra máquina, aponte `REDIS_URL` para o Redis do coordenador, por exemplo `redis://<host>:6379/0`, e use o mesmo `RTSP_QUEUE_NAME`.
Se estiver usando o compose local, monte `./:/app` para coordinator e workers RTSP. O coordinator lê `camfinder-open-rtsp.csv` e `rtsp_paths.txt` da raiz do repo.
Para workers de outras máquinas, use `vps.valmo.dev:666` como endereço do coordenador. Para workers de outras máquinas, use `vps.valmo.dev:666` como endereço do coordenador.
## Scripts ## Scripts
@@ -51,6 +63,36 @@ Para workers de outras máquinas, use `vps.valmo.dev:666` como endereço do coor
- `COORDINATOR_BIND` - `COORDINATOR_BIND`
- `COORDINATOR_ADDR` - `COORDINATOR_ADDR`
- `WORKER_NAME` - `WORKER_NAME`
- `RTSP_TARGETS`
- `RTSP_TARGETS_FILE`
- `RTSP_PATHS`
- `RTSP_PATHS_FILE`
- `REDIS_URL`
- `RTSP_QUEUE_NAME`
- `RTSP_INPUT_CSV`
- `RTSP_WORKER_COUNT`
- `RTSP_QUEUE_BLOCK_SIZE`
- `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`, 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).
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 ## Restrições
File diff suppressed because it is too large Load Diff
+176
View File
@@ -0,0 +1,176 @@
# Faixas Residenciais dos Principais Provedores Brasileiros
# Foco em IPs domésticos (banda larga fixa, fibra, cabo)
# === CLARO/NET (AS28573) ===
# Faixas residenciais típicas de banda larga
177.0.0.0/16
177.1.0.0/16
177.2.0.0/16
187.0.0.0/16
187.1.0.0/16
187.2.0.0/16
189.0.0.0/16
189.1.0.0/16
191.0.0.0/16
191.1.0.0/16
# === VIVO/TELEFÔNICA (AS27699) ===
# Fibra e DSL residencial
177.32.0.0/16
177.33.0.0/16
177.64.0.0/16
177.65.0.0/16
187.32.0.0/16
187.33.0.0/16
187.64.0.0/16
187.65.0.0/16
189.32.0.0/16
189.33.0.0/16
200.144.0.0/16
200.145.0.0/16
# === TIM (AS26615) ===
# Banda larga residencial (antiga Intelig/Ajato)
177.128.0.0/16
177.129.0.0/16
177.130.0.0/16
179.96.0.0/16
179.97.0.0/16
187.128.0.0/16
187.129.0.0/16
189.128.0.0/16
189.129.0.0/16
191.128.0.0/16
191.129.0.0/16
# === OI (AS7738) ===
# Velox/Oi Fibra residencial
177.192.0.0/16
177.193.0.0/16
187.192.0.0/16
187.193.0.0/16
189.192.0.0/16
189.193.0.0/16
191.192.0.0/16
191.193.0.0/16
200.192.0.0/16
200.193.0.0/16
# === ALGAR TELECOM (AS16735) ===
# CTBC residencial
177.16.0.0/16
177.17.0.0/16
187.16.0.0/16
187.17.0.0/16
# === COPEL TELECOM (AS14868) ===
# Fibra residencial
177.96.0.0/16
177.97.0.0/16
# === PROVEDORES REGIONAIS ===
# (faixas frequentemente usadas para residências)
# Brisanet (AS28329) - Nordeste
138.0.0.0/16
138.1.0.0/16
138.2.0.0/16
# Vero/Sercomtel (AS28271)
139.0.0.0/16
139.1.0.0/16
# Desktop/Sumicity (AS28624) - Interior SP
143.0.0.0/16
143.1.0.0/16
# Unifique (AS28343) - Sul
146.0.0.0/16
146.1.0.0/16
# Sim Fibra (AS26615)
147.0.0.0/16
147.1.0.0/16
# V.tal (antiga Oi infraestrutura)
148.0.0.0/16
148.1.0.0/16
# Vogel Telecom (AS262605)
150.0.0.0/16
150.1.0.0/16
# IP Telecom (AS262602)
152.0.0.0/16
152.1.0.0/16
# WCS Conecta (AS263019)
155.0.0.0/16
155.1.0.0/16
# Linktel (AS262591)
157.0.0.0/16
157.1.0.0/16
# MHNet (AS262589)
158.0.0.0/16
158.1.0.0/16
# Brasil TecPar (AS262669)
159.0.0.0/16
159.1.0.0/16
# Conexão Fiber (AS263009)
160.0.0.0/16
160.1.0.0/16
# NET LIFE (AS262648)
161.0.0.0/16
161.1.0.0/16
# IDEAL (AS262569)
162.0.0.0/16
162.1.0.0/16
# RLINE (AS262570)
164.0.0.0/16
164.1.0.0/16
# SUPER IP (AS262974)
165.0.0.0/16
165.1.0.0/16
# PPLINK (AS262998)
167.0.0.0/16
167.1.0.0/16
# RAMO (AS262957)
170.0.0.0/16
170.1.0.0/16
# === FAIXAS ADICIONAIS PEQUENAS ===
# Provedores locais/municipais
# Blocos /24 residenciais comuns
177.10.0.0/16
177.100.0.0/16
179.100.0.0/16
186.200.0.0/16
187.10.0.0/16
187.100.0.0/16
189.10.0.0/16
189.100.0.0/16
191.10.0.0/16
191.100.0.0/16
200.100.0.0/16
201.10.0.0/16
201.100.0.0/16
# Faixas de provedores de bairro
131.0.0.0/16
142.0.0.0/16
156.0.0.0/16
163.0.0.0/16
166.0.0.0/16
168.0.0.0/16
181.0.0.0/16
+70
View File
@@ -0,0 +1,70 @@
services:
redis:
image: redis:7-alpine
command: ["redis-server", "--appendonly", "yes", "--bind", "0.0.0.0", "--protected-mode", "no"]
volumes:
- camfinder-redis:/data
postgres:
image: postgres:16-alpine
environment:
POSTGRES_DB: camfinder
POSTGRES_USER: camfinder
POSTGRES_PASSWORD: camfinder
volumes:
- camfinder-postgres:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U camfinder -d camfinder"]
interval: 5s
timeout: 5s
retries: 10
start_period: 5s
coordinator:
build: .
command: ["redis-coordinator"]
environment:
REDIS_URL: redis://redis:6379/0
DATABASE_URL: postgres://camfinder:camfinder@postgres:5432/camfinder
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_INPUT_CSV: /app/camfinder-open-rtsp.csv
RTSP_WORKER_COUNT: 2
depends_on:
redis:
condition: service_started
postgres:
condition: service_healthy
volumes:
- ./:/app
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_CAPTURE_TIMEOUT_MS: 15000
depends_on:
- redis
- coordinator
volumes:
- ./:/app
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_CAPTURE_TIMEOUT_MS: 15000
depends_on:
- redis
- coordinator
volumes:
- ./:/app
volumes:
camfinder-redis:
camfinder-postgres:
+148
View File
@@ -0,0 +1,148 @@
services:
redis:
image: redis:7-alpine
command: ["redis-server", "--appendonly", "yes", "--bind", "0.0.0.0", "--protected-mode", "no"]
volumes:
- camfinder-redis:/data
ports:
- "6379:6379"
postgres:
image: postgres:16-alpine
environment:
POSTGRES_DB: camfinder
POSTGRES_USER: camfinder
POSTGRES_PASSWORD: camfinder
volumes:
- camfinder-postgres:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U camfinder -d camfinder"]
interval: 5s
timeout: 5s
retries: 10
start_period: 5s
coordinator:
build: .
command: ["redis-coordinator"]
environment:
REDIS_URL: redis://redis:6379/0
DATABASE_URL: postgres://camfinder:camfinder@postgres:5432/camfinder
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_INPUT_CSV: /app/camfinder-open-rtsp.csv
RTSP_WORKER_COUNT: 8
depends_on:
redis:
condition: service_started
postgres:
condition: service_healthy
volumes:
- ./:/app:ro
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_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_CAPTURE_TIMEOUT_MS: 15000
depends_on:
- worker-1
volumes:
- ./dump:/app/dump
worker-3:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://redis:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
depends_on:
- worker-2
volumes:
- ./dump:/app/dump
worker-4:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://redis:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
depends_on:
- worker-3
volumes:
- ./dump:/app/dump
worker-5:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://redis:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
depends_on:
- worker-4
volumes:
- ./dump:/app/dump
worker-6:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://redis:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
depends_on:
- worker-5
volumes:
- ./dump:/app/dump
worker-7:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://redis:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
depends_on:
- worker-6
volumes:
- ./dump:/app/dump
worker-8:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://redis:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
depends_on:
- worker-7
volumes:
- ./dump:/app/dump
volumes:
camfinder-redis:
camfinder-postgres:
+96
View File
@@ -0,0 +1,96 @@
services:
worker-1:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://192.168.15.100:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
restart: unless-stopped
volumes:
- ./dump:/app/dump
worker-2:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://192.168.15.100:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
restart: unless-stopped
volumes:
- ./dump:/app/dump
worker-3:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://192.168.15.100:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
restart: unless-stopped
volumes:
- ./dump:/app/dump
worker-4:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://192.168.15.100:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
restart: unless-stopped
volumes:
- ./dump:/app/dump
worker-5:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://192.168.15.100:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
restart: unless-stopped
volumes:
- ./dump:/app/dump
worker-6:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://192.168.15.100:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
restart: unless-stopped
volumes:
- ./dump:/app/dump
worker-7:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://192.168.15.100:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
restart: unless-stopped
volumes:
- ./dump:/app/dump
worker-8:
build: .
command: ["redis-worker"]
environment:
REDIS_URL: redis://192.168.15.100:6379/0
RTSP_QUEUE_NAME: camfinder:rtsp:queue
RTSP_SNAPSHOT_DIR: /app/dump/feeds
RTSP_CAPTURE_TIMEOUT_MS: 15000
restart: unless-stopped
volumes:
- ./dump:/app/dump
+14 -5
View File
@@ -19,7 +19,7 @@ services:
command: ["coordinator"] command: ["coordinator"]
environment: environment:
DATABASE_URL: postgres://camfinder:camfinder@postgres:5432/camfinder DATABASE_URL: postgres://camfinder:camfinder@postgres:5432/camfinder
SCAN_CIDRS: 192.168.15.0/24 SCAN_CIDR_FILE: /app/cidrs.txt
SCAN_PORT: 554 SCAN_PORT: 554
SCAN_TIMEOUT_MS: 1200 SCAN_TIMEOUT_MS: 1200
SCAN_CONCURRENCY: 16 SCAN_CONCURRENCY: 16
@@ -31,15 +31,24 @@ services:
ports: ports:
- "666:666" - "666:666"
worker: worker-1:
build: . build: .
command: ["worker"] command: ["worker", "--connect", "coordinator:666", "--name", "worker-1"]
environment:
SCAN_PORT: 554
SCAN_TIMEOUT_MS: 1200
SCAN_CONCURRENCY: 16
depends_on:
coordinator:
condition: service_started
worker-2:
build: .
command: ["worker", "--connect", "coordinator:666", "--name", "worker-2"]
environment: environment:
COORDINATOR_ADDR: coordinator:666
SCAN_PORT: 554 SCAN_PORT: 554
SCAN_TIMEOUT_MS: 1200 SCAN_TIMEOUT_MS: 1200
SCAN_CONCURRENCY: 16 SCAN_CONCURRENCY: 16
WORKER_NAME: worker
depends_on: depends_on:
coordinator: coordinator:
condition: service_started condition: service_started
+19
View File
@@ -0,0 +1,19 @@
{
"usernames": [
"",
"admin"
],
"passwords" : [
"",
"admin",
"8888",
"9999",
"1234",
"12345",
"123456",
"123456789",
"1234567890",
"pass",
"password"
]
}
+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
+343 -24
View File
@@ -10,19 +10,31 @@ pub enum Mode {
Local, Local,
Coordinator, Coordinator,
Worker, Worker,
Rtsp,
RedisCoordinator,
RedisWorker,
} }
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct Config { pub struct Config {
pub mode: Mode, pub mode: Mode,
pub cidrs: Vec<String>, 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 redis_queue_block_size: usize,
pub snapshot_dir: PathBuf,
pub capture_timeout_ms: u64,
pub port: u16, pub port: u16,
pub timeout_ms: u64, pub timeout_ms: u64,
pub concurrency: usize, pub concurrency: usize,
pub chunk_size: usize, pub chunk_size: usize,
pub database_url: String, pub database_url: String,
pub coordinator_bind: SocketAddr, pub coordinator_bind: SocketAddr,
pub coordinator_addr: SocketAddr, pub coordinator_addr: String,
pub worker_name: String, pub worker_name: String,
} }
@@ -33,9 +45,37 @@ impl Config {
let cli = CliArgs::from_args(&args)?; let cli = CliArgs::from_args(&args)?;
let cidrs = load_cidrs(cli.cidr_file.as_deref(), cli.cidrs.as_deref())?; 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 { Ok(Self {
mode, mode,
cidrs, 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", "/app/camfinder-open-rtsp.csv")),
redis_worker_count: cli
.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"))),
capture_timeout_ms: cli
.capture_timeout_ms
.or_else(|| parse_u64("RTSP_CAPTURE_TIMEOUT_MS", 15000).ok())
.unwrap_or(15000),
port: cli port: cli
.port .port
.or_else(|| parse_u16("SCAN_PORT", 554).ok()) .or_else(|| parse_u16("SCAN_PORT", 554).ok())
@@ -59,12 +99,7 @@ impl Config {
.unwrap_or_else(|| "0.0.0.0:666".parse().expect("valid default bind address")), .unwrap_or_else(|| "0.0.0.0:666".parse().expect("valid default bind address")),
coordinator_addr: cli coordinator_addr: cli
.coordinator_addr .coordinator_addr
.or_else(|| parse_socket_addr("COORDINATOR_ADDR", "127.0.0.1:666").ok()) .unwrap_or_else(|| read_env("COORDINATOR_ADDR", "127.0.0.1:666")),
.unwrap_or_else(|| {
"127.0.0.1:666"
.parse()
.expect("valid default coordinator address")
}),
worker_name: cli.worker_name.unwrap_or_else(default_worker_name), worker_name: cli.worker_name.unwrap_or_else(default_worker_name),
}) })
} }
@@ -74,9 +109,20 @@ impl Config {
struct CliArgs { struct CliArgs {
cidrs: Option<String>, cidrs: Option<String>,
cidr_file: Option<PathBuf>, 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>,
redis_queue_block_size: Option<usize>,
snapshot_dir: Option<PathBuf>,
capture_timeout_ms: Option<u64>,
database_url: Option<String>, database_url: Option<String>,
coordinator_bind: Option<SocketAddr>, coordinator_bind: Option<SocketAddr>,
coordinator_addr: Option<SocketAddr>, coordinator_addr: Option<String>,
worker_name: Option<String>, worker_name: Option<String>,
port: Option<u16>, port: Option<u16>,
timeout_ms: Option<u64>, timeout_ms: Option<u64>,
@@ -99,6 +145,60 @@ impl CliArgs {
cli.cidr_file = cli.cidr_file =
Some(PathBuf::from(next_value(args, &mut index, "--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",
)?)?);
}
"--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",
)?));
}
"--capture-timeout-ms" => {
cli.capture_timeout_ms = Some(parse_u64_literal(&next_value(
args,
&mut index,
"--capture-timeout-ms",
)?)?);
}
"--database" | "--database-url" => { "--database" | "--database-url" => {
cli.database_url = Some(next_value(args, &mut index, arg)?); cli.database_url = Some(next_value(args, &mut index, arg)?);
} }
@@ -108,11 +208,7 @@ impl CliArgs {
)?)?); )?)?);
} }
"--connect" => { "--connect" => {
cli.coordinator_addr = Some(parse_socket_addr_literal(&next_value( cli.coordinator_addr = Some(next_value(args, &mut index, "--connect")?);
args,
&mut index,
"--connect",
)?)?);
} }
"--name" => { "--name" => {
cli.worker_name = Some(next_value(args, &mut index, "--name")?); cli.worker_name = Some(next_value(args, &mut index, "--name")?);
@@ -176,8 +272,11 @@ fn parse_mode_literal(value: &str) -> Result<Mode> {
"local" | "scan" => Ok(Mode::Local), "local" | "scan" => Ok(Mode::Local),
"coordinator" | "coord" | "server" => Ok(Mode::Coordinator), "coordinator" | "coord" | "server" => Ok(Mode::Coordinator),
"worker" => Ok(Mode::Worker), "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!( 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"
)), )),
} }
} }
@@ -200,16 +299,6 @@ fn load_cidrs(file: Option<&Path>, inline: Option<&str>) -> Result<Vec<String>>
return Err(anyhow!("arquivo de CIDRs vazio: {}", file.display())); return Err(anyhow!("arquivo de CIDRs vazio: {}", file.display()));
} }
if let Ok(value) = env::var("SCAN_CIDRS") {
let value = value.trim();
if !value.is_empty() {
let cidrs = parse_cidr_list(value);
if !cidrs.is_empty() {
return Ok(cidrs);
}
}
}
if let Ok(path) = env::var("SCAN_CIDR_FILE") { if let Ok(path) = env::var("SCAN_CIDR_FILE") {
let path = path.trim(); let path = path.trim();
if !path.is_empty() { if !path.is_empty() {
@@ -223,6 +312,16 @@ fn load_cidrs(file: Option<&Path>, inline: Option<&str>) -> Result<Vec<String>>
} }
} }
if let Ok(value) = env::var("SCAN_CIDRS") {
let value = value.trim();
if !value.is_empty() {
let cidrs = parse_cidr_list(value);
if !cidrs.is_empty() {
return Ok(cidrs);
}
}
}
if let Ok(value) = env::var("SCAN_CIDR") { if let Ok(value) = env::var("SCAN_CIDR") {
let value = value.trim(); let value = value.trim();
if !value.is_empty() { if !value.is_empty() {
@@ -247,6 +346,226 @@ fn parse_cidr_list(input: &str) -> Vec<String> {
.collect() .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 {
match fs::read_to_string(file) {
Ok(content) => {
let targets = parse_csv_targets(&content);
if !targets.is_empty() {
return Ok(targets);
}
return Err(anyhow!("arquivo CSV vazio: {}", file.display()));
}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
let targets = parse_csv_targets(include_str!("../camfinder-open-rtsp.csv"));
if !targets.is_empty() {
return Ok(targets);
}
return Err(anyhow!(
"arquivo CSV padrão embutido vazio: ../camfinder-open-rtsp.csv"
));
}
Err(err) => {
return Err(anyhow!(err))
.with_context(|| format!("failed to read CSV file {}", file.display()));
}
}
}
if let Ok(path) = env::var("RTSP_TARGETS_FILE") {
let path = path.trim();
if !path.is_empty() {
match fs::read_to_string(path) {
Ok(content) => {
let targets = parse_csv_targets(&content);
if !targets.is_empty() {
return Ok(targets);
}
return Err(anyhow!("arquivo CSV vazio: {path}"));
}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
let targets = parse_csv_targets(include_str!("../camfinder-open-rtsp.csv"));
if !targets.is_empty() {
return Ok(targets);
}
return Err(anyhow!(
"arquivo CSV padrão embutido vazio: ../camfinder-open-rtsp.csv"
));
}
Err(err) => {
return Err(err).with_context(|| format!("failed to read CSV file {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 match load_rtsp_paths_from_file(Path::new(path)) {
Ok(paths) => Ok(paths),
Err(err) if is_not_found(&err) => {
let paths = parse_rtsp_paths(include_str!("../rtsp_paths.txt"));
if !paths.is_empty() {
Ok(paths)
} else {
Err(anyhow!(
"arquivo de paths RTSP padrão embutido vazio: ../rtsp_paths.txt"
))
}
}
Err(err) => Err(err),
};
}
}
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);
}
}
}
match load_rtsp_paths_from_file(Path::new("/app/rtsp_paths.txt")) {
Ok(paths) => Ok(paths),
Err(err) if is_not_found(&err) => {
let paths = parse_rtsp_paths(include_str!("../rtsp_paths.txt"));
if !paths.is_empty() {
Ok(paths)
} else {
Err(anyhow!(
"arquivo de paths RTSP padrão embutido vazio: ../rtsp_paths.txt"
))
}
}
Err(err) => Err(err),
}
}
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 is_not_found(err: &anyhow::Error) -> bool {
err.chain()
.find_map(|cause| cause.downcast_ref::<std::io::Error>())
.map(|err| err.kind() == std::io::ErrorKind::NotFound)
.unwrap_or(false)
}
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 { fn database_url_from_env() -> String {
let value = read_env("DATABASE_URL", ""); let value = read_env("DATABASE_URL", "");
if value.is_empty() { if value.is_empty() {
+12 -3
View File
@@ -113,7 +113,7 @@ pub async fn run_coordinator(
} }
maybe_finished = handlers.join_next(), if !handlers.is_empty() => { maybe_finished = handlers.join_next(), if !handlers.is_empty() => {
if let Some(join_result) = maybe_finished { if let Some(join_result) = maybe_finished {
join_result??; log_worker_result(join_result);
} }
} }
} }
@@ -124,7 +124,7 @@ pub async fn run_coordinator(
} }
while let Some(join_result) = handlers.join_next().await { while let Some(join_result) = handlers.join_next().await {
join_result??; log_worker_result(join_result);
} }
let duration = start.elapsed().as_secs_f64(); let duration = start.elapsed().as_secs_f64();
@@ -175,6 +175,14 @@ pub async fn run_coordinator(
Ok(()) Ok(())
} }
fn log_worker_result(result: std::result::Result<Result<()>, tokio::task::JoinError>) {
match result {
Ok(Ok(())) => {}
Ok(Err(err)) => eprintln!("worker handler ended with error: {err:#}"),
Err(err) => eprintln!("worker task failed: {err}"),
}
}
async fn handle_worker( async fn handle_worker(
stream: TcpStream, stream: TcpStream,
peer: std::net::SocketAddr, peer: std::net::SocketAddr,
@@ -194,7 +202,8 @@ async fn handle_worker(
let worker_name = match lines.next_line().await? { let worker_name = match lines.next_line().await? {
Some(line) if line.starts_with("HELLO ") => line[6..].trim().to_string(), Some(line) if line.starts_with("HELLO ") => line[6..].trim().to_string(),
Some(line) => { Some(line) => {
return Err(anyhow!("unexpected worker handshake from {peer}: {line}")); eprintln!("rejected non-worker connection from {peer}: {line}");
return Ok(());
} }
None => return Ok(()), None => return Ok(()),
}; };
+155 -2
View File
@@ -51,6 +51,47 @@ impl Database {
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_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);
CREATE TABLE IF NOT EXISTS rtsp_fetch_results (
id BIGSERIAL PRIMARY KEY,
checked_at TIMESTAMPTZ NOT NULL,
checked_at_ms BIGINT NOT NULL,
target_ip TEXT NOT NULL,
target_port INTEGER NOT NULL,
rtsp_path TEXT NOT NULL,
rtsp_url TEXT NOT NULL,
screenshot_path TEXT,
status TEXT NOT NULL,
error TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS rtsp_fetch_results_checked_at_idx
ON rtsp_fetch_results (checked_at);
CREATE INDEX IF NOT EXISTS rtsp_fetch_results_target_ip_idx
ON rtsp_fetch_results (target_ip);
CREATE INDEX IF NOT EXISTS rtsp_fetch_results_status_idx
ON rtsp_fetch_results (status);
"#, "#,
) )
.await .await
@@ -134,7 +175,7 @@ impl Database {
} }
pub async fn upsert_open_endpoint(&self, ip: std::net::Ipv4Addr, port: u16) -> Result<()> { pub async fn upsert_open_endpoint(&self, ip: std::net::Ipv4Addr, port: u16) -> Result<()> {
let ip = ip.to_string(); let ip = std::net::IpAddr::V4(ip);
let port = i32::from(port); let port = i32::from(port);
self.client self.client
@@ -148,7 +189,7 @@ impl Database {
created_at, created_at,
updated_at updated_at
) )
VALUES ($1::inet, $2, NOW(), NOW(), NOW(), NOW()) VALUES ($1, $2, NOW(), NOW(), NOW(), NOW())
ON CONFLICT (ip) DO UPDATE SET ON CONFLICT (ip) DO UPDATE SET
port = EXCLUDED.port, port = EXCLUDED.port,
scanned_at = NOW(), scanned_at = NOW(),
@@ -178,6 +219,118 @@ impl Database {
i64_to_usize(count, "open endpoint count") 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 insert_rtsp_fetch_result(
&self,
checked_at_ms: i64,
target_ip: &str,
target_port: u16,
rtsp_path: &str,
rtsp_url: &str,
screenshot_path: Option<&str>,
status: &str,
error: Option<&str>,
) -> Result<()> {
let target_port = i32::from(target_port);
self.client
.execute(
r#"
INSERT INTO rtsp_fetch_results (
checked_at,
checked_at_ms,
target_ip,
target_port,
rtsp_path,
rtsp_url,
screenshot_path,
status,
error
)
VALUES (
TO_TIMESTAMP(($1::BIGINT)::DOUBLE PRECISION / 1000.0),
$1,
$2,
$3,
$4,
$5,
$6,
$7,
$8
)
"#,
&[
&checked_at_ms,
&target_ip,
&target_port,
&rtsp_path,
&rtsp_url,
&screenshot_path,
&status,
&error,
],
)
.await
.context("failed to insert RTSP fetch result")?;
Ok(())
}
pub async fn summarize_chunks(&self) -> Result<(usize, usize)> { pub async fn summarize_chunks(&self) -> Result<(usize, usize)> {
let row = self let row = self
.client .client
+218 -35
View File
@@ -3,6 +3,8 @@ mod coordinator;
mod db; mod db;
mod ip; mod ip;
mod models; mod models;
mod redis_rtsp;
mod rtsp;
mod scan; mod scan;
mod terminal; mod terminal;
mod worker; mod worker;
@@ -21,11 +23,15 @@ async fn main() -> Result<()> {
dotenvy::dotenv().ok(); dotenvy::dotenv().ok();
let config = Config::from_env_and_args()?; let config = Config::from_env_and_args()?;
if matches!(config.mode, Mode::Local | Mode::Coordinator) {
assert_local_cidrs(&config.cidrs)?;
}
let database = Arc::new(Database::open(&config.database_url).await?); let config = match config.mode {
Mode::Local | Mode::Coordinator => Config {
cidrs: validate_and_filter_cidrs(&config.cidrs)?,
..config
},
Mode::Worker | Mode::Rtsp | Mode::RedisCoordinator | Mode::RedisWorker => config,
};
let ui = Arc::new(TerminalUi::new(stdout_is_terminal())); let ui = Arc::new(TerminalUi::new(stdout_is_terminal()));
let interrupted = Arc::new(AtomicBool::new(false)); let interrupted = Arc::new(AtomicBool::new(false));
@@ -38,8 +44,20 @@ async fn main() -> Result<()> {
} }
let result = match config.mode { let result = match config.mode {
Mode::Local => scan::run_local_scan(config, database, ui, interrupted).await, Mode::Local => {
Mode::Coordinator => coordinator::run_coordinator(config, database, ui, interrupted).await, let database = Arc::new(Database::open(&config.database_url).await?);
scan::run_local_scan(config, database, ui, interrupted).await
}
Mode::Coordinator => {
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, Mode::Worker => worker::run_worker(config, ui).await,
}; };
@@ -51,68 +69,187 @@ async fn main() -> Result<()> {
Ok(()) Ok(())
} }
fn assert_local_cidrs(cidrs: &[String]) -> Result<()> { /// Valida todos os CIDRs e remove os que são de redes privadas/não roteáveis
fn validate_and_filter_cidrs(cidrs: &[String]) -> Result<Vec<String>> {
if cidrs.is_empty() { if cidrs.is_empty() {
return Err(anyhow::anyhow!( return Err(anyhow::anyhow!(
"nenhum CIDR foi informado. Use SCAN_CIDRS, SCAN_CIDR_FILE ou --cidrs/--cidr-file" "nenhum CIDR foi informado. Use SCAN_CIDRS, SCAN_CIDR_FILE ou --cidrs/--cidr-file"
)); ));
} }
let mut valid_cidrs = Vec::new();
let mut skipped_count = 0;
for cidr in cidrs { for cidr in cidrs {
assert_local_cidr_inner(cidr)?; match validate_cidr(cidr) {
Ok(CidrType::Routable) => {
valid_cidrs.push(cidr.clone());
}
Ok(CidrType::Private) => {
eprintln!("️ Ignorando CIDR privado/não roteável: {cidr}");
skipped_count += 1;
}
Err(err) => {
eprintln!("⚠️ CIDR inválido ignorado: {cidr} ({err})");
}
}
} }
Ok(()) if skipped_count > 0 {
eprintln!(
"️ Total de {} CIDR(s) privados/não roteáveis ignorados",
skipped_count
);
}
if valid_cidrs.is_empty() {
return Err(anyhow::anyhow!(
"nenhum CIDR público roteável encontrado. Todos os CIDRs fornecidos são privados ou inválidos"
));
}
Ok(valid_cidrs)
} }
fn assert_local_cidr_inner(cidr: &str) -> Result<()> { #[derive(Debug, PartialEq)]
enum CidrType {
Routable,
Private,
}
/// Valida um CIDR e determina se é roteável ou privado
fn validate_cidr(cidr: &str) -> Result<CidrType> {
let (network, prefix) = cidr let (network, prefix) = cidr
.split_once('/') .split_once('/')
.ok_or_else(|| anyhow::anyhow!("CIDR inválido: {cidr}"))?; .ok_or_else(|| anyhow::anyhow!("formato CIDR inválido: {cidr}"))?;
let prefix: u8 = prefix let prefix: u8 = prefix
.parse() .parse()
.map_err(|_| anyhow::anyhow!("máscara de sub-rede inválida: {prefix}"))?; .map_err(|_| anyhow::anyhow!("máscara de sub-rede inválida: {prefix}"))?;
if prefix > 32 { if prefix > 32 {
return Err(anyhow::anyhow!("máscara de sub-rede inválida: {prefix}")); return Err(anyhow::anyhow!(
"máscara de sub-rede deve ser <= 32: {prefix}"
));
} }
let ip: std::net::Ipv4Addr = network let ip: std::net::Ipv4Addr = network
.parse() .parse()
.map_err(|_| anyhow::anyhow!("endereço IP inválido: {network}"))?; .map_err(|_| anyhow::anyhow!("endereço IP inválido: {network}"))?;
// Verifica se algum IP na faixa é roteável
// Para /24 ou menores, verificamos o primeiro IP
// Para ranges maiores, verificamos se a faixa toda é privada
let start = u32::from(ip) & cidr_mask(prefix); let start = u32::from(ip) & cidr_mask(prefix);
if is_routable_ip(start) {
return Ok(CidrType::Routable);
}
// Se o primeiro IP é privado, verifica o último
let end = if prefix == 32 { let end = if prefix == 32 {
start start
} else { } else {
start | !cidr_mask(prefix) start | !cidr_mask(prefix)
}; };
let private_ranges = [ if is_routable_ip(end) {
( return Ok(CidrType::Routable);
u32::from(std::net::Ipv4Addr::new(10, 0, 0, 0)),
u32::from(std::net::Ipv4Addr::new(10, 255, 255, 255)),
),
(
u32::from(std::net::Ipv4Addr::new(172, 16, 0, 0)),
u32::from(std::net::Ipv4Addr::new(172, 31, 255, 255)),
),
(
u32::from(std::net::Ipv4Addr::new(192, 168, 0, 0)),
u32::from(std::net::Ipv4Addr::new(192, 168, 255, 255)),
),
];
let allowed = private_ranges
.iter()
.any(|(range_start, range_end)| start >= *range_start && end <= *range_end);
if !allowed {
return Err(anyhow::anyhow!(
"CIDR fora de redes locais permitidas: {cidr}. Use 10.0.0.0/8, 172.16.0.0/12 ou 192.168.0.0/16"
));
} }
Ok(()) // Ambos são privados, então o CIDR é privado
Ok(CidrType::Private)
}
/// Verifica se um endereço IPv4 (como u32) é roteável na internet pública
fn is_routable_ip(addr: u32) -> bool {
let octets = [
((addr >> 24) & 0xFF) as u8,
((addr >> 16) & 0xFF) as u8,
((addr >> 8) & 0xFF) as u8,
(addr & 0xFF) as u8,
];
let first = octets[0];
let second = octets[1];
let third = octets[2];
// 0.0.0.0/8 - Reservado (endereço não especificado)
if first == 0 {
return false;
}
// 10.0.0.0/8 - Privado (redes internas grandes)
if first == 10 {
return false;
}
// 100.64.0.0/10 - CGNAT (compartilhado por operadoras)
if first == 100 && second >= 64 && second <= 127 {
return false;
}
// 127.0.0.0/8 - Loopback
if first == 127 {
return false;
}
// 169.254.0.0/16 - Link-local (APIPA)
if first == 169 && second == 254 {
return false;
}
// 172.16.0.0/12 - Privado (redes internas médias)
if first == 172 && second >= 16 && second <= 31 {
return false;
}
// 192.0.0.0/24 - Reservado (usos especiais IETF)
if first == 192 && second == 0 && third == 0 {
return false;
}
// 192.0.2.0/24 - Documentação (TEST-NET-1)
if first == 192 && second == 0 && third == 2 {
return false;
}
// 192.88.99.0/24 - Reservado (IPv6 to IPv4 relay)
if first == 192 && second == 88 && third == 99 {
return false;
}
// 192.168.0.0/16 - Privado (redes domésticas/escritórios)
if first == 192 && second == 168 {
return false;
}
// 198.18.0.0/15 - Testes de benchmark
if first == 198 && (second == 18 || second == 19) {
return false;
}
// 198.51.100.0/24 - Documentação (TEST-NET-2)
if first == 198 && second == 51 && third == 100 {
return false;
}
// 203.0.113.0/24 - Documentação (TEST-NET-3)
if first == 203 && second == 0 && third == 113 {
return false;
}
// 224.0.0.0/4 - Multicast
if first >= 224 && first <= 239 {
return false;
}
// 240.0.0.0/4 - Reservado (uso futuro e especial)
if first >= 240 {
return false;
}
true
} }
fn cidr_mask(prefix: u8) -> u32 { fn cidr_mask(prefix: u8) -> u32 {
@@ -122,3 +259,49 @@ fn cidr_mask(prefix: u8) -> u32 {
u32::MAX << (32 - prefix) u32::MAX << (32 - prefix)
} }
} }
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_routable_ips() {
assert!(is_routable_ip(u32::from(std::net::Ipv4Addr::new(
1, 2, 3, 4
))));
assert!(is_routable_ip(u32::from(std::net::Ipv4Addr::new(
8, 8, 8, 8
))));
assert!(is_routable_ip(u32::from(std::net::Ipv4Addr::new(
203, 0, 114, 1
))));
}
#[test]
fn test_private_ips() {
assert!(!is_routable_ip(u32::from(std::net::Ipv4Addr::new(
10, 0, 0, 1
))));
assert!(!is_routable_ip(u32::from(std::net::Ipv4Addr::new(
172, 16, 0, 1
))));
assert!(!is_routable_ip(u32::from(std::net::Ipv4Addr::new(
192, 168, 1, 1
))));
assert!(!is_routable_ip(u32::from(std::net::Ipv4Addr::new(
127, 0, 0, 1
))));
}
#[test]
fn test_cidr_validation() {
assert_eq!(validate_cidr("1.2.3.0/24").unwrap(), CidrType::Routable);
assert_eq!(validate_cidr("8.8.8.8/32").unwrap(), CidrType::Routable);
assert_eq!(validate_cidr("192.168.1.0/24").unwrap(), CidrType::Private);
assert_eq!(validate_cidr("10.0.0.0/8").unwrap(), CidrType::Private);
assert!(validate_cidr("invalid").is_err());
assert!(validate_cidr("1.2.3.4/33").is_err());
assert!(validate_cidr("999.999.999.999/24").is_err());
}
}
+1457
View File
File diff suppressed because it is too large Load Diff
+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
}
+165
View File
@@ -1,5 +1,7 @@
use std::io::{self, IsTerminal, Write}; use std::io::{self, IsTerminal, Write};
use std::sync::Mutex; use std::sync::Mutex;
use std::time::Duration;
use terminal_size::{terminal_size, Width};
pub const BANNER: [&str; 9] = [ pub const BANNER: [&str; 9] = [
" ░██████ ░██████ ", " ░██████ ░██████ ",
@@ -25,6 +27,23 @@ pub struct TerminalUi {
lock: Mutex<()>, lock: Mutex<()>,
} }
#[derive(Debug, Clone)]
pub struct RedisWorkerScreenState {
pub worker_name: String,
pub queue_name: String,
pub current_ip: String,
pub current_path: String,
pub current_status: String,
pub processed: usize,
pub captured: usize,
pub global_done: usize,
pub global_total: usize,
pub queue_remaining: Option<usize>,
pub block_done: usize,
pub block_total: usize,
pub elapsed: Duration,
}
impl TerminalUi { impl TerminalUi {
pub fn new(enabled: bool) -> Self { pub fn new(enabled: bool) -> Self {
Self { Self {
@@ -56,6 +75,104 @@ impl TerminalUi {
let _ = stdout.flush(); let _ = stdout.flush();
} }
pub fn render_redis_worker(&self, state: &RedisWorkerScreenState) {
if !self.enabled {
return;
}
let _guard = self.lock.lock().expect("terminal mutex poisoned");
let mut stdout = io::stdout();
let width = terminal_width();
let content_width = width.saturating_sub(2).max(32);
let global_percent = percent(state.global_done, state.global_total);
let block_percent = percent(state.block_done, state.block_total);
let global_bar = progress_bar(global_percent, content_width.saturating_sub(24));
let block_bar = progress_bar(block_percent, content_width.saturating_sub(24));
let queue_remaining = state
.queue_remaining
.map(|value| value.to_string())
.unwrap_or_else(|| "?".to_string());
let _ = write!(stdout, "\x1b[2J\x1b[H\x1b[?25l");
let _ = writeln!(
stdout,
"{}",
fit_line(
&format!(
"CamFinder Redis Worker {} pid:{}",
state.worker_name,
std::process::id()
),
content_width
)
);
let _ = writeln!(stdout, "{}", "".repeat(content_width));
let _ = writeln!(
stdout,
"{}",
fit_line(&format!("Fila: {}", state.queue_name), content_width)
);
let _ = writeln!(
stdout,
"{}",
fit_line(
&format!(
"Global: [{}] {:>5.1}% {}/{} IPs redis:{}",
global_bar,
global_percent,
state.global_done,
state.global_total,
queue_remaining
),
content_width
)
);
let _ = writeln!(
stdout,
"{}",
fit_line(
&format!(
"Bloco : [{}] {:>5.1}% {}/{} tentativas",
block_bar, block_percent, state.block_done, state.block_total
),
content_width
)
);
let _ = writeln!(stdout, "{}", "".repeat(content_width));
let _ = writeln!(
stdout,
"{}",
fit_line(&format!("IP atual : {}", state.current_ip), content_width)
);
let _ = writeln!(
stdout,
"{}",
fit_line(&format!("Path : {}", state.current_path), content_width)
);
let _ = writeln!(
stdout,
"{}",
fit_line(
&format!("Status : {}", state.current_status),
content_width
)
);
let _ = writeln!(
stdout,
"{}",
fit_line(
&format!(
"Tentativas: {} Capturas: {} Tempo: {}",
state.processed,
state.captured,
format_duration(state.elapsed)
),
content_width
)
);
let _ = stdout.flush();
}
pub fn render(&self, state: &ScreenState) { pub fn render(&self, state: &ScreenState) {
if !self.enabled { if !self.enabled {
return; return;
@@ -97,3 +214,51 @@ impl TerminalUi {
pub fn stdout_is_terminal() -> bool { pub fn stdout_is_terminal() -> bool {
io::stdout().is_terminal() io::stdout().is_terminal()
} }
fn terminal_width() -> usize {
terminal_size()
.map(|(Width(width), _)| usize::from(width))
.unwrap_or(80)
}
fn percent(done: usize, total: usize) -> f64 {
if total == 0 {
0.0
} else {
((done as f64 / total as f64) * 100.0).clamp(0.0, 100.0)
}
}
fn progress_bar(percent: f64, width: usize) -> String {
let width = width.clamp(8, 80);
let filled = ((percent / 100.0) * width as f64).round() as usize;
let empty = width.saturating_sub(filled);
format!("{}{}", "".repeat(filled), "".repeat(empty))
}
fn fit_line(value: &str, width: usize) -> String {
let chars = value.chars().collect::<Vec<_>>();
if chars.len() <= width {
return value.to_string();
}
if width <= 1 {
return "".to_string();
}
chars
.into_iter()
.take(width - 1)
.chain(std::iter::once('…'))
.collect()
}
fn format_duration(duration: Duration) -> String {
let total = duration.as_secs();
let hours = total / 3600;
let minutes = (total % 3600) / 60;
let seconds = total % 60;
if hours > 0 {
format!("{hours:02}:{minutes:02}:{seconds:02}")
} else {
format!("{minutes:02}:{seconds:02}")
}
}
+98 -45
View File
@@ -7,7 +7,7 @@ use anyhow::{anyhow, Context, Result};
use std::{net::Ipv4Addr, sync::Arc}; use std::{net::Ipv4Addr, sync::Arc};
use tokio::{ use tokio::{
io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
net::TcpStream, net::{tcp::OwnedWriteHalf, TcpStream},
task::JoinSet, task::JoinSet,
time::sleep, time::sleep,
}; };
@@ -17,45 +17,16 @@ pub async fn run_worker(config: Config, ui: Arc<TerminalUi>) -> Result<()> {
ui.print_banner(); ui.print_banner();
} }
let stream = connect_with_retry(config.coordinator_addr).await?;
let (read_half, mut write_half) = stream.into_split();
let mut lines = BufReader::new(read_half).lines();
write_half
.write_all(format!("HELLO {}\n", config.worker_name).as_bytes())
.await?;
loop { loop {
let Some(line) = lines.next_line().await? else { render_worker_status(&ui, "conectando", 0, 0);
break;
};
if line == "DONE" { match run_worker_session(&config, &ui).await? {
break; WorkerSession::Done => break,
WorkerSession::Disconnected => {
eprintln!("coordinator connection lost; reconnecting...");
sleep(std::time::Duration::from_secs(1)).await;
}
} }
if line == "SHUTDOWN" {
break;
}
let job = parse_job_line(&line)?;
let result = execute_job(
&job.ips,
config.port,
config.timeout_ms,
config.concurrency,
&ui,
)
.await?;
let result_line = format!(
"RESULT {} {} {} {} {}\n",
job.chunk_start,
job.chunk_end,
result.tested,
result.open_ips.len(),
format_ips(&result.open_ips)
);
write_half.write_all(result_line.as_bytes()).await?;
} }
if ui.is_enabled() { if ui.is_enabled() {
@@ -70,7 +41,67 @@ pub async fn run_worker(config: Config, ui: Arc<TerminalUi>) -> Result<()> {
Ok(()) Ok(())
} }
async fn connect_with_retry(addr: std::net::SocketAddr) -> Result<TcpStream> { enum WorkerSession {
Done,
Disconnected,
}
async fn run_worker_session(config: &Config, ui: &TerminalUi) -> Result<WorkerSession> {
let stream = connect_with_retry(&config.coordinator_addr).await?;
let (read_half, mut write_half) = stream.into_split();
let mut lines = BufReader::new(read_half).lines();
write_half
.write_all(format!("HELLO {}\n", config.worker_name).as_bytes())
.await?;
render_worker_status(ui, "aguardando job", 0, 0);
loop {
let Some(line) = lines.next_line().await? else {
return Ok(WorkerSession::Disconnected);
};
if line == "DONE" || line == "SHUTDOWN" {
return Ok(WorkerSession::Done);
}
let job = parse_job_line(&line)?;
render_job_start(ui, &job);
let result = execute_job(
&job.ips,
config.port,
config.timeout_ms,
config.concurrency,
ui,
)
.await?;
if write_result(&mut write_half, &job, &result).await.is_err() {
return Ok(WorkerSession::Disconnected);
}
render_worker_status(ui, "aguardando job", result.tested, job.ips.len());
}
}
async fn write_result(
write_half: &mut OwnedWriteHalf,
job: &Job,
result: &JobOutcome,
) -> Result<()> {
let result_line = format!(
"RESULT {} {} {} {} {}\n",
job.chunk_start,
job.chunk_end,
result.tested,
result.open_ips.len(),
format_ips(&result.open_ips)
);
write_half.write_all(result_line.as_bytes()).await?;
Ok(())
}
async fn connect_with_retry(addr: &str) -> Result<TcpStream> {
let mut attempts = 0usize; let mut attempts = 0usize;
loop { loop {
@@ -133,6 +164,30 @@ fn parse_job_line(line: &str) -> Result<Job> {
}) })
} }
fn render_job_start(ui: &TerminalUi, job: &Job) {
let current_ip = job
.ips
.first()
.map(ToString::to_string)
.unwrap_or_else(|| "sem IP".to_string());
render_worker_status(ui, &current_ip, 0, job.ips.len());
}
fn render_worker_status(ui: &TerminalUi, current_ip: &str, tested: usize, total: usize) {
if ui.is_enabled() {
let display_total = if total == 0 && current_ip != "concluído" {
1
} else {
total
};
ui.render(&ScreenState {
current_ip: current_ip.to_string(),
tested,
total: display_total,
});
}
}
async fn execute_job( async fn execute_job(
ips: &[Ipv4Addr], ips: &[Ipv4Addr],
port: u16, port: u16,
@@ -160,13 +215,11 @@ async fn execute_job(
open_ips.push(ip); open_ips.push(ip);
} }
if ui.is_enabled() { ui.render(&ScreenState {
ui.render(&ScreenState { current_ip: ip.to_string(),
current_ip: ip.to_string(), tested,
tested, total: ips.len(),
total: ips.len(), });
});
}
} }
index = end; index = end;