commit dbf128730e97db0558b577a6a77eba24a380b87c Author: mark <-> Date: Sun Mar 2 21:45:16 2025 +0100 init diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ea8c4bf --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/target diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 0000000..79b4381 --- /dev/null +++ b/Cargo.lock @@ -0,0 +1,687 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "async-channel" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81953c529336010edd6d8e358f886d9581267795c61b19475b71314bffa46d35" +dependencies = [ + "concurrent-queue", + "event-listener 2.5.3", + "futures-core", +] + +[[package]] +name = "async-channel" +version = "2.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89b47800b0be77592da0afd425cc03468052844aff33b84e33cc696f64e77b6a" +dependencies = [ + "concurrent-queue", + "event-listener-strategy", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-executor" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30ca9a001c1e8ba5149f91a74362376cc6bc5b919d92d988668657bd570bdcec" +dependencies = [ + "async-task", + "concurrent-queue", + "fastrand", + "futures-lite", + "slab", +] + +[[package]] +name = "async-fs" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebcd09b382f40fcd159c2d695175b2ae620ffa5f3bd6f664131efff4e8b9e04a" +dependencies = [ + "async-lock", + "blocking", + "futures-lite", +] + +[[package]] +name = "async-global-executor" +version = "2.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05b1b633a2115cd122d73b955eadd9916c18c8f510ec9cd1686404c60ad1c29c" +dependencies = [ + "async-channel 2.3.1", + "async-executor", + "async-io", + "async-lock", + "blocking", + "futures-lite", + "once_cell", +] + +[[package]] +name = "async-io" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "43a2b323ccce0a1d90b449fd71f2a06ca7faa7c54c2751f06c9bd851fc061059" +dependencies = [ + "async-lock", + "cfg-if", + "concurrent-queue", + "futures-io", + "futures-lite", + "parking", + "polling", + "rustix", + "slab", + "tracing", + "windows-sys", +] + +[[package]] +name = "async-lock" +version = "3.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff6e472cdea888a4bd64f342f09b3f50e1886d32afe8df3d663c01140b811b18" +dependencies = [ + "event-listener 5.3.1", + "event-listener-strategy", + "pin-project-lite", +] + +[[package]] +name = "async-net" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b948000fad4873c1c9339d60f2623323a0cfd3816e5181033c6a5cb68b2accf7" +dependencies = [ + "async-io", + "blocking", + "futures-lite", +] + +[[package]] +name = "async-process" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63255f1dc2381611000436537bbedfe83183faa303a5a0edaf191edef06526bb" +dependencies = [ + "async-channel 2.3.1", + "async-io", + "async-lock", + "async-signal", + "async-task", + "blocking", + "cfg-if", + "event-listener 5.3.1", + "futures-lite", + "rustix", + "tracing", +] + +[[package]] +name = "async-signal" +version = "0.2.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "637e00349800c0bdf8bfc21ebbc0b6524abea702b0da4168ac00d070d0c0b9f3" +dependencies = [ + "async-io", + "async-lock", + "atomic-waker", + "cfg-if", + "futures-core", + "futures-io", + "rustix", + "signal-hook-registry", + "slab", + "windows-sys", +] + +[[package]] +name = "async-std" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c634475f29802fde2b8f0b505b1bd00dfe4df7d4a000f0b36f7671197d5c3615" +dependencies = [ + "async-channel 1.9.0", + "async-global-executor", + "async-io", + "async-lock", + "crossbeam-utils", + "futures-channel", + "futures-core", + "futures-io", + "futures-lite", + "gloo-timers", + "kv-log-macro", + "log", + "memchr", + "once_cell", + "pin-project-lite", + "pin-utils", + "slab", + "wasm-bindgen-futures", +] + +[[package]] +name = "async-task" +version = "4.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" + +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + +[[package]] +name = "autocfg" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26" + +[[package]] +name = "bitflags" +version = "2.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b048fb63fd8b5923fc5aa7b340d8e156aec7ec02f0c78fa8a6ddc2613f6f71de" + +[[package]] +name = "blocking" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "703f41c54fc768e63e091340b424302bb1c29ef4aa0c7f10fe849dfb114d29ea" +dependencies = [ + "async-channel 2.3.1", + "async-task", + "futures-io", + "futures-lite", + "piper", +] + +[[package]] +name = "bumpalo" +version = "3.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "79296716171880943b8470b5f8d03aa55eb2e645a4874bdbb28adb49162e012c" + +[[package]] +name = "cfg-if" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" + +[[package]] +name = "concurrent-queue" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" + +[[package]] +name = "errno" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33d852cb9b869c2a9b3df2f71a3074817f01e1844f839a144f5fcef059a4eb5d" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "event-listener" +version = "2.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0" + +[[package]] +name = "event-listener" +version = "5.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6032be9bd27023a771701cc49f9f053c751055f71efb2e0ae5c15809093675ba" +dependencies = [ + "concurrent-queue", + "parking", + "pin-project-lite", +] + +[[package]] +name = "event-listener-strategy" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c3e4e0dd3673c1139bf041f3008816d9cf2946bbfac2945c09e523b8d7b05b2" +dependencies = [ + "event-listener 5.3.1", + "pin-project-lite", +] + +[[package]] +name = "fastrand" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" + +[[package]] +name = "futures-channel" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10" +dependencies = [ + "futures-core", +] + +[[package]] +name = "futures-core" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" + +[[package]] +name = "futures-io" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" + +[[package]] +name = "futures-lite" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cef40d21ae2c515b51041df9ed313ed21e572df340ea58a922a0aefe7e8891a1" +dependencies = [ + "fastrand", + "futures-core", + "futures-io", + "parking", + "pin-project-lite", +] + +[[package]] +name = "gloo-timers" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbb143cf96099802033e0d4f4963b19fd2e0b728bcf076cd9cf7f6634f092994" +dependencies = [ + "futures-channel", + "futures-core", + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "hermit-abi" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fbf6a919d6cf397374f7dfeeea91d974c7c0a7221d0d0f4f20d859d329e53fcc" + +[[package]] +name = "js-sys" +version = "0.3.76" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6717b6b5b077764fb5966237269cb3c64edddde4b14ce42647430a78ced9e7b7" +dependencies = [ + "once_cell", + "wasm-bindgen", +] + +[[package]] +name = "kv-log-macro" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0de8b303297635ad57c9f5059fd9cee7a47f8e8daa09df0fcd07dd39fb22977f" +dependencies = [ + "log", +] + +[[package]] +name = "libc" +version = "0.2.169" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5aba8db14291edd000dfcc4d620c7ebfb122c613afb886ca8803fa4e128a20a" + +[[package]] +name = "linux-raw-sys" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78b3ae25bc7c8c38cec158d1f2757ee79e9b3740fbc7ccf0e59e4b08d793fa89" + +[[package]] +name = "log" +version = "0.4.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24" +dependencies = [ + "value-bag", +] + +[[package]] +name = "memchr" +version = "2.7.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78ca9ab1a0babb1e7d5695e3530886289c18cf2f87ec19a575a0abdce112e3a3" + +[[package]] +name = "once_cell" +version = "1.20.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1261fe7e33c73b354eab43b1273a57c8f967d0391e80353e51f764ac02cf6775" + +[[package]] +name = "parking" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" + +[[package]] +name = "pin-project-lite" +version = "0.2.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "915a1e146535de9163f3987b8944ed8cf49a18bb0056bcebcdcece385cece4ff" + +[[package]] +name = "pin-utils" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" + +[[package]] +name = "piper" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96c8c490f422ef9a4efd2cb5b42b76c8613d7e7dfc1caf667b8a3350a5acc066" +dependencies = [ + "atomic-waker", + "fastrand", + "futures-io", +] + +[[package]] +name = "polling" +version = "3.7.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a604568c3202727d1507653cb121dbd627a58684eb09a820fd746bee38b4442f" +dependencies = [ + "cfg-if", + "concurrent-queue", + "hermit-abi", + "pin-project-lite", + "rustix", + "tracing", + "windows-sys", +] + +[[package]] +name = "proc-macro2" +version = "1.0.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37d3544b3f2748c54e147655edb5025752e2303145b5aefb3c3ea2c78b973bb0" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5b9d34b8991d19d98081b46eacdd8eb58c6f2b201139f7c5f643cc155a633af" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "rustix" +version = "0.38.42" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f93dc38ecbab2eb790ff964bb77fa94faf256fd3e73285fd7ba0903b76bedb85" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys", + "windows-sys", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9e9e0b4211b72e7b8b6e85c807d36c212bdb33ea8587f7569562a84df5465b1" +dependencies = [ + "libc", +] + +[[package]] +name = "slab" +version = "0.4.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f92a496fb766b417c996b9c5e57daf2f7ad3b0bebe1ccfca4856390e3d3bb67" +dependencies = [ + "autocfg", +] + +[[package]] +name = "smol" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a33bd3e260892199c3ccfc487c88b2da2265080acb316cd920da72fdfd7c599f" +dependencies = [ + "async-channel 2.3.1", + "async-executor", + "async-fs", + "async-io", + "async-lock", + "async-net", + "async-process", + "blocking", + "futures-lite", +] + +[[package]] +name = "syn" +version = "2.0.90" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "919d3b74a5dd0ccd15aeb8f93e7006bd9e14c295087c9896a110f490752bcf31" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "tmoth" +version = "0.1.0" +dependencies = [ + "async-std", + "smol", +] + +[[package]] +name = "tracing" +version = "0.1.41" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0" +dependencies = [ + "pin-project-lite", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e672c95779cf947c5311f83787af4fa8fffd12fb27e4993211a84bdfd9610f9c" + +[[package]] +name = "unicode-ident" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "adb9e6ca4f869e1180728b7950e35922a7fc6397f7b641499e8f3ef06e50dc83" + +[[package]] +name = "value-bag" +version = "1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ef4c4aa54d5d05a279399bfa921ec387b7aba77caf7a682ae8d86785b8fdad2" + +[[package]] +name = "wasm-bindgen" +version = "0.2.99" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a474f6281d1d70c17ae7aa6a613c87fce69a127e2624002df63dcb39d6cf6396" +dependencies = [ + "cfg-if", + "once_cell", + "wasm-bindgen-macro", +] + +[[package]] +name = "wasm-bindgen-backend" +version = "0.2.99" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5f89bb38646b4f81674e8f5c3fb81b562be1fd936d84320f3264486418519c79" +dependencies = [ + "bumpalo", + "log", + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.49" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38176d9b44ea84e9184eff0bc34cc167ed044f816accfe5922e54d84cf48eca2" +dependencies = [ + "cfg-if", + "js-sys", + "once_cell", + "wasm-bindgen", + "web-sys", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.99" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2cc6181fd9a7492eef6fef1f33961e3695e4579b9872a6f7c83aee556666d4fe" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.99" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30d7a95b763d3c45903ed6c81f156801839e5ee968bb07e534c44df0fcd330c2" +dependencies = [ + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-backend", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.99" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "943aab3fdaaa029a6e0271b35ea10b72b943135afe9bffca82384098ad0e06a6" + +[[package]] +name = "web-sys" +version = "0.3.76" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "04dd7223427d52553d3702c004d3b2fe07c148165faa56313cb00211e31c12bc" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets", +] + +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..4d63013 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,8 @@ +[package] +name = "tmoth" +version = "0.1.0" +edition = "2021" + +[dependencies] +async-std = "1.13.0" +smol = "2.0.2" diff --git a/src/client.rs b/src/client.rs new file mode 100644 index 0000000..ad464d1 --- /dev/null +++ b/src/client.rs @@ -0,0 +1,15 @@ +use std::{net::SocketAddr, sync::{atomic::AtomicBool, Arc}}; +use smol::{spawn, net::TcpListener}; + +pub async fn run(local: SocketAddr, remote: SocketAddr, name: String, kill: Option>) { + match TcpListener::bind(local).await { + Ok(listener) => { + eprintln!("[OK] forwarding from {local} to {remote} {name}"); + loop { + let (connection, _) = listener.accept().await.unwrap(); + spawn(crate::protocol::client(connection, remote, name.clone(), kill.as_ref().map(Arc::clone))).detach(); + } + }, + Err(e) => eprintln!("[ERR] Couldn't listen on {local}: {e}"), + } +} diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..88538f7 --- /dev/null +++ b/src/config.rs @@ -0,0 +1,37 @@ +use std::net::{SocketAddr, ToSocketAddrs}; + +#[derive(Debug)] +pub struct Config { + pub mappings: Vec<(String, Vec, Vec<(String, String)>)>, +} + +impl Config { + pub fn parse(str: &str) -> Result { + let mut mappings = vec![]; + for (i, line) in str.lines().enumerate() { + let i = i + 1; + if let Some(ln) = line.strip_prefix("from ") { + if let Some(addrs) = ln.trim().to_socket_addrs().ok() { + mappings.push((ln.trim().to_owned(), addrs.collect(), vec![])); + } else { + return Err(format!("Line {i}: {line}: expected `from `")); + } + } else if let Some(ln) = line.strip_prefix("to ") { + if let Some((name, addr)) = ln.split_once(char::is_whitespace) { + if let Some((_, _, mapping_to)) = mappings.last_mut() { + mapping_to.push((name.to_owned(), addr.to_owned())); + } else { + return Err(format!("Line {i}: {line}: `to ...` lines are only valid after the first `from ...` line")); + } + } else { + return Err(format!("Line {i}: {line}: expected `to `")); + } + } else if !(line.trim().is_empty() || line.trim().starts_with('#') || line.trim().starts_with("//")) { + return Err(format!("Line {i}: expected `from` or `to`")); + } + } + Ok(Self { + mappings, + }) + } +} diff --git a/src/main.rs b/src/main.rs new file mode 100644 index 0000000..9569f78 --- /dev/null +++ b/src/main.rs @@ -0,0 +1,94 @@ +pub mod config; +pub mod server; +pub mod client; +pub mod protocol; + +use std::{net::ToSocketAddrs, sync::{atomic::AtomicBool, Arc}}; +use smol::spawn; +use crate::config::Config; + +pub const BUF: usize = 128; + +fn main() { + let args = std::env::args().skip(1).collect::>(); + if args.len() != 1 && args.len() != 3 { + eprintln!("---- 🦋 tmoth 🦋 ----"); + eprintln!("tmoth keeps tcp connections alive,"); + eprintln!("even when they are interrupted."); + eprintln!("it even supports reconnecting."); + eprintln!(" -- Usage --"); + eprintln!("tmoth # server mode"); + eprintln!("tmoth "); + return; + } + if args.len() == 1 { + let config = std::fs::read_to_string(&args[0]).expect("failed to read config"); + let config = Config::parse(&config).expect("failed to parse config"); + smol::block_on(server::run(config)); + } else { + smol::block_on(async { + let kill = Some(Arc::new(AtomicBool::new(false))); + let mut tasks = vec![]; + for addr_from in args[0].to_socket_addrs().unwrap() { + eprintln!("{}: {}", args[0], addr_from); + let addrs = args[1].to_socket_addrs().unwrap().collect::>(); + let addr = if addrs.len() == 1 { + addrs[0] + } else if addrs.is_empty() { + panic!("{}: no addresses", args[1]); + } else { + for (i, addr) in addrs.iter().enumerate() { + eprintln!("> {}: {addr}", i+1); + } + loop { + eprintln!("Enter a number (1-{})", addrs.len()); + if let Some(line) = std::io::stdin().lines().next().and_then(|v| v.ok()) { + let line = line.trim(); + let index = if line.is_empty() { + 1 + } else { + line.parse::().expect("Expected a number") + }; + if index != 0 && index <= addrs.len() { + break addrs[index-1]; + } + } else { + break addrs[0]; + } + } + }; + tasks.push(spawn(client::run(addr_from, addr, args[2].clone(), kill.as_ref().map(Arc::clone)))); + } + let mut line = String::new(); + while async_std::io::stdin().read_line(&mut line).await.is_ok() { + let line = std::mem::replace(&mut line, String::new()); + let line = line.trim(); + match line { + "kill" => { + if let Some(kill) = &kill { + eprintln!("[`kill`] killing connections on next forwarding action. `unkill` will revert this."); + kill.store(true, std::sync::atomic::Ordering::Relaxed); + } else { + eprintln!("[`kill`] cannot kill"); + } + } + "unkill" => { + if let Some(kill) = &kill { + eprintln!("[`unkill`] connections will no longer be killed."); + kill.store(false, std::sync::atomic::Ordering::Relaxed); + } else { + eprintln!("[`unkill`] cannot kill/unkill"); + } + } + _ => { + eprintln!("[`{line}`] Unknown command, try `kill` / `unkill`"); + } + } + } + eprintln!("stdin closed, running noninteractively"); + for task in tasks { + task.await; + } + }); + } +} diff --git a/src/protocol.rs b/src/protocol.rs new file mode 100644 index 0000000..642894f --- /dev/null +++ b/src/protocol.rs @@ -0,0 +1,276 @@ +use std::collections::{HashMap, VecDeque}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::{net::SocketAddr, time::Duration}; +use std::sync::Arc; +use smol::channel::{unbounded, Receiver, Sender}; +use smol::lock::Mutex; +use smol::{spawn, Task}; +use smol::{io::{self, AsyncReadExt, AsyncWriteExt}, net::TcpStream, Timer}; + +use crate::BUF; + +pub const MAJOR: u8 = 0; +pub const MINOR: u8 = 0; +pub const PATCH: u8 = 0; + +pub async fn server(mut connection: TcpStream, mut connection_id: u64, mapping: Arc>, connections: Arc, Arc, Sender>>, Receiver>>, TcpStream)>>)>>>) -> io::Result<()> { + connection.write_all(&[b't', b'm', b'o', b't', b'h', MAJOR, MINOR, PATCH]).await?; + connection.write_all(&connection_id.to_be_bytes()).await?; + let mut b = [0u8]; + connection.read_exact(&mut b).await?; + let mutex = if b[0] < 255 { + // new connection + let mut name_buf = vec![0u8; b[0] as usize]; + connection.read_exact(&mut name_buf).await?; + let name = String::from_utf8(name_buf).map_err(|_| io::ErrorKind::InvalidData)?; + let mut addr = None; + for (mname, maddr) in mapping.iter() { + if mname == name.as_str() { + addr = Some(maddr.to_owned()); + break; + } + } + let o = if let Some(addr) = addr { + match TcpStream::connect(&addr).await { + Ok(mut target) => { + // forward between connection and target, with buffers in case of disconnect + let (send, recv) = unbounded(); + let mutex = Arc::new(Mutex::new((VecDeque::new(), send.clone(), recv, target.clone()))); + let task = spawn(async move { + let mut buf = [0u8; BUF]; + loop { + match target.read(&mut buf).await { + Ok(0) | Err(_) => { + let _ = send.send(Ok(true)).await; + return; + } + Ok(n) => if send.send(Err(buf[0..n].iter().copied().collect())).await.is_err() { return; }, + } + } + }); + connections.lock().await.insert(connection_id, (task, Arc::clone(&mutex))); + mutex + } + Err(e) => { + eprintln!("[{connection_id}] Couldn't connect to {name} ({addr}): {e}"); + return Ok(()); + } + } + } else { + eprintln!("[{connection_id}] No such name: {name}"); + return Ok(()); + }; + eprintln!("[{connection_id}] connected"); + o + } else { + let mut b = [0u8; 8]; + connection.read_exact(&mut b).await?; + connection_id = u64::from_be_bytes(b); + // reconnection + let o = if let Some((_task, mutex)) = connections.lock().await.get(&connection_id) { + Arc::clone(mutex) + } else { + eprintln!("[{connection_id}] no connection with the requested id exists"); + return Ok(()); + }; + eprintln!("[{connection_id}] reconnected"); + o + }; + let mut lock = if let Some(lock) = mutex.try_lock() { + lock + } else { + eprintln!("[{connection_id}] waiting for lock..."); + let lock = mutex.lock().await; + lock + }; + let _task_fwd = { + let mut remote = connection.clone(); + let mut local = lock.3.clone(); + let send = lock.1.clone(); + spawn(async move { + let mut buf = [0u8; BUF]; + loop { + match remote.read(&mut buf).await { + Ok(0) => { + // EOF -> Disconnect + let _ = send.send(Ok(false)).await; + return; + } + Ok(n) => if local.write_all(&buf[0..n]).await.is_err() { + // local disconnect + let _ = send.send(Ok(true)).await; + return; + }, + Err(_) => { + // reconnect + let _ = send.send(Ok(false)).await; + return; + } + }; + } + }) + }; + 'fwd: while let Ok(data) = if lock.0.is_empty() { lock.2.recv().await } else { Ok(Err(std::mem::replace(&mut lock.0, VecDeque::new()))) } { + if let Err(mut data) = data { + while !data.is_empty() { + match connection.write(data.make_contiguous()).await { + Ok(0) => break 'fwd, + Ok(n) => for _ in 0..n { + data.pop_front(); + } + Err(_) => { + lock.0 = data; + eprintln!("[{connection_id}] transport connection closed temporarily"); + return Ok(()); + } + } + } + } else if let Ok(true) = data { + break; + } else { + eprintln!("[{connection_id}] transport connection closed temporarily"); + return Ok(()); + } + } + eprintln!("[{connection_id}] connection closed and removed"); + connections.lock().await.remove(&connection_id); + Ok(()) +} + +pub async fn client(local: TcpStream, remote: SocketAddr, name: String, kill: Option>) -> io::Result<()> { + if name.len() > 254 { + eprintln!("[ERR] name too long"); + Err(io::ErrorKind::InvalidInput)?; + } + let mut connection_id: Option = None; + let mut send_buf = VecDeque::new(); + 'reconnect: loop { + let mut remote = loop { + if !kill.as_ref().is_some_and(|kill| kill.load(Ordering::Relaxed)) { + match TcpStream::connect(remote).await { + Ok(stream) => break stream, + Err(e) => eprintln!("[{}] Could not connect to {remote}: {e}", connection_id.map(|v| v.to_string()).as_deref().unwrap_or("-")), + } + } + Timer::after(Duration::from_secs(3)).await; + }; + let mut init_msg = [0u8; 16]; + remote.read_exact(&mut init_msg).await?; + if init_msg[0..5] != [b't', b'm', b'o', b't', b'h'] { + eprintln!("This does not appear to be a tmoth socket (got {})...", String::from_utf8_lossy(&init_msg[0..5])); + Err(io::ErrorKind::InvalidData)?; + } + let major = init_msg[5]; + let minor = init_msg[6]; + let patch = init_msg[7]; + if MAJOR > major { + eprintln!("[VERSION] Server has older MAJOR version: {major} vs. {MAJOR}"); + Err(io::ErrorKind::Unsupported)?; + } else if MAJOR < major { + eprintln!("[VERSION] Server has newer MAJOR version: {major} vs. {MAJOR}"); + Err(io::ErrorKind::Unsupported)?; + } + if MINOR > minor { + eprintln!("[version] Server has older minor version: {minor} vs. {MINOR}"); + } else if MINOR < minor { + eprintln!("[version] Server has newer minor version: {minor} vs. {MINOR}"); + } + if PATCH > patch { + eprintln!("[version] Server has older patch version: {patch} vs. {PATCH}"); + } else if PATCH < patch { + eprintln!("[version] Server has newer patch version: {patch} vs. {PATCH}"); + } + let connection_id = if let Some(connection_id) = connection_id { + // reconnect + let bytes: [u8; 8] = connection_id.to_be_bytes(); + if remote.write_all(&[0xFF, bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7]]).await.is_err() { + Timer::after(Duration::from_secs(3)).await; + continue 'reconnect; + } + eprintln!("[{connection_id}] reconnected"); + connection_id + } else { + // initial connect + let cid = u64::from_be_bytes([init_msg[8], init_msg[9], init_msg[10], init_msg[11], init_msg[12], init_msg[13], init_msg[14], init_msg[15]]); + remote.write_all(&[name.len() as u8]).await?; + remote.write_all(name.as_bytes()).await?; + eprintln!("[{cid}] connected"); + connection_id = Some(cid); + cid + }; + let (send, recv) = unbounded(); + let _task = { + let mut local = local.clone(); + let send = send.clone(); + spawn(async move { + let mut buf = [0u8; BUF]; + loop { + match local.read(&mut buf).await { + Ok(0) | Err(_) => { + let _ = send.send(Ok(true)).await; + return; + } + Ok(n) => if send.send(Err(buf[0..n].iter().copied().collect())).await.is_err() { return; }, + } + } + }) + }; + let _task_fwd = { + let mut remote = remote.clone(); + let mut local = local.clone(); + let kill = kill.as_ref().map(Arc::clone); + spawn(async move { + let mut buf = [0u8; BUF]; + loop { + if kill.as_ref().is_some_and(|kill| kill.load(Ordering::Relaxed)) { + let _ = send.send(Ok(false)).await; + return; + } else { + match remote.read(&mut buf).await { + Ok(0) => { + // EOF -> Disconnect + let _ = send.send(Ok(false)).await; + return; + } + Ok(n) => if local.write_all(&buf[0..n]).await.is_err() { + // local disconnect + let _ = send.send(Ok(true)).await; + return; + }, + Err(_) => { + // reconnect + let _ = send.send(Ok(false)).await; + return; + } + } + } + } + }) + }; + 'fwd: while let Ok(data) = if send_buf.is_empty() { recv.recv().await } else { Ok(Err(std::mem::replace(&mut send_buf, VecDeque::new()))) } { + if let Err(mut data) = data { + let kill = kill.as_ref().is_some_and(|kill| kill.load(Ordering::Relaxed)); + while !data.is_empty() { + match if kill { None } else { remote.write(data.make_contiguous()).await.ok() } { + Some(0) => break 'fwd, + Some(n) => for _ in 0..n { + data.pop_front(); + } + None => { + send_buf = data; + eprintln!("[{connection_id}] disconnected temporarily"); + Timer::after(Duration::from_secs(3)).await; + continue 'reconnect; + } + } + } + } else if let Ok(true) = data { + return Ok(()); + } else { + eprintln!("[{connection_id}] temporary disconnect"); + Timer::after(Duration::from_secs(3)).await; + continue 'reconnect; + } + } + } +} diff --git a/src/server.rs b/src/server.rs new file mode 100644 index 0000000..139cc3f --- /dev/null +++ b/src/server.rs @@ -0,0 +1,55 @@ +use crate::config::Config; + +use std::collections::VecDeque; +use std::{collections::HashMap, net::SocketAddr}; +use std::sync::Arc; +use smol::channel::{Receiver, Sender}; +use smol::lock::Mutex; +use smol::Task; +use smol::{net::{TcpListener, TcpStream}, spawn}; + +pub async fn run(config: Config) { + let connections = Arc::new(Mutex::new(HashMap::new())); + let mut tasks = vec![]; + for (mapping_from_name, mapping_from, mapping_to) in config.mappings.into_iter() { + eprintln!( + "[OK] listening on {mapping_from_name}:\n from | {}\n to{}", + mapping_from.iter().enumerate().map(|(i, addr)| format!("{}{addr}", if i == 0 { "" } else { " / "})).collect::(), + mapping_to.iter().map(|(name, addr)| format!(" | {name} {addr}")).collect::(), + ); + let mapping_to = Arc::new(mapping_to); + for mapping_from in mapping_from { + match TcpListener::bind(mapping_from).await { + Ok(listener) => { + tasks.push(spawn(listen(mapping_from, listener, Arc::clone(&mapping_to), Arc::clone(&connections)))); + }, + Err(e) => eprintln!("[SKIP] Couldn't listen on {mapping_from}: {e}"), + } + } + } + for task in tasks { + task.await; + } +} + +async fn listen(addr: SocketAddr, listener: TcpListener, mapping: Arc>, connections: Arc, Arc, Sender>>, Receiver>>, TcpStream)>>)>>>) { + let mut counter = u64::MAX; + loop { + counter = counter.wrapping_add(1); + match listener.accept().await { + Ok((con, _)) => { + let mapping = mapping.clone(); + let connections = Arc::clone(&connections); + spawn(async move { + if let Err(e) = crate::protocol::server(con, counter, mapping, connections).await { + eprintln!("[{counter}] Connection setup failed: {e}"); + } + }).detach(); + } + Err(e) => { + eprintln!("[ERR] Can't accept connections on {addr}: {e}"); + return; + } + } + } +} diff --git a/tmoth.conf b/tmoth.conf new file mode 100644 index 0000000..0fa32be --- /dev/null +++ b/tmoth.conf @@ -0,0 +1,3 @@ +from 0.0.0.0:26055 +to A 127.0.0.1:2221 +to B 127.0.0.1:2222