diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..c5dd462 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,2 @@ +target +.env diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..1cca4e1 --- /dev/null +++ b/.env.example @@ -0,0 +1,11 @@ +DATABASE_URL="postgres://camfinder:camfinder@127.0.0.1:5432/camfinder" +SCAN_CIDRS="192.168.15.0/24" +SCAN_CIDR_FILE="" +SCAN_PORT=554 +SCAN_TIMEOUT_MS=1200 +SCAN_CONCURRENCY=16 +SCAN_CHUNK_SIZE=64 +SCAN_MODE="local" +COORDINATOR_BIND="0.0.0.0:666" +COORDINATOR_ADDR="127.0.0.1:666" +WORKER_NAME="worker-1" diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 0000000..3cdbb26 --- /dev/null +++ b/Cargo.lock @@ -0,0 +1,803 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "anyhow" +version = "1.0.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" + +[[package]] +name = "async-trait" +version = "0.1.89" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + +[[package]] +name = "bitflags" +version = "2.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" + +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "bumpalo" +version = "3.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" + +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + +[[package]] +name = "bytes" +version = "1.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" + +[[package]] +name = "camfinder" +version = "0.1.0" +dependencies = [ + "anyhow", + "dotenvy", + "tokio", + "tokio-postgres", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "chacha20" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if", + "cpufeatures", + "rand_core", +] + +[[package]] +name = "cmov" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" + +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + +[[package]] +name = "cpufeatures" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "ctutils" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d5515a3834141de9eafb9717ad39eea8247b5674e6066c404e8c4b365d2a29e" +dependencies = [ + "cmov", +] + +[[package]] +name = "digest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" +dependencies = [ + "block-buffer", + "const-oid", + "crypto-common", + "ctutils", +] + +[[package]] +name = "dotenvy" +version = "0.15.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "fallible-iterator" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4443176a9f2c162692bd3d352d745ef9413eec5782a80d8fd6f8a1ac692a07f7" + +[[package]] +name = "futures-channel" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" +dependencies = [ + "futures-core", + "futures-sink", +] + +[[package]] +name = "futures-core" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" + +[[package]] +name = "futures-sink" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893" + +[[package]] +name = "futures-task" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" + +[[package]] +name = "futures-util" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" +dependencies = [ + "futures-core", + "futures-sink", + "futures-task", + "pin-project-lite", + "slab", +] + +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "rand_core", +] + +[[package]] +name = "hmac" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" +dependencies = [ + "digest", +] + +[[package]] +name = "hybrid-array" +version = "0.4.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c" +dependencies = [ + "typenum", +] + +[[package]] +name = "js-sys" +version = "0.3.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102" +dependencies = [ + "cfg-if", + "futures-util", + "wasm-bindgen", +] + +[[package]] +name = "libc" +version = "0.2.186" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" + +[[package]] +name = "libredox" +version = "0.1.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f02ab6bace2054fb888a3c16f990117b579d14a3088e472d63c6011fa185c9d3" +dependencies = [ + "libc", +] + +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + +[[package]] +name = "log" +version = "0.4.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" + +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest", +] + +[[package]] +name = "memchr" +version = "2.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "88904434abc2901f197fe8cc55f0445e7ded921dba5911dad2e2b39b48e663c4" + +[[package]] +name = "mio" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" +dependencies = [ + "libc", + "wasi 0.11.1+wasi-snapshot-preview1", + "windows-sys", +] + +[[package]] +name = "objc2-core-foundation" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536" +dependencies = [ + "bitflags", +] + +[[package]] +name = "objc2-system-configuration" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7216bd11cbda54ccabcab84d523dc93b858ec75ecfb3a7d89513fa22464da396" +dependencies = [ + "objc2-core-foundation", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + +[[package]] +name = "phf" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c1562dc717473dbaa4c1f85a36410e03c047b2e7df7f45ee938fbef64ae7fadf" +dependencies = [ + "phf_shared", + "serde", +] + +[[package]] +name = "phf_shared" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e57fef6bc5981e38c2ce2d63bfa546861309f875b8a75f092d1d54ae2d64f266" +dependencies = [ + "siphasher", +] + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "postgres-protocol" +version = "0.6.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08808e3c483c46e999108051c78334f473d5adb59d78bb80a1268c7e6aa6c514" +dependencies = [ + "base64", + "byteorder", + "bytes", + "fallible-iterator", + "hmac", + "md-5", + "memchr", + "rand", + "sha2", + "stringprep", +] + +[[package]] +name = "postgres-types" +version = "0.2.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "851ca9db4932932d69f3ea811b1abe63087a0f740a47692619dd40d4899b68be" +dependencies = [ + "bytes", + "fallible-iterator", + "postgres-protocol", +] + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "rand" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2e8e8bcc7961af1fdac401278c6a831614941f6164ee3bf4ce61b7edb162207" +dependencies = [ + "chacha20", + "getrandom", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + +[[package]] +name = "rustversion" +version = "1.0.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" + +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "sha2" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "siphasher" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649" + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "smallvec" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" + +[[package]] +name = "socket2" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "stringprep" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b4df3d392d81bd458a8a621b8bffbd2302a12ffe288a9d931670948749463b1" +dependencies = [ + "unicode-bidi", + "unicode-normalization", + "unicode-properties", +] + +[[package]] +name = "syn" +version = "2.0.118" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "tinyvec" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3e61e67053d25a4e82c844e8424039d9745781b3fc4f32b8d55ed50f5f667ef3" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" + +[[package]] +name = "tokio" +version = "1.52.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "signal-hook-registry", + "socket2", + "tokio-macros", + "windows-sys", +] + +[[package]] +name = "tokio-macros" +version = "2.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tokio-postgres" +version = "0.7.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a528f7d280f6d5b9cd149635c8705b0dd049754bc67d81d31fa25169a93809d3" +dependencies = [ + "async-trait", + "byteorder", + "bytes", + "fallible-iterator", + "futures-channel", + "futures-util", + "log", + "parking_lot", + "percent-encoding", + "phf", + "pin-project-lite", + "postgres-protocol", + "postgres-types", + "rand", + "socket2", + "tokio", + "tokio-util", + "whoami", +] + +[[package]] +name = "tokio-util" +version = "0.7.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + +[[package]] +name = "unicode-bidi" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c1cb5db39152898a79168971543b1cb5020dff7fe43c8dc468b0885f5e29df5" + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "unicode-normalization" +version = "0.1.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5fd4f6878c9cb28d874b009da9e8d183b5abc80117c40bbd187a1fde336be6e8" +dependencies = [ + "tinyvec", +] + +[[package]] +name = "unicode-properties" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d" + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "wasi" +version = "0.14.7+wasi-0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "883478de20367e224c0090af9cf5f9fa85bed63a95c1abf3afc5c083ebc06e8c" +dependencies = [ + "wasip2", +] + +[[package]] +name = "wasip2" +version = "1.0.4+wasi-0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487" +dependencies = [ + "wit-bindgen", +] + +[[package]] +name = "wasite" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66fe902b4a6b8028a753d5424909b764ccf79b7a209eac9bf97e59cda9f71a42" +dependencies = [ + "wasi 0.14.7+wasi-0.2.4", +] + +[[package]] +name = "wasm-bindgen" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "web-sys" +version = "0.3.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8622dcb61c0bcc9fffa6938bed81210af2da9a7e4a1a834b2e37a59b6dfb6141" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "whoami" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "998767ef88740d1f5b0682a9c53c24431453923962269c2db68ee43788c5a40d" +dependencies = [ + "libc", + "libredox", + "objc2-system-configuration", + "wasite", + "web-sys", +] + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "wit-bindgen" +version = "0.57.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..ed29378 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,10 @@ +[package] +name = "camfinder" +version = "0.1.0" +edition = "2021" + +[dependencies] +anyhow = "1" +dotenvy = "0.15" +tokio = { version = "1", features = ["io-util", "macros", "net", "rt-multi-thread", "signal", "time"] } +tokio-postgres = "0.7" diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..8c39ea9 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,17 @@ +FROM rust:1.78-bookworm AS builder +WORKDIR /app + +COPY Cargo.toml Cargo.lock ./ +COPY src ./src + +RUN cargo build --release + +FROM debian:bookworm-slim +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app +COPY --from=builder /app/target/release/camfinder /usr/local/bin/camfinder + +ENTRYPOINT ["camfinder"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..73eea93 --- /dev/null +++ b/README.md @@ -0,0 +1,57 @@ +# CamFinder + +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: + +- `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. + +O progresso é salvo por bloco, então o scan pode ser retomado sem repetir o que já foi concluído. + +## Uso + +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`. + +## Docker Compose + +O `docker-compose.yml` já sobe: + +- `postgres` para persistência. +- `coordinator` exposto no host pela porta `666`. +- `worker` conectado ao coordenador na rede interna do compose. + +Para escalar workers: + +- `docker compose up --scale worker=3` + +Para workers de outras máquinas, use `vps.valmo.dev:666` como endereço do coordenador. + +## Scripts + +- `cargo build` +- `cargo run` +- `npm run scan` +- `npm run coord` +- `npm run worker` + +## Variáveis úteis + +- `DATABASE_URL` +- `SCAN_CIDRS` +- `SCAN_CIDR` +- `SCAN_CIDR_FILE` +- `SCAN_CONCURRENCY` +- `SCAN_CHUNK_SIZE` +- `SCAN_TIMEOUT_MS` +- `SCAN_MODE` +- `COORDINATOR_BIND` +- `COORDINATOR_ADDR` +- `WORKER_NAME` + +## Restrições + +O projeto aceita somente redes privadas `10.0.0.0/8`, `172.16.0.0/12` e `192.168.0.0/16`. diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..d5ea094 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,48 @@ +services: + 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: ["coordinator"] + environment: + DATABASE_URL: postgres://camfinder:camfinder@postgres:5432/camfinder + SCAN_CIDRS: 192.168.15.0/24 + SCAN_PORT: 554 + SCAN_TIMEOUT_MS: 1200 + SCAN_CONCURRENCY: 16 + SCAN_CHUNK_SIZE: 64 + COORDINATOR_BIND: 0.0.0.0:666 + depends_on: + postgres: + condition: service_healthy + ports: + - "666:666" + + worker: + build: . + command: ["worker"] + environment: + COORDINATOR_ADDR: coordinator:666 + SCAN_PORT: 554 + SCAN_TIMEOUT_MS: 1200 + SCAN_CONCURRENCY: 16 + WORKER_NAME: worker + depends_on: + coordinator: + condition: service_started + +volumes: + camfinder-postgres: diff --git a/ips.md b/ips.md new file mode 100644 index 0000000..1a4df9a --- /dev/null +++ b/ips.md @@ -0,0 +1,31 @@ +# Faixas IPv4 + +| Faixa IPv4 | Intervalo completo | Nome | Roteável na internet? | Observação | +|---|---|---|---|---| +| `0.0.0.0/8` | `0.0.0.0` até `0.255.255.255` | Especial | Não | Endereço não especificado e usos reservados | +| `1.0.0.0/8` até `9.255.255.255/8` | `1.0.0.0` até `9.255.255.255` | Público | Sim | Inclui IPs como `1.2.3.4` | +| `10.0.0.0/8` | `10.0.0.0` até `10.255.255.255` | Privado | Não | Redes internas grandes | +| `11.0.0.0/8` até `100.63.255.255` | `11.0.0.0` até `100.63.255.255` | Público | Sim | Faixa pública, exceto blocos reservados menores | +| `100.64.0.0/10` | `100.64.0.0` até `100.127.255.255` | CGNAT | Não | Compartilhado por operadoras | +| `100.128.0.0/9` até `126.255.255.255` | `100.128.0.0` até `126.255.255.255` | Público | Sim | Faixa pública, exceto blocos reservados menores | +| `127.0.0.0/8` | `127.0.0.0` até `127.255.255.255` | Loopback | Não | A própria máquina | +| `128.0.0.0/2` com exceções | `128.0.0.0` até `223.255.255.255` | Público | Sim | Faixa pública geral, com alguns blocos especiais reservados dentro dela | +| `169.254.0.0/16` | `169.254.0.0` até `169.254.255.255` | Link-local | Não | IP automático sem DHCP | +| `172.16.0.0/12` | `172.16.0.0` até `172.31.255.255` | Privado | Não | Redes internas médias | +| `192.0.0.0/24` e blocos próximos | `192.0.0.0` até `192.0.0.255` | Reservado | Não | Usos especiais diversos | +| `192.0.2.0/24` | `192.0.2.0` até `192.0.2.255` | Documentação | Não | Exemplos e manuais | +| `192.88.99.0/24` | `192.88.99.0` até `192.88.99.255` | Reservado | Não | Uso especial histórico | +| `192.168.0.0/16` | `192.168.0.0` até `192.168.255.255` | Privado | Não | Casas, escritórios, laboratórios | +| `198.18.0.0/15` | `198.18.0.0` até `198.19.255.255` | Teste | Não | Benchmarks e testes de rede | +| `198.51.100.0/24` | `198.51.100.0` até `198.51.100.255` | Documentação | Não | Exemplos e manuais | +| `203.0.113.0/24` | `203.0.113.0` até `203.0.113.255` | Documentação | Não | Exemplos e manuais | +| `224.0.0.0/4` | `224.0.0.0` até `239.255.255.255` | Multicast | Não | Um envio para vários destinos | +| `240.0.0.0/4` | `240.0.0.0` até `255.255.255.254` | Reservado | Não | Reservado para uso futuro e especial | +| `255.255.255.255` | `255.255.255.255` | Broadcast limitado | Não | Todos os hosts da rede local | + +## Leitura prática + +- `1.2.3.4` entra na faixa pública roteável. +- `192.168.15.8` entra na faixa privada e não é roteável na internet. +- `127.0.0.1` é loopback. +- `100.72.4.1` é CGNAT, normalmente não é roteável diretamente. diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..652cb97 --- /dev/null +++ b/src/config.rs @@ -0,0 +1,352 @@ +use anyhow::{anyhow, Context, Result}; +use std::{ + env, fs, + net::SocketAddr, + path::{Path, PathBuf}, +}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Mode { + Local, + Coordinator, + Worker, +} + +#[derive(Debug, Clone)] +pub struct Config { + pub mode: Mode, + pub cidrs: Vec, + pub port: u16, + pub timeout_ms: u64, + pub concurrency: usize, + pub chunk_size: usize, + pub database_url: String, + pub coordinator_bind: SocketAddr, + pub coordinator_addr: SocketAddr, + pub worker_name: String, +} + +impl Config { + pub fn from_env_and_args() -> Result { + let mut args = env::args().skip(1).collect::>(); + let mode = parse_mode(&mut args)?; + let cli = CliArgs::from_args(&args)?; + + let cidrs = load_cidrs(cli.cidr_file.as_deref(), cli.cidrs.as_deref())?; + Ok(Self { + mode, + cidrs, + port: cli + .port + .or_else(|| parse_u16("SCAN_PORT", 554).ok()) + .unwrap_or(554), + timeout_ms: cli + .timeout_ms + .or_else(|| parse_u64("SCAN_TIMEOUT_MS", 1200).ok()) + .unwrap_or(1200), + concurrency: cli + .concurrency + .or_else(|| parse_usize("SCAN_CONCURRENCY", default_concurrency()).ok()) + .unwrap_or_else(default_concurrency), + chunk_size: cli + .chunk_size + .or_else(|| parse_usize("SCAN_CHUNK_SIZE", default_chunk_size()).ok()) + .unwrap_or_else(default_chunk_size), + database_url: cli.database_url.unwrap_or_else(database_url_from_env), + coordinator_bind: cli + .coordinator_bind + .or_else(|| parse_socket_addr("COORDINATOR_BIND", "0.0.0.0:666").ok()) + .unwrap_or_else(|| "0.0.0.0:666".parse().expect("valid default bind address")), + coordinator_addr: cli + .coordinator_addr + .or_else(|| parse_socket_addr("COORDINATOR_ADDR", "127.0.0.1:666").ok()) + .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), + }) + } +} + +#[derive(Debug, Default)] +struct CliArgs { + cidrs: Option, + cidr_file: Option, + database_url: Option, + coordinator_bind: Option, + coordinator_addr: Option, + worker_name: Option, + port: Option, + timeout_ms: Option, + concurrency: Option, + chunk_size: Option, +} + +impl CliArgs { + fn from_args(args: &[String]) -> Result { + let mut cli = Self::default(); + let mut index = 0; + + while index < args.len() { + let arg = &args[index]; + match arg.as_str() { + "--cidrs" => { + cli.cidrs = Some(next_value(args, &mut index, "--cidrs")?); + } + "--cidr-file" => { + cli.cidr_file = + Some(PathBuf::from(next_value(args, &mut index, "--cidr-file")?)); + } + "--database" | "--database-url" => { + cli.database_url = Some(next_value(args, &mut index, arg)?); + } + "--bind" => { + cli.coordinator_bind = Some(parse_socket_addr_literal(&next_value( + args, &mut index, "--bind", + )?)?); + } + "--connect" => { + cli.coordinator_addr = Some(parse_socket_addr_literal(&next_value( + args, + &mut index, + "--connect", + )?)?); + } + "--name" => { + cli.worker_name = Some(next_value(args, &mut index, "--name")?); + } + "--port" => { + cli.port = Some(parse_u16_literal(&next_value(args, &mut index, "--port")?)?); + } + "--timeout-ms" => { + cli.timeout_ms = Some(parse_u64_literal(&next_value( + args, + &mut index, + "--timeout-ms", + )?)?); + } + "--concurrency" => { + cli.concurrency = Some(parse_usize_literal(&next_value( + args, + &mut index, + "--concurrency", + )?)?); + } + "--chunk-size" => { + cli.chunk_size = Some(parse_usize_literal(&next_value( + args, + &mut index, + "--chunk-size", + )?)?); + } + _ if arg.starts_with("--") => { + return Err(anyhow!("unknown argument: {arg}")); + } + _ => { + return Err(anyhow!("unexpected positional argument: {arg}")); + } + } + + index += 1; + } + + Ok(cli) + } +} + +fn parse_mode(args: &mut Vec) -> Result { + if let Some(first) = args.first() { + if !first.starts_with('-') { + let mode = parse_mode_literal(first)?; + args.remove(0); + return Ok(mode); + } + } + + match env::var("SCAN_MODE") { + Ok(value) => parse_mode_literal(value.trim()), + Err(_) => Ok(Mode::Local), + } +} + +fn parse_mode_literal(value: &str) -> Result { + match value.trim().to_ascii_lowercase().as_str() { + "local" | "scan" => Ok(Mode::Local), + "coordinator" | "coord" | "server" => Ok(Mode::Coordinator), + "worker" => Ok(Mode::Worker), + other => Err(anyhow!( + "modo inválido: {other}. Use local, coordinator ou worker" + )), + } +} + +fn load_cidrs(file: Option<&Path>, inline: Option<&str>) -> Result> { + if let Some(inline) = inline { + let cidrs = parse_cidr_list(inline); + if !cidrs.is_empty() { + return Ok(cidrs); + } + } + + if let Some(file) = file { + let content = fs::read_to_string(file) + .with_context(|| format!("failed to read CIDR file {}", file.display()))?; + let cidrs = parse_cidr_list(&content); + if !cidrs.is_empty() { + return Ok(cidrs); + } + 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") { + let path = path.trim(); + if !path.is_empty() { + let content = fs::read_to_string(path) + .with_context(|| format!("failed to read CIDR file {path}"))?; + let cidrs = parse_cidr_list(&content); + if !cidrs.is_empty() { + return Ok(cidrs); + } + return Err(anyhow!("arquivo de CIDRs vazio: {path}")); + } + } + + if let Ok(value) = env::var("SCAN_CIDR") { + let value = value.trim(); + if !value.is_empty() { + let cidrs = parse_cidr_list(value); + if !cidrs.is_empty() { + return Ok(cidrs); + } + } + } + + Ok(Vec::new()) +} + +fn parse_cidr_list(input: &str) -> Vec { + input + .lines() + .flat_map(|line| line.split('#').next()) + .flat_map(|line| line.split(|c: char| c == ',' || c.is_whitespace())) + .map(str::trim) + .filter(|entry| !entry.is_empty()) + .map(ToString::to_string) + .collect() +} + +fn database_url_from_env() -> String { + let value = read_env("DATABASE_URL", ""); + if value.is_empty() { + "postgres://postgres:postgres@127.0.0.1:5432/camfinder".to_string() + } else { + value + } +} + +fn default_worker_name() -> String { + env::var("WORKER_NAME") + .ok() + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) + .unwrap_or_else(|| { + let host = env::var("HOSTNAME").unwrap_or_else(|_| "worker".to_string()); + format!("{}-{}", host.trim(), std::process::id()) + }) +} + +fn read_env(name: &str, fallback: &str) -> String { + env::var(name) + .map(|value| value.trim().to_string()) + .unwrap_or_else(|_| fallback.to_string()) +} + +fn parse_socket_addr(name: &str, fallback: &str) -> Result { + parse_socket_addr_literal(&read_env(name, fallback)) +} + +fn parse_socket_addr_literal(value: &str) -> Result { + value + .trim() + .parse() + .map_err(|_| anyhow!("invalid socket address: {value}")) +} + +fn parse_u16(name: &str, fallback: u16) -> Result { + parse_number(name, fallback) +} + +fn parse_u64(name: &str, fallback: u64) -> Result { + parse_number(name, fallback) +} + +fn parse_usize(name: &str, fallback: usize) -> Result { + parse_number(name, fallback) +} + +fn parse_u16_literal(value: &str) -> Result { + value + .trim() + .parse() + .map_err(|_| anyhow!("invalid u16 value: {value}")) +} + +fn parse_u64_literal(value: &str) -> Result { + value + .trim() + .parse() + .map_err(|_| anyhow!("invalid u64 value: {value}")) +} + +fn parse_usize_literal(value: &str) -> Result { + value + .trim() + .parse() + .map_err(|_| anyhow!("invalid usize value: {value}")) +} + +fn next_value(args: &[String], index: &mut usize, flag: &str) -> Result { + let value_index = *index + 1; + let value = args + .get(value_index) + .ok_or_else(|| anyhow!("missing value for {flag}"))? + .clone(); + *index = value_index; + Ok(value) +} + +fn default_concurrency() -> usize { + let cpus = std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(4); + (cpus.saturating_mul(2)).clamp(4, 64) +} + +fn default_chunk_size() -> usize { + default_concurrency().saturating_mul(2).clamp(32, 256) +} + +fn parse_number(name: &str, fallback: T) -> Result +where + T: std::str::FromStr + Copy, +{ + match env::var(name) { + Ok(value) => value + .trim() + .parse::() + .map_err(|_| anyhow!("invalid value for {name}: {value}")), + Err(_) => Ok(fallback), + } +} diff --git a/src/coordinator.rs b/src/coordinator.rs new file mode 100644 index 0000000..b1012f1 --- /dev/null +++ b/src/coordinator.rs @@ -0,0 +1,398 @@ +use crate::{ + config::Config, + db::Database, + models::{ChunkRecord, PendingRange}, + scan::{build_pending_ranges, load_targets}, + terminal::{ScreenState, TerminalUi}, +}; +use anyhow::{anyhow, Context, Result}; +use std::{ + collections::VecDeque, + net::Ipv4Addr, + sync::{ + atomic::{AtomicBool, AtomicUsize, Ordering}, + Arc, Mutex, + }, + time::Instant, +}; +use tokio::{ + io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, + net::{TcpListener, TcpStream}, + task::JoinSet, +}; + +pub async fn run_coordinator( + config: Config, + database: Arc, + ui: Arc, + interrupted: Arc, +) -> Result<()> { + let targets = Arc::new(load_targets(&config.cidrs)?); + let target_count = targets.len(); + let chunk_records = database.load_chunk_records().await?; + let (tested_before_run, _open_before_run) = database.summarize_chunks().await?; + let pending_ranges = build_pending_ranges(target_count, config.chunk_size, &chunk_records); + let pending_count: usize = pending_ranges.iter().map(|range| range.len()).sum(); + + if ui.is_enabled() { + ui.print_banner(); + ui.render(&ScreenState { + current_ip: if pending_count > 0 { + targets[pending_ranges[0].start].to_string() + } else { + "concluído".to_string() + }, + tested: tested_before_run, + total: target_count, + }); + } else { + println!( + "Coordinator ready on {} for {} hosts across {} CIDRs. Pending {}, already tested {}.", + config.coordinator_bind, + target_count, + config.cidrs.len(), + pending_count, + tested_before_run + ); + } + + if pending_ranges.is_empty() { + if ui.is_enabled() { + ui.render(&ScreenState { + current_ip: "concluído".to_string(), + tested: target_count, + total: target_count, + }); + ui.finish(); + } + return Ok(()); + } + + let listener = TcpListener::bind(config.coordinator_bind) + .await + .with_context(|| format!("failed to bind {}", config.coordinator_bind))?; + let queue = Arc::new(Mutex::new(VecDeque::from(pending_ranges))); + let tested_this_run = Arc::new(AtomicUsize::new(0)); + let open_this_run = Arc::new(AtomicUsize::new(0)); + let start = Instant::now(); + let mut handlers = JoinSet::new(); + + loop { + if interrupted.load(Ordering::SeqCst) { + break; + } + + tokio::select! { + accept_result = listener.accept() => { + let (stream, peer) = accept_result?; + let queue = Arc::clone(&queue); + let targets = Arc::clone(&targets); + let database = Arc::clone(&database); + let ui = Arc::clone(&ui); + let interrupted = Arc::clone(&interrupted); + let tested_this_run = Arc::clone(&tested_this_run); + let open_this_run = Arc::clone(&open_this_run); + let config = config.clone(); + + handlers.spawn(async move { + handle_worker( + stream, + peer, + queue, + targets, + database, + ui, + interrupted, + tested_before_run, + tested_this_run, + open_this_run, + config, + ) + .await + }); + } + maybe_finished = handlers.join_next(), if !handlers.is_empty() => { + if let Some(join_result) = maybe_finished { + join_result??; + } + } + } + + if queue.lock().expect("queue mutex poisoned").is_empty() && handlers.is_empty() { + break; + } + } + + while let Some(join_result) = handlers.join_next().await { + join_result??; + } + + let duration = start.elapsed().as_secs_f64(); + let total_open_count = database.count_open_endpoints(config.port).await?; + + if interrupted.load(Ordering::SeqCst) { + if ui.is_enabled() { + ui.render(&ScreenState { + current_ip: "interrompido".to_string(), + tested: tested_before_run + tested_this_run.load(Ordering::SeqCst), + total: target_count, + }); + ui.finish(); + } else { + println!("Interrupted after {:.2}s.", duration); + println!( + "Hosts tested in this run: {}", + tested_this_run.load(Ordering::SeqCst) + ); + println!( + "Open endpoints found: {}", + open_this_run.load(Ordering::SeqCst) + ); + } + return Ok(()); + } + + if ui.is_enabled() { + ui.render(&ScreenState { + current_ip: "concluído".to_string(), + tested: target_count, + total: target_count, + }); + ui.finish(); + } else { + println!("Coordinator finished in {:.2}s", duration); + println!( + "Hosts tested in this run: {}", + tested_this_run.load(Ordering::SeqCst) + ); + println!( + "Open endpoints found in this run: {}", + open_this_run.load(Ordering::SeqCst) + ); + println!("Total open endpoints in DB: {}", total_open_count); + } + + Ok(()) +} + +async fn handle_worker( + stream: TcpStream, + peer: std::net::SocketAddr, + queue: Arc>>, + targets: Arc>, + database: Arc, + ui: Arc, + interrupted: Arc, + tested_before_run: usize, + tested_this_run: Arc, + open_this_run: Arc, + config: Config, +) -> Result<()> { + let (read_half, mut write_half) = stream.into_split(); + let mut lines = BufReader::new(read_half).lines(); + + let worker_name = match lines.next_line().await? { + Some(line) if line.starts_with("HELLO ") => line[6..].trim().to_string(), + Some(line) => { + return Err(anyhow!("unexpected worker handshake from {peer}: {line}")); + } + None => return Ok(()), + }; + + if !worker_name.is_empty() { + println!("worker connected from {peer}: {worker_name}"); + } else { + println!("worker connected from {peer}"); + } + + loop { + if interrupted.load(Ordering::SeqCst) { + let _ = write_half.write_all(b"SHUTDOWN\n").await; + break; + } + + let Some(range) = next_range(&queue) else { + let _ = write_half.write_all(b"DONE\n").await; + break; + }; + + if let Err(err) = process_range( + &mut write_half, + &mut lines, + range, + &targets, + &database, + &ui, + tested_before_run, + &tested_this_run, + &open_this_run, + &config, + ) + .await + { + requeue_range(&queue, range); + return Err(err); + } + } + + Ok(()) +} + +async fn process_range( + write_half: &mut tokio::net::tcp::OwnedWriteHalf, + lines: &mut tokio::io::Lines>, + range: PendingRange, + targets: &Arc>, + database: &Arc, + ui: &Arc, + tested_before_run: usize, + tested_this_run: &Arc, + open_this_run: &Arc, + config: &Config, +) -> Result<()> { + let ips = targets[range.start..range.end].to_vec(); + let job_line = format!( + "JOB {} {} {} {} {}\n", + range.start, + range.end, + config.port, + config.timeout_ms, + format_ips(&ips) + ); + write_half.write_all(job_line.as_bytes()).await?; + + let Some(result_line) = lines.next_line().await? else { + return Err(anyhow!("worker disconnected before returning job")); + }; + + let result = parse_result_line(&result_line)?; + if result.chunk_start != range.start || result.chunk_end != range.end { + return Err(anyhow!( + "worker returned mismatched range: expected {}..{}, got {}..{}", + range.start, + range.end, + result.chunk_start, + result.chunk_end + )); + } + + for ip in result.open_ips { + database.upsert_open_endpoint(ip, config.port).await?; + } + + let previous = database.load_chunk_record(range.chunk_start).await?; + let tested_before = previous + .as_ref() + .map(|record| record.tested_count) + .unwrap_or(0); + let open_before = previous + .as_ref() + .map(|record| record.open_count) + .unwrap_or(0); + let tested_count = tested_before + result.tested; + let open_count = open_before + result.open_count; + database + .save_chunk_record(&ChunkRecord { + chunk_start: range.chunk_start, + chunk_end: range.chunk_end, + tested_count, + open_count, + is_complete: result.tested >= range.len(), + }) + .await?; + + let tested_now = tested_this_run.fetch_add(result.tested, Ordering::SeqCst) + result.tested; + open_this_run.fetch_add(result.open_count, Ordering::SeqCst); + + if ui.is_enabled() { + let current_ip = ips + .last() + .map(|ip| ip.to_string()) + .unwrap_or_else(|| "concluído".to_string()); + ui.render(&ScreenState { + current_ip, + tested: tested_before_run + tested_now, + total: targets.len(), + }); + } + + Ok(()) +} + +fn requeue_range(queue: &Arc>>, range: PendingRange) { + queue + .lock() + .expect("queue mutex poisoned") + .push_front(range); +} + +fn next_range(queue: &Arc>>) -> Option { + queue.lock().expect("queue mutex poisoned").pop_front() +} + +struct JobResult { + chunk_start: usize, + chunk_end: usize, + tested: usize, + open_count: usize, + open_ips: Vec, +} + +fn parse_result_line(line: &str) -> Result { + let mut parts = line.splitn(6, ' '); + let kind = parts.next().unwrap_or(""); + if kind != "RESULT" { + return Err(anyhow!("unexpected worker message: {line}")); + } + + let chunk_start = parse_usize_part(parts.next(), "chunk_start", line)?; + let chunk_end = parse_usize_part(parts.next(), "chunk_end", line)?; + let tested = parse_usize_part(parts.next(), "tested", line)?; + let open_count = parse_usize_part(parts.next(), "open_count", line)?; + let ips_part = parts.next().unwrap_or(""); + let open_ips = if ips_part.trim().is_empty() { + Vec::new() + } else { + ips_part + .split(',') + .map(str::trim) + .filter(|entry| !entry.is_empty()) + .map(|entry| { + entry + .parse::() + .with_context(|| format!("invalid IP in worker result: {entry}")) + }) + .collect::>>()? + }; + + if open_ips.len() != open_count { + return Err(anyhow!( + "open_count mismatch in worker result: declared {}, received {}", + open_count, + open_ips.len() + )); + } + + Ok(JobResult { + chunk_start, + chunk_end, + tested, + open_count, + open_ips, + }) +} + +fn parse_usize_part(part: Option<&str>, name: &str, line: &str) -> Result { + let value = part.ok_or_else(|| anyhow!("missing {name} in worker message: {line}"))?; + value + .trim() + .parse() + .map_err(|_| anyhow!("invalid {name} in worker message: {line}")) +} + +fn format_ips(ips: &[Ipv4Addr]) -> String { + ips.iter() + .map(ToString::to_string) + .collect::>() + .join(",") +} diff --git a/src/db.rs b/src/db.rs new file mode 100644 index 0000000..51300b5 --- /dev/null +++ b/src/db.rs @@ -0,0 +1,218 @@ +use crate::models::ChunkRecord; +use anyhow::{Context, Result}; +use std::convert::TryFrom; +use tokio_postgres::{Client, NoTls}; + +pub struct Database { + client: Client, +} + +impl Database { + pub async fn open(database_url: impl AsRef) -> Result { + let database_url = database_url.as_ref().trim(); + let (client, connection) = tokio_postgres::connect(database_url, NoTls) + .await + .with_context(|| format!("failed to connect to postgres at {database_url}"))?; + + tokio::spawn(async move { + if let Err(err) = connection.await { + eprintln!("postgres connection error: {err}"); + } + }); + + let db = Self { client }; + db.init().await?; + Ok(db) + } + + async fn init(&self) -> Result<()> { + self.client + .batch_execute( + r#" + CREATE TABLE IF NOT EXISTS open_rtsp_endpoint ( + id BIGSERIAL PRIMARY KEY, + ip INET NOT NULL UNIQUE, + port INTEGER NOT NULL, + scanned_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 open_rtsp_endpoint_port_idx + ON open_rtsp_endpoint (port); + + CREATE TABLE IF NOT EXISTS scan_chunks ( + chunk_start BIGINT PRIMARY KEY NOT NULL, + chunk_end BIGINT NOT NULL, + tested_count BIGINT NOT NULL DEFAULT 0, + open_count BIGINT NOT NULL DEFAULT 0, + is_complete BOOLEAN NOT NULL DEFAULT FALSE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + "#, + ) + .await + .context("failed to initialize postgres schema")?; + Ok(()) + } + + pub async fn load_chunk_records(&self) -> Result> { + let rows = self + .client + .query( + r#" + SELECT chunk_start, chunk_end, tested_count, open_count, is_complete + FROM scan_chunks + ORDER BY chunk_start ASC + "#, + &[], + ) + .await + .context("failed to load scan chunks")?; + + rows.into_iter().map(row_to_chunk_record).collect() + } + + pub async fn load_chunk_record(&self, chunk_start: usize) -> Result> { + let chunk_start = usize_to_i64(chunk_start, "chunk_start")?; + let row = self + .client + .query_opt( + r#" + SELECT chunk_start, chunk_end, tested_count, open_count, is_complete + FROM scan_chunks + WHERE chunk_start = $1 + "#, + &[&chunk_start], + ) + .await + .context("failed to load scan chunk")?; + + row.map(row_to_chunk_record).transpose() + } + + pub async fn save_chunk_record(&self, record: &ChunkRecord) -> Result<()> { + let chunk_start = usize_to_i64(record.chunk_start, "chunk_start")?; + let chunk_end = usize_to_i64(record.chunk_end, "chunk_end")?; + let tested_count = usize_to_i64(record.tested_count, "tested_count")?; + let open_count = usize_to_i64(record.open_count, "open_count")?; + + self.client + .execute( + r#" + INSERT INTO scan_chunks ( + chunk_start, + chunk_end, + tested_count, + open_count, + is_complete, + created_at, + updated_at + ) + VALUES ($1, $2, $3, $4, $5, NOW(), NOW()) + ON CONFLICT (chunk_start) DO UPDATE SET + chunk_end = EXCLUDED.chunk_end, + tested_count = EXCLUDED.tested_count, + open_count = EXCLUDED.open_count, + is_complete = EXCLUDED.is_complete, + updated_at = NOW() + "#, + &[ + &chunk_start, + &chunk_end, + &tested_count, + &open_count, + &record.is_complete, + ], + ) + .await + .context("failed to save scan chunk")?; + + Ok(()) + } + + pub async fn upsert_open_endpoint(&self, ip: std::net::Ipv4Addr, port: u16) -> Result<()> { + let ip = ip.to_string(); + let port = i32::from(port); + + self.client + .execute( + r#" + INSERT INTO open_rtsp_endpoint ( + ip, + port, + scanned_at, + last_seen_at, + created_at, + updated_at + ) + VALUES ($1::inet, $2, NOW(), NOW(), NOW(), NOW()) + ON CONFLICT (ip) DO UPDATE SET + port = EXCLUDED.port, + scanned_at = NOW(), + last_seen_at = NOW(), + updated_at = NOW() + "#, + &[&ip, &port], + ) + .await + .context("failed to upsert open endpoint")?; + + Ok(()) + } + + pub async fn count_open_endpoints(&self, port: u16) -> Result { + let port = i32::from(port); + let count: i64 = self + .client + .query_one( + "SELECT COUNT(*)::BIGINT FROM open_rtsp_endpoint WHERE port = $1", + &[&port], + ) + .await + .context("failed to count open endpoints")? + .get(0); + + i64_to_usize(count, "open endpoint count") + } + + pub async fn summarize_chunks(&self) -> Result<(usize, usize)> { + let row = self + .client + .query_one( + r#" + SELECT + COALESCE(SUM(tested_count), 0)::BIGINT, + COALESCE(SUM(open_count), 0)::BIGINT + FROM scan_chunks + "#, + &[], + ) + .await + .context("failed to summarize scan chunks")?; + + let tested = i64_to_usize(row.get::<_, i64>(0), "tested count")?; + let open = i64_to_usize(row.get::<_, i64>(1), "open count")?; + Ok((tested, open)) + } +} + +fn row_to_chunk_record(row: tokio_postgres::Row) -> Result { + Ok(ChunkRecord { + chunk_start: i64_to_usize(row.get::<_, i64>(0), "chunk_start")?, + chunk_end: i64_to_usize(row.get::<_, i64>(1), "chunk_end")?, + tested_count: i64_to_usize(row.get::<_, i64>(2), "tested_count")?, + open_count: i64_to_usize(row.get::<_, i64>(3), "open_count")?, + is_complete: row.get::<_, bool>(4), + }) +} + +fn i64_to_usize(value: i64, field: &str) -> Result { + usize::try_from(value).with_context(|| format!("{field} out of range: {value}")) +} + +fn usize_to_i64(value: usize, field: &str) -> Result { + i64::try_from(value).with_context(|| format!("{field} out of range: {value}")) +} diff --git a/src/ip.rs b/src/ip.rs new file mode 100644 index 0000000..eb98dc8 --- /dev/null +++ b/src/ip.rs @@ -0,0 +1,103 @@ +use anyhow::{anyhow, Context, Result}; +use std::net::Ipv4Addr; + +pub fn cidr_to_hosts(cidr: &str) -> Result> { + let (network, prefix) = cidr + .split_once('/') + .ok_or_else(|| anyhow!("CIDR inválido: {cidr}"))?; + + let prefix: u8 = prefix + .parse() + .with_context(|| format!("máscara de sub-rede inválida: {prefix}"))?; + if prefix > 32 { + return Err(anyhow!("máscara de sub-rede inválida: {prefix}")); + } + + let network: Ipv4Addr = network + .parse() + .with_context(|| format!("endereço IP inválido: {network}"))?; + + let network_u32 = u32::from(network); + let mask = if prefix == 0 { + 0 + } else { + u32::MAX << (32 - prefix) + }; + let network_base = network_u32 & mask; + let broadcast = if prefix == 32 { + network_base + } else { + network_base | !mask + }; + + let start = if prefix >= 31 { + network_base + } else { + network_base + 1 + }; + let end = if prefix >= 31 { + broadcast + } else { + broadcast - 1 + }; + + let mut hosts = Vec::new(); + for ip_u32 in start..=end { + hosts.push(Ipv4Addr::from(ip_u32)); + } + + Ok(hosts) +} + +#[allow(dead_code)] +pub fn is_routable_ipv4(ip: Ipv4Addr) -> bool { + let [a, b, c, _d] = ip.octets(); + + if a == 0 { + return false; + } + if a == 10 { + return false; + } + if a == 100 && (64..=127).contains(&b) { + return false; + } + if a == 127 { + return false; + } + if a == 169 && b == 254 { + return false; + } + if a == 172 && (16..=31).contains(&b) { + return false; + } + if a == 192 && b == 0 && c == 0 { + return false; + } + if a == 192 && b == 0 && c == 2 { + return false; + } + if a == 192 && b == 88 && c == 99 { + return false; + } + if a == 192 && b == 168 { + return false; + } + if a == 198 && (b == 18 || b == 19) { + return false; + } + if a == 198 && b == 51 && c == 100 { + return false; + } + if a == 203 && b == 0 && c == 113 { + return false; + } + if (224..=239).contains(&a) { + return false; + } + if a >= 240 { + return false; + } + + true +} diff --git a/src/main.rs b/src/main.rs new file mode 100644 index 0000000..5fad0b8 --- /dev/null +++ b/src/main.rs @@ -0,0 +1,124 @@ +mod config; +mod coordinator; +mod db; +mod ip; +mod models; +mod scan; +mod terminal; +mod worker; + +use anyhow::Result; +use config::{Config, Mode}; +use db::Database; +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; +use terminal::{stdout_is_terminal, TerminalUi}; + +#[tokio::main] +async fn main() -> Result<()> { + dotenvy::dotenv().ok(); + + 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 ui = Arc::new(TerminalUi::new(stdout_is_terminal())); + let interrupted = Arc::new(AtomicBool::new(false)); + + { + let interrupted = Arc::clone(&interrupted); + tokio::spawn(async move { + let _ = tokio::signal::ctrl_c().await; + interrupted.store(true, Ordering::SeqCst); + }); + } + + let result = match config.mode { + Mode::Local => scan::run_local_scan(config, database, ui, interrupted).await, + Mode::Coordinator => coordinator::run_coordinator(config, database, ui, interrupted).await, + Mode::Worker => worker::run_worker(config, ui).await, + }; + + if let Err(err) = result { + eprintln!("error: {err:#}"); + return Err(err); + } + + Ok(()) +} + +fn assert_local_cidrs(cidrs: &[String]) -> Result<()> { + if cidrs.is_empty() { + return Err(anyhow::anyhow!( + "nenhum CIDR foi informado. Use SCAN_CIDRS, SCAN_CIDR_FILE ou --cidrs/--cidr-file" + )); + } + + for cidr in cidrs { + assert_local_cidr_inner(cidr)?; + } + + Ok(()) +} + +fn assert_local_cidr_inner(cidr: &str) -> Result<()> { + let (network, prefix) = cidr + .split_once('/') + .ok_or_else(|| anyhow::anyhow!("CIDR inválido: {cidr}"))?; + + let prefix: u8 = prefix + .parse() + .map_err(|_| anyhow::anyhow!("máscara de sub-rede inválida: {prefix}"))?; + if prefix > 32 { + return Err(anyhow::anyhow!("máscara de sub-rede inválida: {prefix}")); + } + + let ip: std::net::Ipv4Addr = network + .parse() + .map_err(|_| anyhow::anyhow!("endereço IP inválido: {network}"))?; + + let start = u32::from(ip) & cidr_mask(prefix); + let end = if prefix == 32 { + start + } else { + start | !cidr_mask(prefix) + }; + + let private_ranges = [ + ( + 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(()) +} + +fn cidr_mask(prefix: u8) -> u32 { + if prefix == 0 { + 0 + } else { + u32::MAX << (32 - prefix) + } +} diff --git a/src/models.rs b/src/models.rs new file mode 100644 index 0000000..296a8e8 --- /dev/null +++ b/src/models.rs @@ -0,0 +1,28 @@ +#[derive(Debug, Clone, Copy)] +pub struct PendingRange { + pub chunk_start: usize, + pub chunk_end: usize, + pub start: usize, + pub end: usize, +} + +impl PendingRange { + pub fn len(self) -> usize { + self.end.saturating_sub(self.start) + } +} + +#[derive(Debug, Clone)] +pub struct ChunkRecord { + pub chunk_start: usize, + pub chunk_end: usize, + pub tested_count: usize, + pub open_count: usize, + pub is_complete: bool, +} + +#[derive(Debug, Clone, Default)] +pub struct ScanTotals { + pub tested: usize, + pub open: usize, +} diff --git a/src/scan.rs b/src/scan.rs new file mode 100644 index 0000000..71a18a4 --- /dev/null +++ b/src/scan.rs @@ -0,0 +1,305 @@ +use crate::{ + config::Config, + db::Database, + ip::cidr_to_hosts, + models::{ChunkRecord, PendingRange, ScanTotals}, + terminal::{ScreenState, TerminalUi}, +}; +use anyhow::Result; +use std::{ + net::Ipv4Addr, + sync::{ + atomic::{AtomicBool, AtomicUsize, Ordering}, + Arc, + }, + time::Instant, +}; +use tokio::{net::TcpStream, time::timeout}; + +pub async fn run_local_scan( + config: Config, + database: Arc, + ui: Arc, + interrupted: Arc, +) -> Result<()> { + let targets = load_targets(&config.cidrs)?; + let target_count = targets.len(); + let chunk_records = database.load_chunk_records().await?; + let (tested_before_run, _open_before_run) = database.summarize_chunks().await?; + let pending_ranges = build_pending_ranges(target_count, config.chunk_size, &chunk_records); + let pending_count: usize = pending_ranges.iter().map(|range| range.len()).sum(); + + if ui.is_enabled() { + ui.print_banner(); + ui.render(&ScreenState { + current_ip: if pending_count > 0 { + targets[pending_ranges[0].start].to_string() + } else { + "concluído".to_string() + }, + tested: tested_before_run, + total: target_count, + }); + } else { + println!( + "Scanning {} hosts on {} CIDRs, port {}, concurrency {}, chunk size {}. Pending {}, already tested {}.", + target_count, + config.cidrs.len(), + config.port, + config.concurrency, + config.chunk_size, + pending_count, + tested_before_run + ); + } + + if pending_ranges.is_empty() { + if ui.is_enabled() { + ui.render(&ScreenState { + current_ip: "concluído".to_string(), + tested: target_count, + total: target_count, + }); + ui.finish(); + } else { + println!("Scan concluído em 0.00s"); + println!("Total de hosts testados: 0"); + println!("Portas abertas encontradas neste scan: 0"); + println!( + "Total de portas abertas no banco: {}", + database.count_open_endpoints(config.port).await? + ); + } + return Ok(()); + } + + let start = Instant::now(); + let next_range = Arc::new(AtomicUsize::new(0)); + let tested_this_run = Arc::new(AtomicUsize::new(0)); + let open_this_run = Arc::new(AtomicUsize::new(0)); + let mut handles = Vec::new(); + for _ in 0..config.concurrency.max(1) { + let ranges = pending_ranges.clone(); + let next_range = Arc::clone(&next_range); + let tested_this_run = Arc::clone(&tested_this_run); + let open_this_run = Arc::clone(&open_this_run); + let database = Arc::clone(&database); + let ui = Arc::clone(&ui); + let targets = targets.clone(); + 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_range.fetch_add(1, Ordering::SeqCst); + let Some(range) = ranges.get(index).copied() else { + break; + }; + + let chunk_stats = scan_range( + range, + &targets, + &config, + &database, + &ui, + &tested_this_run, + &open_this_run, + &interrupted, + tested_before_run, + ) + .await?; + + persist_range(&database, range, chunk_stats).await?; + } + + Ok::<(), anyhow::Error>(()) + })); + } + + for handle in handles { + handle.await??; + } + + let duration = start.elapsed().as_secs_f64(); + let total_open_count = database.count_open_endpoints(config.port).await?; + + if interrupted.load(Ordering::SeqCst) { + if ui.is_enabled() { + ui.render(&ScreenState { + current_ip: "interrompido".to_string(), + tested: tested_before_run + tested_this_run.load(Ordering::SeqCst), + total: target_count, + }); + ui.finish(); + } else { + println!("Interrompido após {:.2}s. Progresso salvo.", duration); + println!( + "Hosts testados nesta execução: {}", + tested_this_run.load(Ordering::SeqCst) + ); + println!( + "Portas abertas encontradas: {}", + open_this_run.load(Ordering::SeqCst) + ); + } + return Ok(()); + } + + if ui.is_enabled() { + ui.render(&ScreenState { + current_ip: "concluído".to_string(), + tested: target_count, + total: target_count, + }); + ui.finish(); + } else { + println!("Scan concluído em {:.2}s", duration); + println!( + "Total de hosts testados: {}", + tested_this_run.load(Ordering::SeqCst) + ); + println!( + "Portas abertas encontradas neste scan: {}", + open_this_run.load(Ordering::SeqCst) + ); + println!("Total de portas abertas no banco: {}", total_open_count); + } + + Ok(()) +} + +async fn scan_range( + range: PendingRange, + targets: &[Ipv4Addr], + config: &Config, + database: &Database, + ui: &TerminalUi, + tested_this_run: &AtomicUsize, + open_this_run: &AtomicUsize, + interrupted: &AtomicBool, + tested_before_run: usize, +) -> Result { + let mut totals = ScanTotals::default(); + + for index in range.start..range.end { + if interrupted.load(Ordering::SeqCst) { + break; + } + + let ip = targets[index]; + let is_open = check_port_open(ip, config.port, config.timeout_ms).await; + totals.tested += 1; + let tested_now = tested_this_run.fetch_add(1, Ordering::SeqCst) + 1; + + if is_open { + totals.open += 1; + open_this_run.fetch_add(1, Ordering::SeqCst); + database.upsert_open_endpoint(ip, config.port).await?; + } + + if should_render_progress(tested_now, range.end - range.start) || index + 1 == range.end { + ui.render(&ScreenState { + current_ip: ip.to_string(), + tested: tested_before_run + tested_now, + total: targets.len(), + }); + } + } + + Ok(totals) +} + +async fn persist_range(database: &Database, range: PendingRange, totals: ScanTotals) -> Result<()> { + let previous = database.load_chunk_record(range.chunk_start).await?; + let tested_before = previous + .as_ref() + .map(|record| record.tested_count) + .unwrap_or(0); + let open_before = previous + .as_ref() + .map(|record| record.open_count) + .unwrap_or(0); + let tested_count = tested_before + totals.tested; + let open_count = open_before + totals.open; + let record = ChunkRecord { + chunk_start: range.chunk_start, + chunk_end: range.chunk_end, + tested_count, + open_count, + is_complete: tested_count >= range.len() + tested_before, + }; + + database.save_chunk_record(&record).await?; + Ok(()) +} + +pub fn build_pending_ranges( + total_targets: usize, + chunk_size: usize, + records: &[ChunkRecord], +) -> Vec { + let mut ranges = Vec::new(); + let chunk_size = chunk_size.max(1); + + for chunk_start in (0..total_targets).step_by(chunk_size) { + let chunk_end = (chunk_start + chunk_size).min(total_targets); + let chunk_len = chunk_end.saturating_sub(chunk_start); + let record = records + .iter() + .find(|record| record.chunk_start == chunk_start); + let tested_count = record + .map(|record| { + if record.is_complete { + chunk_len + } else { + record.tested_count.min(chunk_len) + } + }) + .unwrap_or(0); + + if tested_count < chunk_len { + ranges.push(PendingRange { + chunk_start, + chunk_end, + start: chunk_start + tested_count, + end: chunk_end, + }); + } + } + + ranges +} + +pub async fn check_port_open(ip: Ipv4Addr, port: u16, timeout_ms: u64) -> bool { + let address = std::net::SocketAddr::from((ip, port)); + matches!( + timeout( + std::time::Duration::from_millis(timeout_ms), + TcpStream::connect(address) + ) + .await, + Ok(Ok(_)) + ) +} + +pub fn load_targets(cidrs: &[String]) -> Result> { + let mut targets = Vec::new(); + + for cidr in cidrs { + targets.extend(cidr_to_hosts(cidr)?); + } + + Ok(targets) +} + +fn should_render_progress(test_index: usize, chunk_len: usize) -> bool { + if chunk_len <= 8 { + return true; + } + + test_index % 4 == 0 +} diff --git a/src/terminal.rs b/src/terminal.rs new file mode 100644 index 0000000..45c4456 --- /dev/null +++ b/src/terminal.rs @@ -0,0 +1,99 @@ +use std::io::{self, IsTerminal, Write}; +use std::sync::Mutex; + +pub const BANNER: [&str; 9] = [ +" ░██████ ░██████ ", +" ░██ ░██ ░██ ░██ ", +"░██ ░██████ ░█████████████░██ ░███████ ░██████ ░████████ ░████████ ░███████ ░██░████ ", +"░██ ░██░██ ░██ ░██░████████░██ ░██ ░██ ░██ ░██░██ ░██░██ ░██░███ ", +"░██ ░███████░██ ░██ ░██ ░██░██ ░███████ ░██ ░██░██ ░██░█████████░██ ", +" ░██ ░██░██ ░██░██ ░██ ░██░██ ░██░██ ░██░██ ░██░██ ░██░██ ░██░██ ░██ ", +" ░██████ ░█████░█░██ ░██ ░██ ░██████ ░███████ ░█████░██░██ ░██░██ ░██ ░███████ ░██ ", +" ", +" ", +]; + +#[derive(Debug, Clone)] +pub struct ScreenState { + pub current_ip: String, + pub tested: usize, + pub total: usize, +} + +pub struct TerminalUi { + enabled: bool, + lock: Mutex<()>, +} + +impl TerminalUi { + pub fn new(enabled: bool) -> Self { + Self { + enabled, + lock: Mutex::new(()), + } + } + + pub fn is_enabled(&self) -> bool { + self.enabled + } + + pub fn print_banner(&self) { + if !self.enabled { + return; + } + + let _guard = self.lock.lock().expect("terminal mutex poisoned"); + let mut stdout = io::stdout(); + let _ = write!(stdout, "\x1b[2J\x1b[H\x1b[?25l"); + for line in BANNER { + let _ = writeln!(stdout, "{line}"); + } + let _ = writeln!(stdout, "IP: aguardando..."); + let _ = writeln!( + stdout, + "Progress: [░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░] | 0.0%" + ); + let _ = stdout.flush(); + } + + pub fn render(&self, state: &ScreenState) { + if !self.enabled { + return; + } + + let _guard = self.lock.lock().expect("terminal mutex poisoned"); + let mut stdout = io::stdout(); + let percent = if state.total == 0 { + 100.0 + } else { + ((state.tested as f64 / state.total as f64) * 100.0).min(100.0) + }; + let bar_width = 36usize; + let filled = ((percent / 100.0) * bar_width as f64).round() as usize; + let empty = bar_width.saturating_sub(filled); + let bar = format!("{}{}", "█".repeat(filled), "░".repeat(empty)); + let ip = format!("{:<15}", state.current_ip); + let percent = format!("{:>5.1}%", percent); + + let _ = write!( + stdout, + "\x1b[2F\x1b[2KIP: {ip}\n\x1b[2KProgress: [{bar}] | {percent}\n" + ); + let _ = stdout.flush(); + } + + pub fn finish(&self) { + if !self.enabled { + return; + } + + let _guard = self.lock.lock().expect("terminal mutex poisoned"); + let mut stdout = io::stdout(); + let _ = write!(stdout, "\x1b[?25h"); + let _ = stdout.flush(); + } +} + +pub fn stdout_is_terminal() -> bool { + io::stdout().is_terminal() +} diff --git a/src/worker.rs b/src/worker.rs new file mode 100644 index 0000000..8a4d968 --- /dev/null +++ b/src/worker.rs @@ -0,0 +1,191 @@ +use crate::{ + config::Config, + scan::check_port_open, + terminal::{ScreenState, TerminalUi}, +}; +use anyhow::{anyhow, Context, Result}; +use std::{net::Ipv4Addr, sync::Arc}; +use tokio::{ + io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, + net::TcpStream, + task::JoinSet, + time::sleep, +}; + +pub async fn run_worker(config: Config, ui: Arc) -> Result<()> { + if ui.is_enabled() { + 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 { + let Some(line) = lines.next_line().await? else { + break; + }; + + if line == "DONE" { + break; + } + + 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() { + ui.render(&ScreenState { + current_ip: "concluído".to_string(), + tested: 0, + total: 0, + }); + ui.finish(); + } + + Ok(()) +} + +async fn connect_with_retry(addr: std::net::SocketAddr) -> Result { + let mut attempts = 0usize; + + loop { + match TcpStream::connect(addr).await { + Ok(stream) => return Ok(stream), + Err(err) => { + attempts += 1; + if attempts >= 30 { + return Err(err).with_context(|| format!("failed to connect to {addr}")); + } + + sleep(std::time::Duration::from_secs(1)).await; + } + } + } +} + +struct Job { + chunk_start: usize, + chunk_end: usize, + ips: Vec, +} + +struct JobOutcome { + tested: usize, + open_ips: Vec, +} + +fn parse_job_line(line: &str) -> Result { + let mut parts = line.splitn(6, ' '); + let kind = parts.next().unwrap_or(""); + if kind != "JOB" { + return Err(anyhow!("unexpected coordinator message: {line}")); + } + + let chunk_start = parse_usize_part(parts.next(), "chunk_start", line)?; + let chunk_end = parse_usize_part(parts.next(), "chunk_end", line)?; + let _port = parse_usize_part(parts.next(), "port", line)?; + let _timeout_ms = parse_usize_part(parts.next(), "timeout_ms", line)?; + let ips_part = parts.next().unwrap_or(""); + let ips = if ips_part.trim().is_empty() { + Vec::new() + } else { + ips_part + .split(',') + .map(str::trim) + .filter(|entry| !entry.is_empty()) + .map(|entry| { + entry + .parse::() + .with_context(|| format!("invalid IP in job: {entry}")) + }) + .collect::>>()? + }; + + Ok(Job { + chunk_start, + chunk_end, + ips, + }) +} + +async fn execute_job( + ips: &[Ipv4Addr], + port: u16, + timeout_ms: u64, + concurrency: usize, + ui: &TerminalUi, +) -> Result { + let mut tested = 0usize; + let mut open_ips = Vec::new(); + let concurrency = concurrency.max(1); + let mut index = 0usize; + + while index < ips.len() { + let end = (index + concurrency).min(ips.len()); + let mut join_set = JoinSet::new(); + + for ip in ips[index..end].iter().copied() { + join_set.spawn(async move { (ip, check_port_open(ip, port, timeout_ms).await) }); + } + + while let Some(join_result) = join_set.join_next().await { + let (ip, is_open) = join_result?; + tested += 1; + if is_open { + open_ips.push(ip); + } + + if ui.is_enabled() { + ui.render(&ScreenState { + current_ip: ip.to_string(), + tested, + total: ips.len(), + }); + } + } + + index = end; + } + + Ok(JobOutcome { tested, open_ips }) +} + +fn parse_usize_part(part: Option<&str>, name: &str, line: &str) -> Result { + let value = part.ok_or_else(|| anyhow!("missing {name} in message: {line}"))?; + value + .trim() + .parse() + .map_err(|_| anyhow!("invalid {name} in message: {line}")) +} + +fn format_ips(ips: &[Ipv4Addr]) -> String { + ips.iter() + .map(ToString::to_string) + .collect::>() + .join(",") +}