fujin/tora
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: fujin/tora:6309b87323290d3d9638edbfc2e754293989997f
Branches Tags
fujin/tora:master
...
compare: fujin/tora:master
Branches Tags
fujin/tora:master

10 Commits
6309b87323 ... master

Author SHA1 Message Date
Alexander b489ff7135 feat(torad): route source fetches through the tunnel and package VPN mode 
Three things, all tail end of VPN mode.

Source fetches are HTTP, not BitTorrent, so librqbit's SO_BINDTODEVICE
never covered them. SourceResolver now holds two clients and picks one
per URL: remote indexer and .torrent fetches go through a client bound
to wg0, so a future route change cannot quietly send them around the
tunnel; loopback URLs keep the unbound client, because pasta splices the
namespace's loopback to the host's and that is how a self-hosted Jackett
stays reachable. That traffic never leaves the machine, so keeping it off
the tunnel is deliberate.

This replaces the planned request-time URL rewriting, which turned out to
be unnecessary: measured, pasta reaches host services on 127.0.0.1 from
inside the namespace even when they bind after the namespace starts, so
neither Jackett URLs nor a loopback DATABASE_URL need touching.

Second, a defect the packaging work surfaced: killing the pid in the pid
file killed pasta but left torad running, reparented to init, with a dead
tap interface -- the daemon outliving the only documented way to stop it.
The re-executed process now sets PR_SET_PDEATHSIG so it dies with pasta.

Third, packaging: passt, wireguard-go and iproute2 in both devenv files,
and the module documentation states the Linux-only, leech-only and
pinned-endpoint limitations along with what happens when the tunnel drops.
wireguard-tools is deliberately absent -- the device is configured over
UAPI.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-17 23:09:03 +02:00
Alexander 62db1e2a9e feat(torad): bring up a WireGuard tunnel in user space from a config file 
`--wireguard-config` is now the only thing an operator supplies. torad
re-executes itself inside a private user+network namespace, brings up a
userspace WireGuard datapath, and binds every torrent socket to it. No
root, no wg-quick, no pre-created interface, no kernel module, no one-time
host setup.

The mechanism differs from what was planned, because the plan's
assumptions did not survive being checked against the actual binaries:

- pasta already creates an unprivileged user+net namespace with full
  capabilities and a configured tap interface, so the hand-rolled
  unshare(CLONE_NEWUSER|CLONE_NEWNET) with uid_map, gid_map and
  setgroups=deny is gone. torad re-execs itself under pasta instead.

- WG_SOCKET_DIRECTORY does not exist; wireguard-go's socket directory is
  a build-time linker variable. WG_UAPI_FD does not avoid it either,
  because UAPIListen inotify-watches that path even when handed a
  pre-bound socket. Since /var/run is a symlink to /run, a tmpfs over
  /var inside a private mount namespace makes /var/run/wireguard
  writable while leaving the real /run visible -- which matters, because
  torad's gRPC socket lives under /run/user and the aggregator connects
  to it from the host.

- The peer endpoint needs a host route via pasta's gateway before the
  default route moves to wg0, or WireGuard's own handshake is routed
  into the tunnel it is trying to establish.

The device is configured by speaking WireGuard's UAPI protocol over its
unix socket, so wireguard-tools is not a dependency either.

If the datapath exits, torad shuts down instead of routing around it: a
live torad with a dead tunnel is the failure mode that leaks.

pasta also creates a PID namespace, so the pid file is written on the
host before the re-exec and not again inside -- otherwise it would
record PID 1, which names init when read from the host.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-17 23:01:33 +02:00
Alexander 6e5d27497e feat(torad): validate WireGuard config at parse time 
The parser stored every value as a raw String and only ever rejected a
missing key. Those values are about to be handed to `ip addr add`, route
setup, and the wireguard-go UAPI socket in Wave 3, none of which give a
diagnostic worth reading: a mistyped address comes back as a bare EINVAL
or a UAPI errno with no hint of which config line caused it.

Parse into typed values instead — Key, IpCidr, Endpoint — and report
failures with a line number and the offending value.

Key validation was not in the original scope, but the fixture in this
file's own tests was a 41-character "private key" that parsed clean:
long enough to look right, short enough to fail at the UAPI socket three
waves later. Decoding here also yields the hex encoding that UAPI
expects, so it replaces work rather than adding it. Errors on the
private key deliberately describe only the shape of the problem, never
the input, and there is a test that fails if the value leaks.

Address and DNS become lists because wg-quick allows comma-separated
values and dual-stack providers emit them; parsing only the first would
have silently dropped the IPv6 address.

Also reject what used to be accepted in silence: a second [Peer] or
[Interface] section, a repeated key within a section, an unknown
section, and a key appearing before any section header.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-17 22:33:41 +02:00
Alexander e321300f67 feat(torad): add WireGuard config parser and user-namespace preflight 
Groundwork for VPN mode; nothing calls it yet, so the module is dead code
until the namespace bootstrap lands.

vpn::config parses the ProtonVPN-style WireGuard INI into typed sections.
`InterfaceSection` gets a hand-written Debug that redacts the private
key — the derived one would print it verbatim, and this struct is about
to start flowing through error contexts during interface setup.

vpn::namespace::preflight checks user.max_user_namespaces and fails with
the sysctl name and its remediation when unprivileged user namespaces are
disabled. The decision logic is split into `evaluate` so the tests
exercise the real code path rather than a copy of it — the /proc file
itself cannot be mocked.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-15 00:17:33 +02:00
Alexander 7103184c8e fix(torad): bypass system proxy env vars when fetching .torrent sources 
reqwest honours HTTP_PROXY/ALL_PROXY from the environment by default. A
self-hosted Jackett is normally reached over loopback, and routing
loopback through an inherited proxy breaks it — which matters once torad
runs in an environment where those vars are set for VPN reasons.

Extracts the builder into `build_http_client` so the behaviour is
directly testable, and adds a test that serves one response from a
loopback listener while ALL_PROXY points at a dead port. The test is
falsifiable: removing .no_proxy() makes it fail with ConnectionRefused
against the proxy address, which was verified before committing.

The proxy URL is deliberately http:// rather than socks5:// — reqwest is
built here without its `socks` feature, so a SOCKS proxy would be ignored
and the test would pass either way.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-15 00:17:19 +02:00
Alexander dd41f1961f feat(torad): upgrade librqbit to 9.0.0-rc.0 and add --bind-device 
Pinned exactly (=9.0.0-rc.0) — no v9 stable exists yet, so a caret range
would silently drift onto a future prerelease.

Two source changes were needed for the upgrade:

- `torrent_from_bytes_ext` moved to `librqbit_core::torrent_metainfo::
  torrent_from_bytes` and no longer wraps the result in a `meta` field.
- `peer_stats.live` / `.not_needed` are now `u32`, so the casts are
  redundant.

v9 also adds `SessionOptions::bind_device_name`, which is the reason for
the upgrade: it applies SO_BINDTODEVICE to every librqbit socket — peer
connections, trackers, DHT and LSD. Binding those to a VPN interface
means that when the interface goes away the sockets error out instead of
falling back to the host route, giving a kernel-enforced kill switch.
Exposed as --bind-device / TORAD_BIND_DEVICE; unset reproduces today's
behaviour exactly.

Note that `listen` stays at its default of None, so there is no listener
and no uTP socket in either direction — torad is TCP-only and leech-only.
That is unchanged from v8 but now written down, since incoming
connections need NAT-PMP port forwarding that we have not built.

Also drops torad's `nix` pin from 0.29 to 0.31.3 to match the sibling
tora crate; the workspace was carrying three copies.

Verified end-to-end against a real swarm rather than by compiling alone:
a 755 MiB torrent added via HTTP .torrent URL, driven through
pending -> downloading -> finished, with pause/resume on an active
torrent, remove, and the notification stream all exercised. Evidence in
.omo/evidence/task-5-torad-vpn-namespace.txt.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-15 00:17:09 +02:00
Alexander 7842b5aeba Add readiness probe to devenv 2026-07-25 12:44:20 +02:00
Alexander 3f3a6d354a Move torrents to completed after finish 2026-07-24 16:47:42 +02:00
Alexander 9c2499a4a3 Implement notification stream rpc 2026-07-22 20:32:44 +02:00
Alexander e1d35f81bc Fix torad hanging on add of torrent 2026-07-21 17:47:54 +02:00
26 changed files with 6061 additions and 498 deletions
Expand all files Collapse all files
Cargo.lock
Generated
+344 -87
View File
@@ -97,6 +97,12 @@ dependencies = [
"rustversion",
]
[[package]]
name = "arrayvec"
version = "0.7.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56"
[[package]]
name = "assert_cfg"
version = "0.1.0"
@@ -208,6 +214,8 @@ dependencies = [
"http",
"http-body",
"http-body-util",
"hyper",
"hyper-util",
"itoa",
"matchit",
"memchr",
@@ -216,9 +224,11 @@ dependencies = [
"pin-project-lite",
"serde_core",
"sync_wrapper",
"tokio",
"tower",
"tower-layer",
"tower-service",
"tracing",
]
[[package]]
@@ -237,17 +247,43 @@ dependencies = [
"sync_wrapper",
"tower-layer",
"tower-service",
"tracing",
]
[[package]]
name = "backoff"
version = "0.4.0"
name = "axum-extra"
version = "0.10.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b62ddb9cb1ec0a098ad4bbf9344d0713fa193ae1a80af55febcff2627b6a00c1"
checksum = "9963ff19f40c6102c76756ef0a46004c0d58957d87259fc9208ff8441c12ab96"
dependencies = [
"getrandom 0.2.17",
"instant",
"rand 0.8.6",
"axum",
"axum-core",
"bytes",
"form_urlencoded",
"futures-util",
"http",
"http-body",
"http-body-util",
"mime",
"pin-project-lite",
"rustversion",
"serde_core",
"serde_html_form",
"serde_path_to_error",
"tower-layer",
"tower-service",
"tracing",
]
[[package]]
name = "backon"
version = "1.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cffb0e931875b666fc4fcb20fee52e9bbd1ef836fd9e9e04ec21555f9f85f7ef"
dependencies = [
"fastrand",
"gloo-timers",
"tokio",
]
[[package]]
@@ -256,15 +292,6 @@ version = "0.22.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
[[package]]
name = "bincode"
version = "1.3.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b1f45e9417d87227c7a56d22e471c6206462cba514c7590c09aff4cf6d1ddcad"
dependencies = [
"serde",
]
[[package]]
name = "bincode"
version = "2.0.1"
@@ -397,6 +424,17 @@ dependencies = [
"rand_core 0.10.1",
]
[[package]]
name = "chardetng"
version = "0.1.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "14b8f0b65b7b08ae3c8187e8d77174de20cb6777864c6b832d8ad365999cf1ea"
dependencies = [
"cfg-if",
"encoding_rs",
"memchr",
]
[[package]]
name = "chrono"
version = "0.4.45"
@@ -724,6 +762,18 @@ dependencies = [
"syn",
]
[[package]]
name = "dontfrag"
version = "1.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c117949f5a8b25ba471b4dfea927a07a2f8730c585532a2eb4294e18f5155397"
dependencies = [
"cfg-if",
"libc",
"tokio",
"windows-sys 0.52.0",
]
[[package]]
name = "dotenvy"
version = "0.15.7"
@@ -751,6 +801,15 @@ dependencies = [
"serde",
]
[[package]]
name = "encoding_rs"
version = "0.8.35"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3"
dependencies = [
"cfg-if",
]
[[package]]
name = "equivalent"
version = "1.0.2"
@@ -788,6 +847,12 @@ dependencies = [
"pin-project-lite",
]
[[package]]
name = "fastrand"
version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223"
[[package]]
name = "find-msvc-tools"
version = "0.1.9"
@@ -1011,6 +1076,18 @@ dependencies = [
"rand_core 0.10.1",
]
[[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 = "governor"
version = "0.10.4"
@@ -1405,15 +1482,6 @@ dependencies = [
"serde_core",
]
[[package]]
name = "instant"
version = "0.1.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e0242819d153cba4b4b05a5a8f2a7e9bbf97b6055b2a002b395c96b5ff3c0222"
dependencies = [
"cfg-if",
]
[[package]]
name = "intervaltree"
version = "0.2.7"
@@ -1505,18 +1573,19 @@ dependencies = [
[[package]]
name = "librqbit"
version = "8.1.1"
version = "9.0.0-rc.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dadca8f521242010a4c846ef5f224c217009c92e272709cdc08ba9cdabe62983"
checksum = "b9451def41c18d1b8e7b68c097bcb0c9a0579cdc9b779803368a65b86cae03d2"
dependencies = [
"anyhow",
"arc-swap",
"async-compression",
"async-stream",
"async-trait",
"backoff",
"axum-extra",
"backon",
"base64",
"bincode 2.0.1",
"bincode",
"bitvec",
"byteorder",
"bytes",
@@ -1532,22 +1601,29 @@ dependencies = [
"librqbit-clone-to-owned",
"librqbit-core",
"librqbit-dht",
"librqbit-dualstack-sockets",
"librqbit-lsd",
"librqbit-peer-protocol",
"librqbit-sha1-wrapper",
"librqbit-tracker-comms",
"librqbit-upnp",
"librqbit-utp",
"memmap2",
"mime_guess",
"nix 0.30.1",
"parking_lot",
"rand 0.9.4",
"regex",
"reqwest",
"rlimit",
"serde",
"serde_derive",
"serde_json",
"serde_urlencoded",
"serde_with",
"size_format",
"socket2",
"thiserror 2.0.18",
"tokio",
"tokio-socks",
"tokio-stream",
@@ -1557,61 +1633,71 @@ dependencies = [
"urlencoding",
"uuid",
"walkdir",
"windows",
]
[[package]]
name = "librqbit-bencode"
version = "3.1.0"
version = "9.0.0-rc.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "606dff526ba81e3eca33e2bb28b53afa2bc0b2c41d252333fa44e6c11abb37da"
checksum = "97490fb9b35fb575f498cc1e804b23be6bbf3603d5a2358733e5b867d1ee52e8"
dependencies = [
"anyhow",
"arrayvec",
"atoi",
"bytes",
"librqbit-buffers",
"librqbit-clone-to-owned",
"librqbit-sha1-wrapper",
"serde",
"serde_derive",
"thiserror 2.0.18",
]
[[package]]
name = "librqbit-buffers"
version = "4.2.0"
version = "9.0.0-rc.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d78c78b907d6171a7191c162b2b60db46d254ebde6a95282b77372af556c1463"
checksum = "e845dbd14174712471c3430c958d5c25e391561c2dd31f39f895bb60f63c46f0"
dependencies = [
"bytes",
"librqbit-clone-to-owned",
"serde",
"serde_derive",
]
[[package]]
name = "librqbit-clone-to-owned"
version = "3.0.1"
version = "9.0.0-rc.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cbd1e66d773ba9c475ff89286dc1d6f9d167cbb898603797467dd0ea6844c445"
checksum = "76ef698eb8d7745f425f2b9a93e6a9d946bcb8bf84a559303e0fc3f01e8ad07c"
dependencies = [
"bytes",
]
[[package]]
name = "librqbit-core"
version = "5.0.0"
version = "9.0.0-rc.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "55a02cc6fce6743ad38661ccd6fafc6cf1ae5e0106a9922836b0524dbe752378"
checksum = "a38e805aad201d96024849649174015d48a575348b35bbd927d99b47464e665e"
dependencies = [
"anyhow",
"assert_cfg",
"bytes",
"chardetng",
"data-encoding",
"directories",
"encoding_rs",
"hex",
"itertools",
"librqbit-bencode",
"librqbit-buffers",
"librqbit-clone-to-owned",
"librqbit-sha1-wrapper",
"memchr",
"parking_lot",
"rand 0.9.4",
"serde",
"serde_derive",
"thiserror 2.0.18",
"tokio",
"tokio-util",
"tracing",
@@ -1620,27 +1706,29 @@ dependencies = [
[[package]]
name = "librqbit-dht"
version = "5.3.1"
version = "9.0.0-rc.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c7cc129194337771a86b0399956c4d9bf1cd97c5f24d14a50be38e170f76a54b"
checksum = "d59a9374b33a52a065171641ae518db97969d3db849a57824c72c53d2109918b"
dependencies = [
"anyhow",
"backoff",
"byteorder",
"backon",
"bytes",
"chrono",
"dashmap",
"futures",
"hex",
"indexmap 2.14.0",
"leaky-bucket",
"librqbit-bencode",
"librqbit-buffers",
"librqbit-clone-to-owned",
"librqbit-core",
"librqbit-dualstack-sockets",
"parking_lot",
"rand 0.9.4",
"serde",
"serde_derive",
"serde_json",
"thiserror 2.0.18",
"tokio",
"tokio-stream",
"tokio-util",
@@ -1648,13 +1736,49 @@ dependencies = [
]
[[package]]
name = "librqbit-peer-protocol"
version = "4.3.0"
name = "librqbit-dualstack-sockets"
version = "0.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a73129497b500505f33d1dc0426319b6a6a208f13fdfaae56224ab8c2346a773"
checksum = "f7189f1ff66ca75055137a38b062ba7df9691d071b7172e5090997c62eb4b4de"
dependencies = [
"axum",
"backon",
"futures",
"libc",
"network-interface",
"socket2",
"thiserror 2.0.18",
"tokio",
"tracing",
]
[[package]]
name = "librqbit-lsd"
version = "9.0.0-rc.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4e65ddc3b64392a71378944712c9830aac8ef98f860b180419dd42b4596e7faa"
dependencies = [
"anyhow",
"atoi",
"bstr",
"futures",
"httparse",
"librqbit-core",
"librqbit-dualstack-sockets",
"parking_lot",
"rand 0.9.4",
"tokio",
"tokio-util",
"tracing",
]
[[package]]
name = "librqbit-peer-protocol"
version = "9.0.0-rc.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "76ade1bbea8cf3e7eb51978b32a549787d17bbf872f3f0ed47d7c85e7377705a"
dependencies = [
"anyhow",
"bincode 1.3.3",
"bitvec",
"byteorder",
"bytes",
@@ -1664,13 +1788,16 @@ dependencies = [
"librqbit-clone-to-owned",
"librqbit-core",
"serde",
"serde_derive",
"thiserror 2.0.18",
"tracing",
]
[[package]]
name = "librqbit-sha1-wrapper"
version = "4.1.0"
version = "9.0.0-rc.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "79373a02db73159e4de7ca5d27b6eeae2d540df66c6801db2b01c5513d087524"
checksum = "eeb119e5c05f1aa4011cf2e4dc19f83403064d2210804e2f35f06abac4e5b710"
dependencies = [
"assert_cfg",
"aws-lc-rs",
@@ -1678,21 +1805,26 @@ dependencies = [
[[package]]
name = "librqbit-tracker-comms"
version = "3.0.0"
version = "9.0.0-rc.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "08204944c5be677a5de8e1230e0249fce5c14abef23048e26452c6fb03f1b260"
checksum = "0ae0c520cc5f028fb50d746eaa6ba2aa082228db96c2baf1a312f9cacfceeb58"
dependencies = [
"anyhow",
"async-stream",
"backon",
"byteorder",
"futures",
"itertools",
"librqbit-bencode",
"librqbit-buffers",
"librqbit-core",
"librqbit-dualstack-sockets",
"parking_lot",
"rand 0.9.4",
"reqwest",
"serde",
"serde_derive",
"serde_with",
"tokio",
"tokio-util",
"tracing",
@@ -1702,23 +1834,49 @@ dependencies = [
[[package]]
name = "librqbit-upnp"
version = "1.0.0"
version = "9.0.0-rc.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "545aad6124c97201055983137e12a19f34acad565120c3cd30596cbd72e8fa86"
checksum = "058af0d5e2a99c567b73369455e3852c48addcf751836f08cfc25ca89dd16cc3"
dependencies = [
"anyhow",
"bstr",
"futures",
"httparse",
"librqbit-dualstack-sockets",
"network-interface",
"quick-xml",
"reqwest",
"serde",
"serde_derive",
"socket2",
"tokio",
"tracing",
"url",
]
[[package]]
name = "librqbit-utp"
version = "0.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4f3bfdc73944bc76cab24d5690a98816770040a654c449edf5ff2b9ba22626aa"
dependencies = [
"bitvec",
"dontfrag",
"lazy_static",
"libc",
"librqbit-dualstack-sockets",
"metrics",
"parking_lot",
"rand 0.9.4",
"ringbuf",
"rustc-hash",
"socket2",
"thiserror 2.0.18",
"tokio",
"tokio-util",
"tracing",
]
[[package]]
name = "libsqlite3-sys"
version = "0.37.0"
@@ -1787,6 +1945,16 @@ dependencies = [
"libc",
]
[[package]]
name = "metrics"
version = "0.24.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "89550ee9f79e88fef3119de263694973a8adb26c21d75322164fb8c493039fe2"
dependencies = [
"portable-atomic",
"rapidhash",
]
[[package]]
name = "mime"
version = "0.3.17"
@@ -1836,6 +2004,18 @@ dependencies = [
"winapi",
]
[[package]]
name = "nix"
version = "0.30.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "74523f3a35e05aba87a1d978330aef40f67b0304ac79c1c00b294c9830543db6"
dependencies = [
"bitflags",
"cfg-if",
"cfg_aliases",
"libc",
]
[[package]]
name = "nix"
version = "0.31.3"
@@ -2033,6 +2213,15 @@ version = "1.13.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49"
[[package]]
name = "portable-atomic-util"
version = "0.2.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618"
dependencies = [
"portable-atomic",
]
[[package]]
name = "potential_utf"
version = "0.1.5"
@@ -2106,9 +2295,9 @@ dependencies = [
[[package]]
name = "quick-xml"
version = "0.37.5"
version = "0.38.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "331e97a1af0bf59823e6eadffe373d7b27f485be8748f71471c662c1f269b7fb"
checksum = "b66c2058c55a409d601666cffe35f04333cf1013010882cec174a7467cd4e21c"
dependencies = [
"memchr",
"serde",
@@ -2196,24 +2385,13 @@ version = "0.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc33ff2d4973d518d823d61aa239014831e521c75da58e3df4840d3f47749d09"
[[package]]
name = "rand"
version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a"
dependencies = [
"libc",
"rand_chacha 0.3.1",
"rand_core 0.6.4",
]
[[package]]
name = "rand"
version = "0.9.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea"
dependencies = [
"rand_chacha 0.9.0",
"rand_chacha",
"rand_core 0.9.5",
]
@@ -2228,16 +2406,6 @@ dependencies = [
"rand_core 0.10.1",
]
[[package]]
name = "rand_chacha"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88"
dependencies = [
"ppv-lite86",
"rand_core 0.6.4",
]
[[package]]
name = "rand_chacha"
version = "0.9.0"
@@ -2248,15 +2416,6 @@ dependencies = [
"rand_core 0.9.5",
]
[[package]]
name = "rand_core"
version = "0.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c"
dependencies = [
"getrandom 0.2.17",
]
[[package]]
name = "rand_core"
version = "0.9.5"
@@ -2272,6 +2431,15 @@ version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"
[[package]]
name = "rapidhash"
version = "4.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5da7e78a036ce858e8d55b7e7dc8ba3a88b78350fd2155d3591bbd966b58589e"
dependencies = [
"rustversion",
]
[[package]]
name = "raw-cpuid"
version = "11.6.0"
@@ -2405,6 +2573,17 @@ dependencies = [
"windows-sys 0.52.0",
]
[[package]]
name = "ringbuf"
version = "0.4.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fe47b720588c8702e34b5979cb3271a8b1842c7cb6f57408efa70c779363488c"
dependencies = [
"crossbeam-utils",
"portable-atomic",
"portable-atomic-util",
]
[[package]]
name = "rlimit"
version = "0.10.2"
@@ -2536,6 +2715,19 @@ dependencies = [
"syn",
]
[[package]]
name = "serde_html_form"
version = "0.2.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b2f2d7ff8a2140333718bb329f5c40fc5f0865b84c426183ce14c97d2ab8154f"
dependencies = [
"form_urlencoded",
"indexmap 2.14.0",
"itoa",
"ryu",
"serde_core",
]
[[package]]
name = "serde_json"
version = "1.0.150"
@@ -2549,6 +2741,17 @@ dependencies = [
"zmij",
]
[[package]]
name = "serde_path_to_error"
version = "0.1.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457"
dependencies = [
"itoa",
"serde",
"serde_core",
]
[[package]]
name = "serde_urlencoded"
version = "0.7.1"
@@ -3200,12 +3403,14 @@ name = "tora"
version = "0.1.0"
dependencies = [
"anyhow",
"chrono",
"clap",
"colored",
"hyper-util",
"nix",
"nix 0.31.3",
"serde_json",
"tokio",
"tokio-stream",
"tonic",
"tonic-health",
"tora-proto",
@@ -3232,6 +3437,7 @@ dependencies = [
"clap",
"librqbit",
"librqbit-core",
"nix 0.31.3",
"reqwest",
"sha1 0.10.7",
"sqlx",
@@ -3655,6 +3861,27 @@ version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
[[package]]
name = "windows"
version = "0.62.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "527fadee13e0c05939a6a05d5bd6eec6cd2e3dbd648b9f8e447c6518133d8580"
dependencies = [
"windows-collections",
"windows-core",
"windows-future",
"windows-numerics",
]
[[package]]
name = "windows-collections"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "23b2d95af1a8a14a3c7367e1ed4fc9c20e0a26e79551b1454d72583c97cc6610"
dependencies = [
"windows-core",
]
[[package]]
name = "windows-core"
version = "0.62.2"
@@ -3668,6 +3895,17 @@ dependencies = [
"windows-strings",
]
[[package]]
name = "windows-future"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e1d6f90251fe18a279739e78025bd6ddc52a7e22f921070ccdc67dde84c605cb"
dependencies = [
"windows-core",
"windows-link",
"windows-threading",
]
[[package]]
name = "windows-implement"
version = "0.60.2"
@@ -3696,6 +3934,16 @@ version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
[[package]]
name = "windows-numerics"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6e2e40844ac143cdb44aead537bbf727de9b044e107a0f1220392177d15b0f26"
dependencies = [
"windows-core",
"windows-link",
]
[[package]]
name = "windows-result"
version = "0.4.1"
@@ -3748,6 +3996,15 @@ dependencies = [
"windows_x86_64_msvc",
]
[[package]]
name = "windows-threading"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3949bd5b99cafdf1c7ca86b43ca564028dfe27d66958f2470940f73d86d75b37"
dependencies = [
"windows-link",
]
[[package]]
name = "windows_aarch64_gnullvm"
version = "0.52.6"
crates/proto/src/generated/torrent/torrent.rs
+878 -249
View File
File diff suppressed because it is too large Load Diff
crates/proto/src/generated/torrent/torrent.serde.rs
+1296 -12
View File
File diff suppressed because it is too large Load Diff
crates/proto/src/generated/torrent/torrent.tonic.rs
+125
View File
@@ -173,6 +173,37 @@ pub mod torrents_client {
.insert(GrpcMethod::new("torrent.Torrents", "Resume"));
self.inner.unary(req, path, codec).await
}
pub async fn notifications(
&mut self,
request: impl tonic::IntoRequest<super::NotificationsRequest>,
) -> std::result::Result<
tonic::Response<tonic::codec::Streaming<super::Notification>>,
tonic::Status,
> {
self.inner.ready().await.map_err(|e| {
tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
})?;
let codec = tonic_prost::ProstCodec::default();
let path = http::uri::PathAndQuery::from_static("/torrent.Torrents/Notifications");
let mut req = request.into_request();
req.extensions_mut()
.insert(GrpcMethod::new("torrent.Torrents", "Notifications"));
self.inner.server_streaming(req, path, codec).await
}
pub async fn files(
&mut self,
request: impl tonic::IntoRequest<super::FilesRequest>,
) -> std::result::Result<tonic::Response<super::FilesResponse>, tonic::Status> {
self.inner.ready().await.map_err(|e| {
tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
})?;
let codec = tonic_prost::ProstCodec::default();
let path = http::uri::PathAndQuery::from_static("/torrent.Torrents/Files");
let mut req = request.into_request();
req.extensions_mut()
.insert(GrpcMethod::new("torrent.Torrents", "Files"));
self.inner.unary(req, path, codec).await
}
}
}
/// Generated server implementations.
@@ -212,6 +243,19 @@ pub mod torrents_server {
&self,
request: tonic::Request<super::ResumeRequest>,
) -> std::result::Result<tonic::Response<super::ResumeResponse>, tonic::Status>;
/// Server streaming response type for the Notifications method.
type NotificationsStream: tonic::codegen::tokio_stream::Stream<
Item = std::result::Result<super::Notification, tonic::Status>,
> + std::marker::Send
+ 'static;
async fn notifications(
&self,
request: tonic::Request<super::NotificationsRequest>,
) -> std::result::Result<tonic::Response<Self::NotificationsStream>, tonic::Status>;
async fn files(
&self,
request: tonic::Request<super::FilesRequest>,
) -> std::result::Result<tonic::Response<super::FilesResponse>, tonic::Status>;
}
#[derive(Debug)]
pub struct TorrentsServer<T> {
@@ -508,6 +552,87 @@ pub mod torrents_server {
};
Box::pin(fut)
}
"/torrent.Torrents/Notifications" => {
#[allow(non_camel_case_types)]
struct NotificationsSvc<T: Torrents>(pub Arc<T>);
impl<T: Torrents>
tonic::server::ServerStreamingService<super::NotificationsRequest>
for NotificationsSvc<T>
{
type Response = super::Notification;
type ResponseStream = T::NotificationsStream;
type Future =
BoxFuture<tonic::Response<Self::ResponseStream>, tonic::Status>;
fn call(
&mut self,
request: tonic::Request<super::NotificationsRequest>,
) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move {
<T as Torrents>::notifications(&inner, request).await
};
Box::pin(fut)
}
}
let accept_compression_encodings = self.accept_compression_encodings;
let send_compression_encodings = self.send_compression_encodings;
let max_decoding_message_size = self.max_decoding_message_size;
let max_encoding_message_size = self.max_encoding_message_size;
let inner = self.inner.clone();
let fut = async move {
let method = NotificationsSvc(inner);
let codec = tonic_prost::ProstCodec::default();
let mut grpc = tonic::server::Grpc::new(codec)
.apply_compression_config(
accept_compression_encodings,
send_compression_encodings,
)
.apply_max_message_size_config(
max_decoding_message_size,
max_encoding_message_size,
);
let res = grpc.server_streaming(method, req).await;
Ok(res)
};
Box::pin(fut)
}
"/torrent.Torrents/Files" => {
#[allow(non_camel_case_types)]
struct FilesSvc<T: Torrents>(pub Arc<T>);
impl<T: Torrents> tonic::server::UnaryService<super::FilesRequest> for FilesSvc<T> {
type Response = super::FilesResponse;
type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
fn call(
&mut self,
request: tonic::Request<super::FilesRequest>,
) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move { <T as Torrents>::files(&inner, request).await };
Box::pin(fut)
}
}
let accept_compression_encodings = self.accept_compression_encodings;
let send_compression_encodings = self.send_compression_encodings;
let max_decoding_message_size = self.max_decoding_message_size;
let max_encoding_message_size = self.max_encoding_message_size;
let inner = self.inner.clone();
let fut = async move {
let method = FilesSvc(inner);
let codec = tonic_prost::ProstCodec::default();
let mut grpc = tonic::server::Grpc::new(codec)
.apply_compression_config(
accept_compression_encodings,
send_compression_encodings,
)
.apply_max_message_size_config(
max_decoding_message_size,
max_encoding_message_size,
);
let res = grpc.unary(method, req).await;
Ok(res)
};
Box::pin(fut)
}
_ => Box::pin(async move {
let mut response = http::Response::new(tonic::body::Body::default());
let headers = response.headers_mut();
crates/proto/src/lib.rs
+5 -2
View File
@@ -2,9 +2,12 @@ pub mod torrent {
include!("generated/torrent/torrent.rs");
}
pub use torrent::notification as notification_detail;
pub use torrent::{
AddRequest, AddResponse, ListRequest, ListResponse, PauseRequest, PauseResponse, RemoveRequest,
RemoveResponse, ResumeRequest, ResumeResponse, State, StatusRequest, TorrentStatus,
AddRequest, AddResponse, AddedInfo, FileState, FilesRequest, FilesResponse, ListRequest,
ListResponse, Notification, NotificationKind, NotificationsRequest, PauseRequest,
PauseResponse, ProgressMark, RemoveRequest, RemoveResponse, RemovedInfo, ResumeRequest,
ResumeResponse, State, StateChanged, StatusRequest, TorrentFile, TorrentStatus,
torrents_client::TorrentsClient,
torrents_server::{Torrents, TorrentsServer},
};
crates/tora/Cargo.toml
+3 -1
View File
@@ -5,12 +5,14 @@ edition = "2024"
[dependencies]
anyhow = "1.0.103"
chrono = { version = "0.4", default-features = false, features = ["clock"] }
clap = { version = "4.6.1", features = ["derive", "env"] }
colored = "3"
hyper-util = { version = "0.1.20", features = ["tokio"] }
nix = { version = "0.31", features = ["signal"] }
serde_json = "1.0.140"
tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "net"] }
tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "net", "signal"] }
tokio-stream = "0.1.18"
tonic = "0.14.6"
tonic-health = "0.14.6"
tora-proto = { version = "0.1.0", path = "../proto" }
crates/tora/src/cli.rs
+43 -2
View File
@@ -1,6 +1,7 @@
use std::path::PathBuf;
use clap::{Parser, Subcommand, ValueEnum};
use tora_proto::NotificationKind;
#[derive(Parser)]
#[command(
@@ -43,6 +44,31 @@ pub enum SortKey {
Added,
}
/// CLI-facing mirror of `NotificationKind` so `clap`'s `ValueEnum` derive can
/// enumerate the values. Mapped back to the proto enum via `to_proto`.
#[derive(Copy, Clone, Debug, ValueEnum)]
pub enum KindFilter {
/// `TORRENT_ADDED`
Added,
/// `TORRENT_REMOVED`
Removed,
/// `STATE_CHANGED` (includes DOWNLOADING <-> STALE, * -> FINISHED, etc.)
StateChanged,
/// `PROGRESS` (one event per integer percent advance)
Progress,
}
impl KindFilter {
pub fn to_proto(self) -> i32 {
match self {
KindFilter::Added => NotificationKind::TorrentAdded as i32,
KindFilter::Removed => NotificationKind::TorrentRemoved as i32,
KindFilter::StateChanged => NotificationKind::StateChanged as i32,
KindFilter::Progress => NotificationKind::Progress as i32,
}
}
}
#[derive(Subcommand)]
pub enum Command {
/// Check whether torad is reachable and healthy.
@@ -51,9 +77,13 @@ pub enum Command {
Download {
/// Magnet link, HTTP(S) URL, or path to a .torrent file.
source: String,
/// Directory to save the torrent into. Defaults to torad's configured download directory.
/// Folder to download into. Defaults to torad's `<download_dir>/<torrent_name>/`.
#[arg(long)]
output_dir: Option<String>,
download_folder: Option<String>,
/// Folder to move the torrent into once it finishes downloading.
/// Defaults to torad's `<download_dir>/completed/<torrent_name>/`.
#[arg(long)]
move_after_completion_folder: Option<String>,
},
/// Show the status of a single torrent.
Status {
@@ -95,4 +125,15 @@ pub enum Command {
},
/// Stop the torad daemon.
Stop,
/// Stream torrent notifications live from torad. Runs until interrupted
/// (Ctrl-C) or the daemon disconnects.
Watch {
/// Filter to these torrent ids (UUIDs or unique prefixes). Repeatable.
/// Omit to observe all torrents.
#[arg(long = "id", value_name = "ID")]
ids: Vec<String>,
/// Only emit these event kinds. Repeatable. Omit for all kinds.
#[arg(long = "kind", value_enum)]
kinds: Vec<KindFilter>,
},
}
crates/tora/src/main.rs
+64 -6
View File
@@ -6,17 +6,18 @@ use std::time::Duration;
use anyhow::{Context, Result};
use clap::Parser;
use cli::{Cli, Command, SortKey};
use cli::{Cli, Command, KindFilter, SortKey};
use hyper_util::rt::TokioIo;
use output::Formatter;
use tokio::net::UnixStream;
use tokio_stream::StreamExt;
use tonic::transport::{Channel, Endpoint, Uri};
use tonic_health::pb::HealthCheckRequest;
use tonic_health::pb::health_check_response::ServingStatus;
use tonic_health::pb::health_client::HealthClient;
use tora_proto::{
AddRequest, ListRequest, PauseRequest, RemoveRequest, ResumeRequest, StatusRequest,
TorrentsClient,
AddRequest, ListRequest, NotificationsRequest, PauseRequest, RemoveRequest, ResumeRequest,
StatusRequest, TorrentsClient,
};
use tower::service_fn;
@@ -40,7 +41,20 @@ async fn main() -> Result<()> {
match cli.command {
Command::Health => health(&socket, f).await,
Command::Download { source, output_dir } => download(&socket, source, output_dir, f).await,
Command::Download {
source,
download_folder,
move_after_completion_folder,
} => {
download(
&socket,
source,
download_folder,
move_after_completion_folder,
f,
)
.await
}
Command::Status { id } => status(&socket, id, f).await,
Command::List { sort } => list(&socket, sort, f).await,
Command::Remove { id, delete_files } => remove(&socket, id, delete_files, f).await,
@@ -51,6 +65,7 @@ async fn main() -> Result<()> {
download_dir,
} => start_daemon(&socket, &pid_file, &database_url, download_dir, f).await,
Command::Stop => stop_daemon(&socket, &pid_file, f).await,
Command::Watch { ids, kinds } => watch(&socket, ids, kinds, f).await,
}
}
@@ -99,7 +114,8 @@ async fn health(socket: &Path, f: &dyn Formatter) -> Result<()> {
async fn download(
socket: &Path,
source: String,
output_dir: Option<String>,
download_folder: Option<String>,
move_after_completion_folder: Option<String>,
f: &dyn Formatter,
) -> Result<()> {
let channel = connect(socket).await?;
@@ -109,7 +125,8 @@ async fn download(
RPC_TIMEOUT,
client.add(AddRequest {
magnet: source,
output_dir: output_dir.unwrap_or_default(),
download_folder: download_folder.unwrap_or_default(),
move_after_completion_folder: move_after_completion_folder.unwrap_or_default(),
}),
)
.await
@@ -269,3 +286,44 @@ async fn stop_daemon(socket: &Path, pid_file: &Path, f: &dyn Formatter) -> Resul
Err(err) => Err(err).context("failed to send SIGTERM to torad"),
}
}
async fn watch(
socket: &Path,
ids: Vec<String>,
kinds: Vec<KindFilter>,
f: &dyn Formatter,
) -> Result<()> {
let channel = connect(socket).await?;
let mut client = TorrentsClient::new(channel);
let request = NotificationsRequest {
ids,
kinds: kinds.iter().map(|k| k.to_proto()).collect(),
};
let stream = client
.notifications(request)
.await
.context("Notifications RPC failed")?
.into_inner();
tokio::pin!(stream);
let stdout = std::io::stdout();
while let Some(message) = stream.next().await {
match message {
Ok(notification) => {
let line = f.watch(&notification)?;
let mut guard = stdout.lock();
use std::io::Write;
writeln!(guard, "{line}")?;
guard.flush()?;
}
Err(status) => {
anyhow::bail!("stream error: {status}");
}
}
}
Ok(())
}
crates/tora/src/output.rs
+109 -1
View File
@@ -1,8 +1,12 @@
use std::fmt::Write;
use anyhow::Result;
use chrono::{Local, TimeZone};
use colored::Colorize;
use tora_proto::{State, TorrentStatus};
use tora_proto::{
Notification, NotificationKind, ProgressMark, State, StateChanged, TorrentStatus,
notification_detail::Detail,
};
use crate::cli::OutputFormat;
@@ -18,6 +22,7 @@ pub trait Formatter {
fn resume(&self) -> Result<String>;
fn start(&self) -> Result<String>;
fn stop(&self) -> Result<String>;
fn watch(&self, event: &Notification) -> Result<String>;
}
pub fn from_format(format: OutputFormat) -> Box<dyn Formatter> {
@@ -77,6 +82,10 @@ impl Formatter for LongFormatter {
fn stop(&self) -> Result<String> {
Ok(format!(" {} torad stopped", "●".yellow()))
}
fn watch(&self, event: &Notification) -> Result<String> {
Ok(format_event_long(event))
}
}
struct JsonFormatter;
@@ -130,6 +139,10 @@ impl Formatter for JsonFormatter {
&serde_json::json!({ "stopped": true }),
)?)
}
fn watch(&self, event: &Notification) -> Result<String> {
Ok(serde_json::to_string(event)?)
}
}
fn format_status_long(status: &TorrentStatus) -> String {
@@ -216,6 +229,99 @@ fn format_status_long(status: &TorrentStatus) -> String {
out
}
fn format_event_long(event: &Notification) -> String {
let time = Local
.timestamp_millis_opt(event.observed_at_unix_millis as i64)
.single()
.map(|t| t.format("%H:%M:%S").to_string())
.unwrap_or_else(|| "??".to_string());
let kind = NotificationKind::try_from(event.kind).unwrap_or(NotificationKind::Unspecified);
let torrent = event.torrent.as_ref();
let id = torrent.map(|t| short_id(&t.id)).unwrap_or_default();
let name = torrent
.map(|t| display_name(&t.name))
.unwrap_or_else(|| "(unknown)".to_string());
let state = torrent
.and_then(|t| State::try_from(t.state).ok())
.unwrap_or(State::Unspecified);
let id_w = 8usize;
let name_w = 30usize;
let (icon, suffix) = match (&kind, event.detail.as_ref()) {
(NotificationKind::TorrentAdded, _) => ("+".green(), String::new()),
(NotificationKind::TorrentRemoved, Some(Detail::Removed(r))) => {
let note = if r.files_deleted {
" (files deleted)"
} else {
""
};
("−".red(), format!("{}{}", colored_state(state), note.red()))
}
(NotificationKind::TorrentRemoved, _) => ("−".red(), colored_state(state).to_string()),
(
NotificationKind::StateChanged,
Some(Detail::StateChanged(StateChanged { current, .. })),
) => {
let cur = State::try_from(*current).unwrap_or(State::Unspecified);
let icon = state_icon(cur);
(icon, colored_state(cur).to_string())
}
(
NotificationKind::Progress,
Some(Detail::Progress(ProgressMark {
percent,
bytes_delta,
})),
) => {
let delta = humanize_bytes(*bytes_delta);
let pct_bar = format_pct_inline(*percent);
(
"↑".yellow(),
format!(
"{} {} {}",
pct_bar.yellow(),
format!("{}%", percent).bold(),
format!("(Δ {delta})").dimmed()
),
)
}
(_, _) => ("·".dimmed(), colored_state(state).to_string()),
};
format!(
" {} {} {} {} {}",
time.dimmed(),
icon,
pad(&id, id_w).dimmed(),
pad(&name, name_w),
suffix,
)
}
fn state_icon(state: State) -> colored::ColoredString {
match state {
State::Finished => "✓".green(),
State::Downloading => "▶".yellow(),
State::Paused => "⏸".cyan(),
State::Error => "✗".red(),
State::Stale => "⚠".red(),
State::Pending | State::Unspecified => "·".dimmed(),
}
}
fn format_pct_inline(percent: u32) -> String {
let filled = (percent as usize * BAR_WIDTH) / 100;
let empty = BAR_WIDTH - filled;
format!(
"{}{} {:>3}%",
"█".repeat(filled),
"░".repeat(empty),
percent
)
}
fn format_list_long(torrents: &[TorrentStatus]) -> String {
if torrents.is_empty() {
return " no torrents".dimmed().to_string();
@@ -301,6 +407,7 @@ fn colored_state(state: State) -> colored::ColoredString {
State::Downloading => s.yellow(),
State::Paused => s.cyan(),
State::Error => s.red().bold(),
State::Stale => s.red(),
State::Pending | State::Unspecified => s.dimmed(),
}
}
@@ -312,6 +419,7 @@ fn pad_colored(s: &str, width: usize, state: State) -> colored::ColoredString {
State::Downloading => padded.yellow(),
State::Paused => padded.cyan(),
State::Error => padded.red().bold(),
State::Stale => padded.red(),
State::Pending | State::Unspecified => padded.dimmed(),
}
}
crates/torad/Cargo.toml
+3 -2
View File
@@ -7,8 +7,9 @@ edition = "2024"
anyhow = "1.0.103"
async-trait = "0.1"
clap = { version = "4.6.1", features = ["derive", "env"] }
librqbit = { version = "8.1.1", default-features = false, features = ["rust-tls", "http-api-client"] }
librqbit-core = { version = "5.0.0", default-features = false }
librqbit = { version = "=9.0.0-rc.0", default-features = false, features = ["rust-tls", "http-api-client"] }
librqbit-core = { version = "=9.0.0-rc.0", default-features = false }
nix = { version = "0.31.3", features = ["user", "sched", "fs", "process", "mount"] }
reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }
sha1 = "0.10"
sqlx = { version = "0.9.0", features = ["postgres", "runtime-tokio", "tls-rustls", "uuid"] }
crates/torad/src/db.rs
+43 -15
View File
@@ -23,6 +23,7 @@ pub enum TorrentState {
Paused,
Finished,
Error,
Stale,
}
#[derive(Debug, Clone, sqlx::FromRow)]
@@ -31,7 +32,9 @@ pub struct TorrentRow {
pub info_hash: String,
pub name: Option<String>,
pub source: String,
pub output_path: String,
pub output_path: Option<String>,
pub download_folder: Option<String>,
pub move_after_completion_folder: Option<String>,
pub total_bytes: Option<i64>,
pub downloaded_bytes: i64,
pub state: TorrentState,
@@ -42,17 +45,20 @@ pub async fn insert_pending(
pool: &PgPool,
info_hash: &str,
source: &str,
output_path: &str,
download_folder: Option<&str>,
move_after_completion_folder: Option<&str>,
) -> Result<TorrentRow> {
let row = sqlx::query_as::<_, TorrentRow>(
"INSERT INTO torrents (info_hash, source, output_path)
VALUES ($1, $2, $3)
"INSERT INTO torrents (info_hash, source, download_folder, move_after_completion_folder)
VALUES ($1, $2, $3, $4)
ON CONFLICT (info_hash) DO UPDATE SET updated_at = now()
RETURNING id, info_hash, name, source, output_path, total_bytes, downloaded_bytes, state, error_message",
RETURNING id, info_hash, name, source, output_path, download_folder, \
move_after_completion_folder, total_bytes, downloaded_bytes, state, error_message",
)
.bind(info_hash)
.bind(source)
.bind(output_path)
.bind(download_folder)
.bind(move_after_completion_folder)
.fetch_one(pool)
.await?;
Ok(row)
@@ -60,7 +66,8 @@ pub async fn insert_pending(
pub async fn get(pool: &PgPool, id: Uuid) -> Result<Option<TorrentRow>> {
let row = sqlx::query_as::<_, TorrentRow>(
"SELECT id, info_hash, name, source, output_path, total_bytes, downloaded_bytes, state, error_message
"SELECT id, info_hash, name, source, output_path, download_folder, \
move_after_completion_folder, total_bytes, downloaded_bytes, state, error_message
FROM torrents WHERE id = $1",
)
.bind(id)
@@ -71,7 +78,8 @@ pub async fn get(pool: &PgPool, id: Uuid) -> Result<Option<TorrentRow>> {
pub async fn list(pool: &PgPool) -> Result<Vec<TorrentRow>> {
let rows = sqlx::query_as::<_, TorrentRow>(
"SELECT id, info_hash, name, source, output_path, total_bytes, downloaded_bytes, state, error_message
"SELECT id, info_hash, name, source, output_path, download_folder, \
move_after_completion_folder, total_bytes, downloaded_bytes, state, error_message
FROM torrents ORDER BY added_at DESC",
)
.fetch_all(pool)
@@ -82,7 +90,8 @@ pub async fn list(pool: &PgPool) -> Result<Vec<TorrentRow>> {
pub async fn delete(pool: &PgPool, id: Uuid) -> Result<Option<TorrentRow>> {
let row = sqlx::query_as::<_, TorrentRow>(
"DELETE FROM torrents WHERE id = $1
RETURNING id, info_hash, name, source, output_path, total_bytes, downloaded_bytes, state, error_message",
RETURNING id, info_hash, name, source, output_path, download_folder, \
move_after_completion_folder, total_bytes, downloaded_bytes, state, error_message",
)
.bind(id)
.fetch_optional(pool)
@@ -92,7 +101,8 @@ pub async fn delete(pool: &PgPool, id: Uuid) -> Result<Option<TorrentRow>> {
pub async fn list_pending(pool: &PgPool) -> Result<Vec<TorrentRow>> {
let rows = sqlx::query_as::<_, TorrentRow>(
"SELECT id, info_hash, name, source, output_path, total_bytes, downloaded_bytes, state, error_message
"SELECT id, info_hash, name, source, output_path, download_folder, \
move_after_completion_folder, total_bytes, downloaded_bytes, state, error_message
FROM torrents WHERE state = 'pending' ORDER BY added_at",
)
.fetch_all(pool)
@@ -118,6 +128,18 @@ pub async fn set_state(pool: &PgPool, id: Uuid, state: TorrentState) -> Result<(
Ok(())
}
/// Used by the poller to record the actual output path once librqbit
/// resolves metadata and creates the directory (or after a post-completion
/// move). Leaves all other columns untouched.
pub async fn update_output_path(pool: &PgPool, id: Uuid, output_path: &str) -> Result<()> {
sqlx::query("UPDATE torrents SET output_path = $2, updated_at = now() WHERE id = $1")
.bind(id)
.bind(output_path)
.execute(pool)
.await?;
Ok(())
}
pub struct Progress<'a> {
pub name: Option<&'a str>,
pub total_bytes: i64,
@@ -126,8 +148,12 @@ pub struct Progress<'a> {
pub error_message: Option<&'a str>,
}
pub async fn update_progress(pool: &PgPool, id: Uuid, progress: Progress<'_>) -> Result<()> {
sqlx::query(
pub async fn update_progress(
pool: &PgPool,
id: Uuid,
progress: Progress<'_>,
) -> Result<TorrentRow> {
let row = sqlx::query_as::<_, TorrentRow>(
"UPDATE torrents
SET name = COALESCE($2, name),
total_bytes = $3,
@@ -136,7 +162,9 @@ pub async fn update_progress(pool: &PgPool, id: Uuid, progress: Progress<'_>) ->
error_message = $6,
updated_at = now(),
completed_at = CASE WHEN $5 = 'finished' THEN now() ELSE completed_at END
WHERE id = $1",
WHERE id = $1
RETURNING id, info_hash, name, source, output_path, download_folder, \
move_after_completion_folder, total_bytes, downloaded_bytes, state, error_message",
)
.bind(id)
.bind(progress.name)
@@ -144,7 +172,7 @@ pub async fn update_progress(pool: &PgPool, id: Uuid, progress: Progress<'_>) ->
.bind(progress.downloaded_bytes)
.bind(progress.state)
.bind(progress.error_message)
.execute(pool)
.fetch_one(pool)
.await?;
Ok(())
Ok(row)
}
crates/torad/src/main.rs
+132 -11
View File
@@ -1,6 +1,7 @@
mod db;
mod source;
mod torrents;
mod vpn;
use std::path::PathBuf;
use std::time::Duration;
@@ -39,6 +40,22 @@ struct Args {
/// (e.g. `kill $(cat /tmp/torad.pid)`).
#[arg(long, env = "TORAD_PID_FILE")]
pid_file: Option<PathBuf>,
/// Network device to bind torrent sockets to (Linux: SO_BINDTODEVICE).
/// When set, all BT peer, tracker, and DHT traffic egresses only through
/// this interface. When unset, librqbit uses default routing.
///
/// Ignored when --wireguard-config is given, which binds to its own tunnel.
#[arg(long, env = "TORAD_BIND_DEVICE")]
bind_device: Option<String>,
/// Path to a wg-quick style WireGuard configuration file.
///
/// When set, torad re-executes itself inside a private user+network
/// namespace, brings the tunnel up in user space, and binds all torrent
/// traffic to it. Requires no root and no host setup. Linux only.
#[arg(long, env = "TORAD_WIREGUARD_CONFIG")]
wireguard_config: Option<PathBuf>,
}
fn default_socket_path() -> PathBuf {
@@ -56,27 +73,102 @@ fn default_download_dir() -> PathBuf {
.join("Downloads")
}
#[tokio::main]
async fn main() -> Result<()> {
/// Deliberately not `#[tokio::main]`.
///
/// VPN mode re-executes this process inside a network namespace, and nothing
/// survives `execvpe` — a runtime built before that point would leave worker
/// threads and their fds behind in the old image. So the namespace bootstrap
/// runs first, on a bare thread, and the runtime is built afterwards.
fn main() -> Result<()> {
tracing_subscriber::fmt::init();
let args = Args::parse();
let socket = args.socket.unwrap_or_else(default_socket_path);
let pid_file = args.pid_file.unwrap_or_else(default_pid_file_path);
let download_dir = args.download_dir.unwrap_or_else(default_download_dir);
let tunnel = match &args.wireguard_config {
// First launch with VPN mode requested: check the environment, prove
// the config parses before throwing the process away, then re-exec.
Some(path) if !vpn::namespace::already_inside() => {
vpn::namespace::preflight()?;
load_wireguard_config(path)?;
// Written here, on the host, and deliberately not again inside.
// pasta puts torad in a PID namespace, so the inner process sees
// itself as PID 1 and would record a number that names init on the
// host. `execvpe` keeps this PID for pasta, and killing pasta tears
// the namespace down with everything in it — so this is both the
// correct number and the right process to signal.
write_pid_file(&pid_file_path(&args), std::process::id())?;
let never = vpn::namespace::reexec_under_pasta()?;
match never {}
}
// The re-executed process: we are inside the namespace now.
Some(path) => Some(bring_up_tunnel(path)?),
None => None,
};
tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.context("failed to build the tokio runtime")?
.block_on(run(args, tunnel))
}
fn pid_file_path(args: &Args) -> PathBuf {
args.pid_file.clone().unwrap_or_else(default_pid_file_path)
}
fn write_pid_file(path: &std::path::Path, pid: u32) -> Result<()> {
std::fs::write(path, pid.to_string())
.with_context(|| format!("failed to write pid file at {}", path.display()))
}
fn load_wireguard_config(path: &std::path::Path) -> Result<vpn::WireguardConfig> {
let text = std::fs::read_to_string(path)
.with_context(|| format!("failed to read WireGuard config at {}", path.display()))?;
vpn::parse(&text).with_context(|| format!("failed to parse {}", path.display()))
}
/// Enter a mount namespace and bring the tunnel up. Runs in the re-executed
/// process, which already holds `CAP_NET_ADMIN` in its own user namespace.
fn bring_up_tunnel(path: &std::path::Path) -> Result<vpn::Tunnel> {
let config = load_wireguard_config(path)?;
let socket_dir = vpn::namespace::enter_mount_namespace()?;
vpn::tunnel::up(&config, socket_dir)
}
async fn run(args: Args, tunnel: Option<vpn::Tunnel>) -> Result<()> {
let socket = args.socket.clone().unwrap_or_else(default_socket_path);
let pid_file = pid_file_path(&args);
let download_dir = args
.download_dir
.clone()
.unwrap_or_else(default_download_dir);
let pool = db::connect(&args.database_url)
.await
.context("failed to connect to Postgres")?;
info!("connected to Postgres");
let manager = TorrentManager::new(pool.clone(), download_dir)
// In VPN mode the tunnel is the only interface torrent traffic may use, so
// it overrides --bind-device rather than competing with it.
let bind_device = if tunnel.is_some() {
if args.bind_device.is_some() {
info!("--bind-device is ignored in VPN mode; binding to the tunnel instead");
}
Some(vpn::INTERFACE.to_owned())
} else {
args.bind_device
};
let manager = TorrentManager::new(pool.clone(), download_dir, bind_device)
.await
.context("failed to start torrent manager")?;
manager.clone().spawn_poller();
std::fs::write(&pid_file, std::process::id().to_string())
.with_context(|| format!("failed to write pid file at {}", pid_file.display()))?;
// In VPN mode the host-visible pid was already recorded before the
// re-exec; overwriting it here would replace it with this process's
// namespace-local PID 1.
if !vpn::namespace::already_inside() {
write_pid_file(&pid_file, std::process::id())?;
}
if socket.exists() {
std::fs::remove_file(&socket)
@@ -91,10 +183,24 @@ async fn main() -> Result<()> {
info!(socket = %socket.display(), "torad listening");
// A live torad with a dead tunnel is the failure mode that leaks, so the
// datapath exiting is treated exactly like SIGTERM.
let tunnel_down = tunnel.map(|mut tunnel| {
let (tx, rx) = tokio::sync::oneshot::channel();
std::thread::spawn(move || {
let reason = match tunnel.wait() {
Ok(status) => format!("wireguard-go exited with {status}"),
Err(err) => format!("lost track of wireguard-go: {err:#}"),
};
let _ = tx.send(reason);
});
rx
});
Server::builder()
.add_service(health_service)
.add_service(TorrentsServer::new(GrpcTorrents::new(manager)))
.serve_with_incoming_shutdown(uds_stream, shutdown_signal())
.serve_with_incoming_shutdown(uds_stream, shutdown_signal(tunnel_down))
.await
.context("gRPC server error")?;
@@ -124,14 +230,29 @@ async fn poll_db_health(pool: sqlx::PgPool, reporter: HealthReporter) {
}
}
async fn shutdown_signal() {
async fn shutdown_signal(tunnel_down: Option<tokio::sync::oneshot::Receiver<String>>) {
use tokio::signal::unix::{SignalKind, signal};
let mut sigint = signal(SignalKind::interrupt()).expect("failed to install SIGINT handler");
let mut sigterm = signal(SignalKind::terminate()).expect("failed to install SIGTERM handler");
// Without a tunnel there is nothing to watch, so this branch must never
// complete — otherwise it would fire immediately and shut torad down.
let tunnel = async {
match tunnel_down {
Some(rx) => match rx.await {
Ok(reason) => reason,
Err(_) => std::future::pending().await,
},
None => std::future::pending().await,
}
};
tokio::select! {
_ = sigint.recv() => {},
_ = sigterm.recv() => {},
reason = tunnel => {
error!(%reason, "VPN tunnel went down; shutting down rather than routing around it");
},
}
}
crates/torad/src/source/http_torrent.rs
+4 -4
View File
@@ -11,12 +11,12 @@ use super::info_hash;
use super::{Resolved, SourceHandler};
pub(crate) struct HttpTorrentHandler {
client: reqwest::Client,
clients: std::sync::Arc<super::HttpClients>,
}
impl HttpTorrentHandler {
pub(crate) fn new(client: reqwest::Client) -> Self {
Self { client }
pub(crate) fn new(clients: std::sync::Arc<super::HttpClients>) -> Self {
Self { clients }
}
}
@@ -27,7 +27,7 @@ impl SourceHandler for HttpTorrentHandler {
}
async fn resolve(&self, source: &str) -> Result<Resolved> {
let outcome = http_fetch::fetch(&self.client, source).await?;
let outcome = http_fetch::fetch(self.clients.for_url(source), source).await?;
match outcome {
HttpFetchOutcome::Magnet(magnet) => {
let parsed = Magnet::parse(&magnet)
crates/torad/src/source/info_hash.rs
+3 -3
View File
@@ -13,14 +13,14 @@
//! parsing it) yields the correct hash even for spec-violating files.
use anyhow::{Context, Result};
use librqbit::torrent_from_bytes_ext;
use librqbit_core::torrent_metainfo::torrent_from_bytes;
use sha1::{Digest, Sha1};
/// Extracts the v1 info-hash from `.torrent` bytes.
pub(super) fn from_bytes(bytes: &[u8]) -> Result<String> {
// Fast path: spec-compliant torrent, librqbit's parser handles it.
if let Ok(parsed) = torrent_from_bytes_ext::<Vec<u8>>(bytes) {
return Ok(parsed.meta.info_hash.as_string());
if let Ok(parsed) = torrent_from_bytes(bytes) {
return Ok(parsed.info_hash.as_string());
}
// Fallback: locate the `info` value's byte span and SHA1 it directly.
let span = locate_info_value(bytes)
crates/torad/src/source/jackett.rs
+4 -4
View File
@@ -26,12 +26,12 @@ use super::{Resolved, SourceHandler};
const JACKETT_MARKER: &str = "jackett_apikey=";
pub(crate) struct JackettHandler {
client: reqwest::Client,
clients: std::sync::Arc<super::HttpClients>,
}
impl JackettHandler {
pub(crate) fn new(client: reqwest::Client) -> Self {
Self { client }
pub(crate) fn new(clients: std::sync::Arc<super::HttpClients>) -> Self {
Self { clients }
}
}
@@ -43,7 +43,7 @@ impl SourceHandler for JackettHandler {
}
async fn resolve(&self, source: &str) -> Result<Resolved> {
let outcome = http_fetch::fetch(&self.client, source).await?;
let outcome = http_fetch::fetch(self.clients.for_url(source), source).await?;
match outcome {
HttpFetchOutcome::Magnet(magnet) => {
let parsed = Magnet::parse(&magnet)
crates/torad/src/source/mod.rs
+207 -10
View File
@@ -48,20 +48,99 @@ pub(crate) struct SourceResolver {
handlers: Vec<Box<dyn SourceHandler>>,
}
/// Builds the HTTP client shared by the Jackett and generic-`.torrent` handlers.
///
/// Two non-default settings, both load-bearing:
///
/// - **Manual redirect policy.** Jackett 302-redirects to `magnet:` URIs that
/// reqwest refuses to follow; we inspect `Location` ourselves so the magnet
/// handler can take over.
/// - **No system proxy.** reqwest otherwise honours `HTTP_PROXY`/`ALL_PROXY`
/// from the environment. A self-hosted Jackett is typically reached over
/// loopback, and routing loopback through an inherited proxy breaks it. Note
/// this client also serves arbitrary remote `.torrent` URLs via
/// [`http_torrent`], so the setting applies to those too.
///
/// `bind_device` binds the client's sockets to one interface, mirroring what
/// librqbit does for BitTorrent traffic.
fn build_http_client(bind_device: Option<&str>) -> Result<reqwest::Client> {
let builder = reqwest::Client::builder()
.redirect(reqwest::redirect::Policy::none())
.no_proxy();
let builder = match bind_device {
Some(device) => builder.interface(device),
None => builder,
};
builder.build().context("failed to build HTTP client")
}
/// The two clients a source fetch may need.
///
/// In VPN mode a remote `.torrent` URL must leave through the tunnel, and the
/// default route already sends it there — but routing is a property of the
/// namespace's route table, while `SO_BINDTODEVICE` is enforced by the kernel
/// per socket. Binding means a future route change cannot silently reroute an
/// indexer fetch around the tunnel.
///
/// Loopback sources must *not* be bound. pasta splices the namespace's loopback
/// to the host's, which is how a self-hosted Jackett at
/// `http://localhost:9117/...` stays reachable; binding those sockets to the
/// tunnel would break them, and that traffic is deliberately off the tunnel
/// anyway since it never leaves the machine.
pub(crate) struct HttpClients {
direct: reqwest::Client,
tunnel: Option<reqwest::Client>,
}
impl HttpClients {
fn build(bind_device: Option<&str>) -> Result<Self> {
Ok(Self {
direct: build_http_client(None)?,
tunnel: bind_device
.map(|device| build_http_client(Some(device)))
.transpose()?,
})
}
/// Pick the client this URL should be fetched with.
pub(super) fn for_url(&self, url: &str) -> &reqwest::Client {
match &self.tunnel {
Some(tunnel) if !is_loopback_url(url) => tunnel,
_ => &self.direct,
}
}
}
/// Whether a URL points at this machine.
///
/// A URL that cannot be parsed is treated as non-loopback: in VPN mode the safe
/// default is the tunnel, since guessing "local" for something remote would
/// send it out unbound.
fn is_loopback_url(url: &str) -> bool {
let Ok(parsed) = reqwest::Url::parse(url) else {
return false;
};
let Some(host) = parsed.host_str() else {
return false;
};
if host.eq_ignore_ascii_case("localhost") {
return true;
}
// `host_str` keeps IPv6 literals in their bracketed form.
let host = host.strip_prefix('[').unwrap_or(host);
let host = host.strip_suffix(']').unwrap_or(host);
host.parse::<std::net::IpAddr>()
.is_ok_and(|ip| ip.is_loopback())
}
impl SourceResolver {
pub(crate) fn new() -> Result<Self> {
// Manual redirect policy: Jackett 302-redirects to `magnet:` URIs that
// reqwest would refuse to follow; we want to inspect `Location`
// ourselves so the magnet handler can take over.
let http_client = reqwest::Client::builder()
.redirect(reqwest::redirect::Policy::none())
.build()
.context("failed to build HTTP client")?;
pub(crate) fn new(bind_device: Option<&str>) -> Result<Self> {
let clients = std::sync::Arc::new(HttpClients::build(bind_device)?);
Ok(Self {
handlers: vec![
Box::new(magnet::MagnetHandler),
Box::new(jackett::JackettHandler::new(http_client.clone())),
Box::new(http_torrent::HttpTorrentHandler::new(http_client)),
Box::new(jackett::JackettHandler::new(clients.clone())),
Box::new(http_torrent::HttpTorrentHandler::new(clients)),
Box::new(file::FileHandler),
],
})
@@ -76,3 +155,121 @@ impl SourceResolver {
anyhow::bail!("no handler matched source: {source}")
}
}
#[cfg(test)]
mod tests {
use super::*;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
/// Serves exactly one minimal HTTP/1.1 response, then stops. Returns the
/// bound address so the test can build a URL for it.
async fn spawn_oneshot_server() -> std::net::SocketAddr {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
if let Ok((mut sock, _)) = listener.accept().await {
let mut buf = [0u8; 1024];
let _ = sock.read(&mut buf).await;
let _ = sock
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nhi")
.await;
let _ = sock.flush().await;
}
});
addr
}
/// Regression for [`build_http_client`]'s `.no_proxy()`: reqwest reads
/// `ALL_PROXY` at build time, so without it this request would be sent to a
/// dead proxy instead of the loopback server and fail.
///
/// The proxy URL is deliberately `http://`, not `socks5://` — reqwest is
/// built here without its `socks` feature, so a SOCKS proxy would simply be
/// ignored and the test would pass with or without `.no_proxy()`.
///
/// Env vars are process-global and `set_var` is `unsafe` under the 2024
/// edition. This is the only test in the crate that touches the
/// environment, and it restores it before yielding, so nothing else here
/// can observe the change — but it is not safe to add a second such test
/// without serialising them.
#[tokio::test]
async fn http_client_ignores_system_proxy() {
let addr = spawn_oneshot_server().await;
unsafe { std::env::set_var("ALL_PROXY", "http://127.0.0.1:1") };
let client = build_http_client(None);
unsafe { std::env::remove_var("ALL_PROXY") };
let response = client
.expect("client should build")
.get(format!("http://{addr}/"))
.send()
.await
.expect("request must reach the loopback server, not the dead proxy");
assert_eq!(response.status(), reqwest::StatusCode::OK);
assert_eq!(response.text().await.unwrap(), "hi");
}
#[test]
fn recognises_loopback_urls() {
for url in [
"http://localhost:9117/dl/x",
"http://LOCALHOST/dl/x",
"http://127.0.0.1:9117/dl/x",
// The whole 127.0.0.0/8 block, not just .0.1.
"http://127.1.2.3/dl/x",
"http://[::1]:9117/dl/x",
] {
assert!(is_loopback_url(url), "{url} should be loopback");
}
for url in [
"http://example.com/x.torrent",
"https://tracker.example.org:8080/dl",
// pasta's host-loopback alias is reachable without the tunnel, but
// it is not itself a loopback address and does not need to be:
// it is not on the tunnel's route either way.
"http://169.254.1.1:5432/",
"http://10.0.0.5/x.torrent",
"magnet:?xt=urn:btih:0000",
"not a url at all",
] {
assert!(!is_loopback_url(url), "{url} should not be loopback");
}
}
/// Without a bind device there is one client and it is used for everything.
#[test]
fn unbound_mode_uses_the_direct_client_for_every_url() {
let clients = HttpClients::build(None).unwrap();
let remote = clients.for_url("http://example.com/x.torrent");
let local = clients.for_url("http://localhost:9117/dl/x");
// Same client instance for both.
assert!(std::ptr::eq(remote, local));
assert!(std::ptr::eq(remote, &clients.direct));
}
/// With a bind device, remote fetches are enforced onto it and loopback
/// fetches are deliberately left off it.
#[test]
fn bound_mode_sends_only_remote_urls_through_the_tunnel() {
// `lo` rather than `wg0`: it exists on any host, so the client can
// actually be built. Only the selection logic is under test.
let clients = HttpClients::build(Some("lo")).unwrap();
let tunnel = clients.tunnel.as_ref().expect("tunnel client must exist");
assert!(std::ptr::eq(
clients.for_url("http://example.com/x.torrent"),
tunnel
));
assert!(std::ptr::eq(
clients.for_url("http://localhost:9117/dl/x"),
&clients.direct
));
assert!(std::ptr::eq(
clients.for_url("http://127.0.0.1:9117/dl/x"),
&clients.direct
));
}
}
crates/torad/src/torrents.rs
+871 -85
View File
File diff suppressed because it is too large Load Diff
crates/torad/src/vpn/config.rs
+797
View File
@@ -0,0 +1,797 @@
//! Parser for `wg-quick`-style WireGuard configuration files.
//!
//! Everything here is validated at parse time and stored as a typed value, not
//! a string. The values flow into `ip addr add`, route setup, and the
//! `wireguard-go` UAPI socket, none of which give a usable diagnostic when
//! handed garbage — a mistyped address surfaces there as a bare `EINVAL` or a
//! UAPI `errno` with no indication of which config line caused it. Catching it
//! here means the operator gets a file, a line number, and the offending value.
use std::fmt;
use std::net::{IpAddr, Ipv6Addr};
use std::str::FromStr;
use anyhow::{Context, Result, bail};
/// A WireGuard key: exactly 32 bytes, written in the config as 44 base64
/// characters.
///
/// The original base64 text is retained so the key can be echoed back in the
/// form the operator wrote it, and the decoded bytes so Wave 3 can render the
/// hex encoding the UAPI protocol expects without re-parsing.
#[derive(Clone, PartialEq, Eq)]
pub struct Key {
bytes: [u8; 32],
text: String,
}
impl Key {
/// Decode a 44-character base64 WireGuard key.
///
/// Errors deliberately describe only the *shape* of the problem, never the
/// input: this runs on private keys, and the error text ends up in logs.
pub fn from_base64(s: &str) -> Result<Self> {
let bytes = s.as_bytes();
if bytes.len() != 44 {
bail!(
"expected 44 base64 characters (a 32-byte key), got {}",
bytes.len()
);
}
if bytes[43] != b'=' {
bail!("expected the 44th character to be '=' padding");
}
let mut out = [0u8; 32];
let mut acc: u32 = 0;
let mut acc_bits = 0u32;
let mut written = 0usize;
for (i, &c) in bytes[..43].iter().enumerate() {
let sextet = base64_value(c)
.with_context(|| format!("invalid base64 character at position {}", i + 1))?;
acc = (acc << 6) | u32::from(sextet);
acc_bits += 6;
if acc_bits >= 8 {
acc_bits -= 8;
out[written] = ((acc >> acc_bits) & 0xff) as u8;
written += 1;
}
}
debug_assert_eq!(written, 32);
Ok(Self {
bytes: out,
text: s.to_owned(),
})
}
/// The key as base64, exactly as it appeared in the config.
///
/// Calling this on a private key puts it somewhere; that is the point of
/// making it an explicit call rather than a `Display` impl.
pub fn as_base64(&self) -> &str {
&self.text
}
/// Lowercase hex, the encoding the WireGuard UAPI protocol expects.
pub fn to_hex(&self) -> String {
use fmt::Write;
self.bytes
.iter()
.fold(String::with_capacity(64), |mut s, b| {
let _ = write!(s, "{b:02x}");
s
})
}
}
/// Redacted unconditionally: `Key` is used for both private and public keys,
/// and a `Debug` that printed the bytes would leak the private one through any
/// `{:?}` on a containing struct. Public keys are surfaced deliberately by
/// [`PeerSection`]'s `Debug`, which calls [`Key::as_base64`].
impl fmt::Debug for Key {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("Key(<redacted>)")
}
}
fn base64_value(c: u8) -> Result<u8> {
const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
ALPHABET
.iter()
.position(|&b| b == c)
.map(|p| p as u8)
.context("not a base64 character")
}
/// An IP address with a prefix length, e.g. `10.2.0.2/32` or `::/0`.
#[derive(Clone, Copy, PartialEq, Eq)]
pub struct IpCidr {
pub addr: IpAddr,
pub prefix_len: u8,
}
impl FromStr for IpCidr {
type Err = anyhow::Error;
/// A bare address with no `/` is accepted and takes the full prefix length
/// (`/32` or `/128`), matching what `wg` itself does with `AllowedIPs`.
fn from_str(s: &str) -> Result<Self> {
let (addr_part, prefix_part) = match s.split_once('/') {
Some((a, p)) => (a.trim(), Some(p.trim())),
None => (s.trim(), None),
};
let addr: IpAddr = addr_part
.parse()
.with_context(|| format!("invalid IP address {addr_part:?}"))?;
let max_prefix = if addr.is_ipv4() { 32 } else { 128 };
let prefix_len = match prefix_part {
Some(p) => {
let parsed: u8 = p
.parse()
.with_context(|| format!("invalid prefix length {p:?}"))?;
if parsed > max_prefix {
bail!("prefix length /{parsed} exceeds the maximum /{max_prefix} for {addr}");
}
parsed
}
None => max_prefix,
};
Ok(Self { addr, prefix_len })
}
}
impl fmt::Display for IpCidr {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}/{}", self.addr, self.prefix_len)
}
}
/// Renders as the config wrote it. The derived form
/// (`IpCidr { addr: 10.2.0.2, prefix_len: 32 }`) makes a list of them unreadable
/// in a log line, which is where these are most often seen.
impl fmt::Debug for IpCidr {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{self}")
}
}
/// A peer endpoint: a host (IP literal or DNS name) and a port.
///
/// The host is not resolved here. WireGuard resolves endpoint names at
/// handshake time and re-resolves on roaming, so pinning an address at parse
/// time would be wrong; this only rejects shapes that can never resolve.
#[derive(Clone, PartialEq, Eq)]
pub struct Endpoint {
pub host: String,
pub port: u16,
}
impl FromStr for Endpoint {
type Err = anyhow::Error;
fn from_str(s: &str) -> Result<Self> {
let s = s.trim();
let (host, port_part) = if let Some(rest) = s.strip_prefix('[') {
let (host, after) = rest
.split_once(']')
.context("unterminated IPv6 literal; expected [address]:port")?;
host.parse::<Ipv6Addr>()
.with_context(|| format!("invalid IPv6 address {host:?}"))?;
let port = after
.strip_prefix(':')
.context("missing ':port' after the IPv6 literal")?;
(host, port)
} else {
let (host, port) = s
.rsplit_once(':')
.with_context(|| format!("missing ':port' in endpoint {s:?}"))?;
if host.contains(':') {
bail!("a bare IPv6 address must be bracketed, e.g. [{host}]:{port}");
}
(host, port)
};
if host.is_empty() {
bail!("endpoint host must not be empty");
}
if host.parse::<IpAddr>().is_err()
&& host.contains(|c: char| c.is_whitespace() || c == '/' || c == '@')
{
bail!("invalid endpoint host {host:?}");
}
let port: u16 = port_part
.parse()
.with_context(|| format!("invalid port {port_part:?}"))?;
if port == 0 {
bail!("endpoint port must not be 0");
}
Ok(Self {
host: host.to_owned(),
port,
})
}
}
impl fmt::Display for Endpoint {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
if self.host.parse::<Ipv6Addr>().is_ok() {
write!(f, "[{}]:{}", self.host, self.port)
} else {
write!(f, "{}:{}", self.host, self.port)
}
}
}
/// See [`IpCidr`]'s `Debug`: same reasoning, and it keeps the IPv6 bracketing
/// visible so a malformed endpoint reads the way `wg` would print it.
impl fmt::Debug for Endpoint {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{self}")
}
}
#[derive(Debug)]
pub struct WireguardConfig {
pub interface: InterfaceSection,
pub peer: PeerSection,
}
pub struct InterfaceSection {
pub private_key: Key,
/// One or more addresses to assign to the tunnel interface. `wg-quick`
/// allows a comma-separated list, and dual-stack providers use it.
pub addresses: Vec<IpCidr>,
/// Resolvers to write into the namespace's `resolv.conf`. Empty when the
/// config has no `DNS` line.
pub dns: Vec<IpAddr>,
}
/// Hand-written so the private key can never reach a log through a `{:?}`
/// format on this struct or anything containing it. [`Key`]'s own `Debug` is
/// already redacted; this is the second layer, and it also keeps the field
/// visibly marked as a secret at the point where it is most likely to be read.
impl fmt::Debug for InterfaceSection {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("InterfaceSection")
.field("private_key", &"<redacted>")
.field("addresses", &self.addresses)
.field("dns", &self.dns)
.finish()
}
}
pub struct PeerSection {
pub public_key: Key,
pub allowed_ips: Vec<IpCidr>,
pub endpoint: Endpoint,
pub persistent_keepalive: Option<u16>,
}
/// Hand-written only to undo [`Key`]'s blanket redaction for the public key,
/// which is not a secret and is the first thing worth seeing when a handshake
/// fails against the wrong peer.
impl fmt::Debug for PeerSection {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("PeerSection")
.field("public_key", &self.public_key.as_base64())
.field("allowed_ips", &self.allowed_ips)
.field("endpoint", &self.endpoint)
.field("persistent_keepalive", &self.persistent_keepalive)
.finish()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Section {
None,
Interface,
Peer,
}
pub fn parse(text: &str) -> Result<WireguardConfig> {
let mut section = Section::None;
let mut seen_interface = false;
let mut seen_peer = false;
let mut private_key = None;
let mut addresses = None;
let mut dns = None;
let mut public_key = None;
let mut allowed_ips = None;
let mut endpoint = None;
let mut persistent_keepalive = None;
for (index, raw_line) in text.lines().enumerate() {
let lineno = index + 1;
let line = strip_comment(raw_line).trim();
if line.is_empty() {
continue;
}
if let Some(rest) = line.strip_prefix('[') {
let name = rest
.strip_suffix(']')
.with_context(|| format!("line {lineno}: unterminated section header {line:?}"))?;
section = match name.trim() {
"Interface" => {
if seen_interface {
bail!("line {lineno}: duplicate [Interface] section");
}
seen_interface = true;
Section::Interface
}
"Peer" => {
if seen_peer {
// A second [Peer] previously overwrote the first in
// silence. torad brings up one tunnel with one peer;
// saying so beats picking a peer at random.
bail!(
"line {lineno}: multiple [Peer] sections are not supported \
(torad configures a single tunnel with one peer)"
);
}
seen_peer = true;
Section::Peer
}
other => bail!("line {lineno}: unknown section [{other}]"),
};
continue;
}
let Some((key, value)) = line.split_once('=') else {
bail!("line {lineno}: expected `Key = Value`, got {line:?}");
};
let key = key.trim();
let value = value.trim();
match section {
Section::Interface => match key {
"PrivateKey" => set_once(
&mut private_key,
// No value in the context: this is the secret.
Key::from_base64(value)
.with_context(|| format!("line {lineno}: invalid PrivateKey"))?,
key,
lineno,
)?,
"Address" => set_once(
&mut addresses,
parse_cidr_list(value)
.with_context(|| format!("line {lineno}: invalid Address {value:?}"))?,
key,
lineno,
)?,
"DNS" => set_once(
&mut dns,
parse_dns_list(value)
.with_context(|| format!("line {lineno}: invalid DNS {value:?}"))?,
key,
lineno,
)?,
_ => {}
},
Section::Peer => match key {
"PublicKey" => set_once(
&mut public_key,
Key::from_base64(value)
.with_context(|| format!("line {lineno}: invalid PublicKey {value:?}"))?,
key,
lineno,
)?,
"AllowedIPs" => set_once(
&mut allowed_ips,
parse_cidr_list(value)
.with_context(|| format!("line {lineno}: invalid AllowedIPs {value:?}"))?,
key,
lineno,
)?,
"Endpoint" => set_once(
&mut endpoint,
value
.parse::<Endpoint>()
.with_context(|| format!("line {lineno}: invalid Endpoint {value:?}"))?,
key,
lineno,
)?,
"PersistentKeepalive" => set_once(
&mut persistent_keepalive,
value.parse::<u16>().with_context(|| {
format!("line {lineno}: invalid PersistentKeepalive {value:?}")
})?,
key,
lineno,
)?,
_ => {}
},
Section::None => {
bail!("line {lineno}: key {key:?} appears before any section header")
}
}
}
if !seen_interface {
bail!("missing [Interface] section");
}
if !seen_peer {
bail!("missing [Peer] section");
}
let interface = InterfaceSection {
private_key: private_key
.context("missing required key 'PrivateKey' in [Interface] section")?,
addresses: addresses.context("missing required key 'Address' in [Interface] section")?,
dns: dns.unwrap_or_default(),
};
let peer = PeerSection {
public_key: public_key.context("missing required key 'PublicKey' in [Peer] section")?,
allowed_ips: allowed_ips.context("missing required key 'AllowedIPs' in [Peer] section")?,
endpoint: endpoint.context("missing required key 'Endpoint' in [Peer] section")?,
persistent_keepalive,
};
Ok(WireguardConfig { interface, peer })
}
/// Assign to a slot that must not already be occupied.
///
/// A repeated key used to overwrite silently, so a config carrying two
/// `Endpoint` lines would connect to whichever came last with no hint that the
/// other existed.
fn set_once<T>(slot: &mut Option<T>, value: T, key: &str, lineno: usize) -> Result<()> {
if slot.is_some() {
bail!("line {lineno}: duplicate key {key:?}");
}
*slot = Some(value);
Ok(())
}
fn parse_cidr_list(value: &str) -> Result<Vec<IpCidr>> {
value
.split(',')
.map(|part| {
let part = part.trim();
if part.is_empty() {
bail!("empty entry in comma-separated list");
}
part.parse::<IpCidr>()
})
.collect()
}
fn parse_dns_list(value: &str) -> Result<Vec<IpAddr>> {
value
.split(',')
.map(|part| {
let part = part.trim();
part.parse::<IpAddr>().with_context(|| {
format!(
"{part:?} is not an IP address \
(search-domain entries in DNS are not supported)"
)
})
})
.collect()
}
/// Strip inline comments (# ...) but preserve the rest.
/// Only strips when `#` is preceded by whitespace or is at column 0.
fn strip_comment(line: &str) -> &str {
if let Some(pos) = line.find('#')
&& (pos == 0 || line.as_bytes()[pos - 1] == b' ' || line.as_bytes()[pos - 1] == b'\t')
{
return &line[..pos];
}
line
}
#[cfg(test)]
mod tests {
use super::*;
/// Valid 32-byte keys. The previous fixtures (`aaaa`, and a 41-character
/// private key) are exactly what parse-time validation now rejects.
const PRIVATE_KEY: &str = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=";
const PUBLIC_KEY: &str = "ICEiIyQlJicoKSorLC0uLzAxMjM0NTY3ODk6Ozw9Pj8=";
fn minimal() -> String {
format!(
"[Interface]\n\
PrivateKey = {PRIVATE_KEY}\n\
Address = 10.0.0.2/32\n\
\n\
[Peer]\n\
PublicKey = {PUBLIC_KEY}\n\
AllowedIPs = 0.0.0.0/0\n\
Endpoint = 1.2.3.4:51820\n"
)
}
fn v4(a: u8, b: u8, c: u8, d: u8, prefix_len: u8) -> IpCidr {
IpCidr {
addr: IpAddr::from([a, b, c, d]),
prefix_len,
}
}
fn protonvpn_sample() -> String {
format!(
"[Interface]\n\
# Key for nixarr\n\
# Bouncing = 1\n\
# NAT-PMP (Port Forwarding) = on\n\
PrivateKey = {PRIVATE_KEY}\n\
Address = 10.2.0.2/32\n\
DNS = 10.2.0.1\n\
\n\
[Peer]\n\
# UA#44\n\
PublicKey = {PUBLIC_KEY}\n\
AllowedIPs = 0.0.0.0/0, ::/0\n\
Endpoint = 156.146.50.5:51820\n\
PersistentKeepalive = 25\n"
)
}
#[test]
fn parses_protonvpn_sample() {
let cfg = parse(&protonvpn_sample()).unwrap();
assert_eq!(cfg.interface.private_key.as_base64(), PRIVATE_KEY);
assert_eq!(cfg.interface.addresses, vec![v4(10, 2, 0, 2, 32)]);
assert_eq!(cfg.interface.dns, vec![IpAddr::from([10, 2, 0, 1])]);
assert_eq!(cfg.peer.public_key.as_base64(), PUBLIC_KEY);
assert_eq!(cfg.peer.allowed_ips.len(), 2);
assert_eq!(cfg.peer.allowed_ips[0], v4(0, 0, 0, 0, 0));
assert_eq!(cfg.peer.allowed_ips[1].to_string(), "::/0");
assert_eq!(cfg.peer.endpoint.host, "156.146.50.5");
assert_eq!(cfg.peer.endpoint.port, 51820);
assert_eq!(cfg.peer.persistent_keepalive, Some(25));
}
#[test]
fn parses_config_without_dns_or_keepalive() {
let cfg = parse(&minimal()).unwrap();
assert!(cfg.interface.dns.is_empty());
assert!(cfg.peer.persistent_keepalive.is_none());
}
#[test]
fn parses_multiple_addresses_and_allowed_ips() {
let text = minimal()
.replace(
"Address = 10.0.0.2/32",
"Address = 10.0.0.2/32, fd00::2/128",
)
.replace(
"AllowedIPs = 0.0.0.0/0",
"AllowedIPs = 0.0.0.0/0, ::/0, 192.168.0.0/16",
);
let cfg = parse(&text).unwrap();
assert_eq!(cfg.interface.addresses.len(), 2);
assert_eq!(cfg.interface.addresses[1].to_string(), "fd00::2/128");
assert_eq!(cfg.peer.allowed_ips.len(), 3);
assert_eq!(cfg.peer.allowed_ips[2], v4(192, 168, 0, 0, 16));
}
/// `wg` treats a prefixless `AllowedIPs` entry as a host route; so do we.
#[test]
fn bare_address_takes_the_full_prefix() {
let text = minimal().replace("AllowedIPs = 0.0.0.0/0", "AllowedIPs = 8.8.8.8, fd00::1");
let cfg = parse(&text).unwrap();
assert_eq!(cfg.peer.allowed_ips[0], v4(8, 8, 8, 8, 32));
assert_eq!(cfg.peer.allowed_ips[1].prefix_len, 128);
}
#[test]
fn parses_bracketed_ipv6_endpoint() {
let text = minimal().replace("Endpoint = 1.2.3.4:51820", "Endpoint = [2001:db8::1]:51820");
let cfg = parse(&text).unwrap();
assert_eq!(cfg.peer.endpoint.host, "2001:db8::1");
assert_eq!(cfg.peer.endpoint.port, 51820);
assert_eq!(cfg.peer.endpoint.to_string(), "[2001:db8::1]:51820");
}
#[test]
fn parses_hostname_endpoint() {
let text = minimal().replace(
"Endpoint = 1.2.3.4:51820",
"Endpoint = node-ua-44.protonvpn.net:51820",
);
let cfg = parse(&text).unwrap();
assert_eq!(cfg.peer.endpoint.host, "node-ua-44.protonvpn.net");
assert_eq!(
cfg.peer.endpoint.to_string(),
"node-ua-44.protonvpn.net:51820"
);
}
#[test]
fn ignores_comment_lines_and_strips_inline_comments() {
let text = minimal().replace(
"Address = 10.0.0.2/32",
"# a comment\nAddress = 10.0.0.2/32 # the VPN address",
);
let cfg = parse(&text).unwrap();
assert_eq!(cfg.interface.addresses, vec![v4(10, 0, 0, 2, 32)]);
}
fn parse_err(text: &str) -> String {
// `{:#}` renders the whole anyhow chain, so assertions can match on the
// inner cause as well as the line-number context wrapped around it.
format!("{:#}", parse(text).unwrap_err())
}
#[test]
fn rejects_missing_required_keys() {
let no_private = minimal().replace(&format!("PrivateKey = {PRIVATE_KEY}\n"), "");
assert!(parse_err(&no_private).contains("PrivateKey"));
let no_peer_section =
format!("[Interface]\nPrivateKey = {PRIVATE_KEY}\nAddress = 10.0.0.2/32\n");
assert!(parse_err(&no_peer_section).contains("[Peer]"));
let no_endpoint = minimal().replace("Endpoint = 1.2.3.4:51820\n", "");
assert!(parse_err(&no_endpoint).contains("Endpoint"));
}
#[test]
fn rejects_a_line_that_is_not_key_equals_value() {
let text = minimal().replace("Address = 10.0.0.2/32", "Address 10.0.0.2/32");
let err = parse_err(&text);
assert!(err.contains("line 3"), "want the line number, got: {err}");
assert!(err.contains("Key = Value"), "got: {err}");
}
#[test]
fn rejects_a_second_peer_section() {
let text = format!("{}\n[Peer]\nPublicKey = {PUBLIC_KEY}\n", minimal());
let err = parse_err(&text);
assert!(err.contains("line 10"), "want the line number, got: {err}");
assert!(err.contains("multiple [Peer]"), "got: {err}");
}
#[test]
fn rejects_a_second_interface_section() {
let text = format!("{}\n[Interface]\nAddress = 10.0.0.3/32\n", minimal());
assert!(parse_err(&text).contains("duplicate [Interface]"));
}
#[test]
fn rejects_a_duplicate_key() {
let text = minimal().replace(
"Endpoint = 1.2.3.4:51820",
"Endpoint = 1.2.3.4:51820\nEndpoint = 5.6.7.8:51820",
);
let err = parse_err(&text);
assert!(err.contains("duplicate key"), "got: {err}");
assert!(err.contains("Endpoint"), "got: {err}");
}
#[test]
fn rejects_an_unknown_section() {
let text = minimal().replace("[Peer]", "[Peers]");
assert!(parse_err(&text).contains("unknown section [Peers]"));
}
#[test]
fn rejects_a_key_before_any_section() {
let text = format!("Address = 10.0.0.2/32\n{}", minimal());
assert!(parse_err(&text).contains("before any section header"));
}
#[test]
fn rejects_invalid_addresses() {
for bad in [
"10.0.0.256/32", // octet out of range
"10.0.0.2/33", // prefix too long for v4
"fd00::1/129", // prefix too long for v6
"10.0.0.2/", // empty prefix
"not-an-ip/32",
"10.0.0.2, ", // empty trailing entry
] {
let text = minimal().replace("Address = 10.0.0.2/32", &format!("Address = {bad}"));
assert!(
parse(&text).is_err(),
"expected {bad:?} to be rejected as an Address"
);
}
}
#[test]
fn rejects_invalid_endpoints() {
for bad in [
"1.2.3.4", // no port
"1.2.3.4:0", // reserved port
"1.2.3.4:70000", // port out of range
"1.2.3.4:http", // non-numeric port
"2001:db8::1:51820", // unbracketed IPv6
"[2001:db8::1]", // bracketed, no port
"[2001:db8::zz]:51820", // bad IPv6 literal
":51820", // empty host
] {
let text = minimal().replace("Endpoint = 1.2.3.4:51820", &format!("Endpoint = {bad}"));
assert!(
parse(&text).is_err(),
"expected {bad:?} to be rejected as an Endpoint"
);
}
}
#[test]
fn rejects_a_non_ip_dns_entry() {
let text = minimal().replace(
"Address = 10.0.0.2/32",
"Address = 10.0.0.2/32\nDNS = example.com",
);
assert!(parse_err(&text).contains("search-domain"));
}
#[test]
fn rejects_malformed_keys() {
// A 41-character key: long enough to look right at a glance, which is
// precisely why it needs to fail here rather than at the UAPI socket.
let short = "aFzq1Vzq1Vzq1Vzq1Vzq1Vzq1Vzq1Vzq1Vzq1Vzq=";
assert_eq!(short.len(), 41);
for bad in [short, "aaaa", "", &"A".repeat(44)] {
let text = minimal().replace(
&format!("PublicKey = {PUBLIC_KEY}"),
&format!("PublicKey = {bad}"),
);
assert!(parse(&text).is_err(), "expected key {bad:?} to be rejected");
}
}
#[test]
fn private_key_never_appears_in_an_error() {
let text = minimal().replace(
&format!("PrivateKey = {PRIVATE_KEY}"),
"PrivateKey = SECRETSECRETSECRETSECRETSECRETSECRETSECRET=",
);
let err = parse_err(&text);
assert!(!err.contains("SECRET"), "private key leaked: {err}");
assert!(err.contains("PrivateKey"), "got: {err}");
}
#[test]
fn debug_redacts_the_private_key_but_not_the_public_one() {
let cfg = parse(&protonvpn_sample()).unwrap();
let rendered = format!("{cfg:?}");
assert!(
!rendered.contains(PRIVATE_KEY),
"private key leaked into Debug output: {rendered}"
);
assert!(rendered.contains("<redacted>"), "got: {rendered}");
// The rest of the config must still be inspectable.
assert!(rendered.contains(PUBLIC_KEY), "got: {rendered}");
assert!(rendered.contains("10.2.0.2/32"), "got: {rendered}");
}
/// A bare `Key` must stay redacted wherever it is printed, since the same
/// type carries private keys.
#[test]
fn key_debug_is_redacted() {
let key = Key::from_base64(PRIVATE_KEY).unwrap();
assert_eq!(format!("{key:?}"), "Key(<redacted>)");
}
/// Wave 3 sends keys to the `wireguard-go` UAPI socket as hex.
#[test]
fn key_round_trips_base64_to_hex() {
let key = Key::from_base64(PRIVATE_KEY).unwrap();
// PRIVATE_KEY is base64 of the bytes 0x00..=0x1f.
let expected: String = (0u8..32).map(|b| format!("{b:02x}")).collect();
assert_eq!(key.to_hex(), expected);
assert_eq!(key.to_hex().len(), 64);
let all_ones = Key::from_base64("//////////////////////////////////////////8=").unwrap();
assert_eq!(all_ones.to_hex(), "ff".repeat(32));
}
}
crates/torad/src/vpn/mod.rs
+88
View File
@@ -0,0 +1,88 @@
//! VPN mode: bring up a WireGuard tunnel from a config file, entirely in user
//! space, and confine every torrent socket to it.
//!
//! # Using it
//!
//! ```text
//! torad --wireguard-config /path/to/wg0.conf ...
//! ```
//!
//! That is the whole interface. No `sudo`, no `wg-quick`, no pre-created
//! interface, no kernel module, and no one-time host setup. `--bind-device` is
//! ignored when this is given; the tunnel is the bind device.
//!
//! Requires `pasta` (from passt), `wireguard-go`, and `ip` on `PATH`.
//! [`namespace::preflight`] checks for all three before anything is torn down,
//! so a missing one fails immediately with a message naming it.
//!
//! # How it works
//!
//! ```text
//! host │ torad's user+network namespace
//! ────────────────────────────────┼────────────────────────────────────────
//! aggregator ──unix socket───────────► torad gRPC server
//! postgres, jackett ◄────────────┐ (mount ns shares the path, so the
//! │ host still reaches the socket)
//! │
//! │ pasta ── tap iface, host addressing
//! │ ├─ 127.0.0.1 → host loopback
//! │ └─ 169.254.1.1 → host loopback
//! │
//! │ wireguard-go ── wg0 (TUN)
//! │ └─ default route ──► internet
//! │
//! │ librqbit: SO_BINDTODEVICE(wg0)
//! │ peers, trackers, DHT, LSD
//! ```
//!
//! 1. torad re-executes itself under `pasta`, which supplies an unprivileged
//! user+network namespace with `CAP_NET_ADMIN` ([`namespace`]).
//! 2. It unshares a mount namespace so `wireguard-go` has a writable socket
//! directory ([`namespace::enter_mount_namespace`]).
//! 3. It starts `wireguard-go` and configures the device over its UAPI socket
//! ([`uapi`]) — no `wg` binary involved, so `wireguard-tools` is not a
//! dependency.
//! 4. It assigns the address, pins the peer endpoint outside the tunnel, moves
//! the default route onto `wg0`, and installs the tunnel's DNS ([`tunnel`]).
//! 5. librqbit binds every socket to `wg0` via `SO_BINDTODEVICE`.
//!
//! Host services stay reachable without any URL rewriting: pasta splices the
//! namespace's loopback to the host's, so `http://localhost:9117/...` for
//! Jackett and a loopback `--database-url` for Postgres work unchanged, and
//! deliberately do not traverse the tunnel.
//!
//! # When the tunnel drops
//!
//! Two independent mechanisms, both fail-closed:
//!
//! - `SO_BINDTODEVICE` is kernel-enforced. If `wg0` goes away, librqbit's
//! sockets error rather than falling back to the host route.
//! - torad supervises `wireguard-go` and shuts down if it exits. A live torad
//! with a dead tunnel is the failure mode that leaks.
//!
//! DNS is inside the tunnel too, so a dead tunnel means tracker hostnames stop
//! resolving rather than being resolved by the host's resolver.
//!
//! # Limitations
//!
//! - **Linux only.** User namespaces, `SO_BINDTODEVICE`, and pasta are all
//! Linux-specific.
//! - **Leech-only.** librqbit runs with `listen: None`, so there is no
//! listening socket and no uTP in either direction. Incoming peer
//! connections would need NAT-PMP port forwarding through the provider,
//! which is not implemented.
//! - **The peer endpoint is pinned at startup.** It is resolved once and given
//! a host route so the handshake does not enter its own tunnel; an endpoint
//! that roams to a new address mid-session is not followed. Restart to pick
//! up a new one.
//! - **Userspace crypto.** Throughput is lower than the kernel module's.
//! - **One peer.** [`config`] rejects configs with more than one `[Peer]`
//! rather than silently using the last.
pub mod config;
pub mod namespace;
pub mod tunnel;
pub mod uapi;
pub use config::{WireguardConfig, parse};
pub use tunnel::{INTERFACE, Tunnel};
crates/torad/src/vpn/namespace.rs
+331
View File
@@ -0,0 +1,331 @@
//! Namespace bootstrap for VPN mode.
//!
//! torad must run inside its own network namespace so that a WireGuard
//! interface can be created and every socket confined to it, without root and
//! without any one-time host setup.
//!
//! It does not build that namespace by hand. `pasta` (from passt) already
//! creates an unprivileged user+network namespace, configures a tap interface
//! with the host's addressing, and executes a command inside it — so torad
//! re-executes *itself* under pasta and continues from there. Measured inside
//! that namespace:
//!
//! ```text
//! uid_map -> 0 1000 1 (we are root in the namespace, uid 1000 outside)
//! CapEff -> 000001ffffffffff (all capabilities, including CAP_NET_ADMIN)
//! ns/net -> distinct from the host's
//! /dev/net/tun writable
//! ```
//!
//! What pasta does not give us is a private *mount* namespace, and one is
//! required: `wireguard-go` hardcodes its UAPI socket directory to
//! `/var/run/wireguard` (it is a build-time linker variable with no environment
//! override, and `WG_UAPI_FD` does not avoid it — `UAPIListen` inotify-watches
//! that path even when handed a pre-bound socket). We hold `CAP_SYS_ADMIN` in
//! our own user namespace, so [`enter_mount_namespace`] unshares one and mounts
//! a tmpfs over `/var`.
//!
//! `/var` and not `/run`: `/var/run` is a symlink to `/run`, so a tmpfs on
//! `/var` yields a fresh writable `/var/run/wireguard` while leaving the real
//! `/run` visible. That matters — torad's gRPC socket lives under
//! `/run/user/<uid>` and the aggregator connects to it from the host.
use std::ffi::{CString, OsString};
use std::fs;
use std::os::unix::ffi::{OsStrExt, OsStringExt};
use std::path::Path;
use anyhow::{Context, Result, bail};
use nix::mount::{MsFlags, mount};
use nix::sched::{CloneFlags, unshare};
/// Path to the kernel sysctl controlling how many user namespaces may be created.
const MAX_USER_NAMESPACES_PROC: &str = "/proc/sys/user/max_user_namespaces";
/// Set on the re-executed process so it knows not to re-exec again.
pub const IN_NETNS_ENV: &str = "TORAD_IN_NETNS";
/// Address at which pasta makes the host's loopback reachable.
///
/// Note that plain `127.0.0.1` also reaches the *host's* loopback from inside
/// the namespace — pasta splices namespace loopback to host loopback by
/// default, including for services bound after the namespace starts. So
/// loopback URLs for Jackett and Postgres need no rewriting. This address is
/// the explicit, documented route to the same place, and exists as a fallback
/// if that default is ever turned off.
pub const HOST_LOOPBACK: &str = "169.254.1.1";
/// Directory `wireguard-go` hardcodes for its UAPI sockets.
const WIREGUARD_SOCKET_DIR: &str = "/var/run/wireguard";
/// True when this process is already the re-executed one running inside pasta.
pub fn already_inside() -> bool {
std::env::var_os(IN_NETNS_ENV).is_some_and(|v| v == "1")
}
/// Preflight: confirm the kernel and the environment can support VPN mode.
///
/// Runs before the re-exec so a missing dependency fails with an explanation
/// rather than halfway through namespace setup.
pub fn preflight() -> Result<()> {
match fs::read_to_string(MAX_USER_NAMESPACES_PROC) {
Ok(s) => evaluate(&s)?,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
tracing::warn!(
path = MAX_USER_NAMESPACES_PROC,
"sysctl file not found; assuming unprivileged user namespaces are enabled"
);
}
Err(e) => {
return Err(e).with_context(|| format!("failed to read {MAX_USER_NAMESPACES_PROC}"));
}
}
for (binary, purpose) in [
("pasta", "creates the unprivileged network namespace"),
("wireguard-go", "provides the userspace WireGuard datapath"),
("ip", "assigns the tunnel address and routes"),
] {
which(binary).with_context(|| {
format!("VPN mode requires `{binary}` on PATH ({purpose}), and it was not found")
})?;
}
Ok(())
}
/// Decides whether the sysctl's contents permit unprivileged user namespaces.
///
/// Split out from [`preflight`] so tests exercise the real logic rather than a
/// copy of it — the file itself can't be mocked.
fn evaluate(contents: &str) -> Result<()> {
let trimmed = contents.trim();
let count: u64 = trimmed.parse().with_context(|| {
format!("failed to parse {MAX_USER_NAMESPACES_PROC} as number: {trimmed:?}")
})?;
if count == 0 {
bail!(
"unprivileged user namespaces are disabled ({MAX_USER_NAMESPACES_PROC} = 0). \
VPN mode requires them. Remediation: \
`sudo sysctl -w user.max_user_namespaces=125445`, then re-run."
);
}
Ok(())
}
/// Locate a binary on `PATH`, mirroring what `execvp` will do moments later.
fn which(binary: &str) -> Result<std::path::PathBuf> {
let path = std::env::var_os("PATH").context("PATH is not set")?;
std::env::split_paths(&path)
.map(|dir| dir.join(binary))
.find(|candidate| candidate.is_file())
.with_context(|| format!("{binary} not found on PATH"))
}
/// Replace this process with itself running inside a pasta namespace.
///
/// Returns only on failure: on success the process image is gone. It must be
/// called before any resource is acquired — before the Postgres pool, the pid
/// file, the listening socket, and any tokio runtime — because none of that
/// survives `execvpe`, and a half-opened socket would be inherited into the
/// new image as a stray fd.
pub fn reexec_under_pasta() -> Result<std::convert::Infallible> {
let exe = std::env::current_exe().context("failed to determine torad's own path")?;
let mut argv: Vec<OsString> = vec![
"pasta".into(),
// Configure the tap interface, address and routes inside the namespace.
"--config-net".into(),
// Reach host services at a stable address regardless of the host's
// own addressing.
"--map-host-loopback".into(),
HOST_LOOPBACK.into(),
// pasta's informational chatter would otherwise interleave with
// torad's structured logs on the same stderr.
"--quiet".into(),
"--".into(),
exe.into_os_string(),
];
// Skip argv[0] and pass the original arguments through verbatim, so the
// re-executed process sees exactly the command line the operator typed.
argv.extend(std::env::args_os().skip(1));
let argv: Vec<CString> = argv
.into_iter()
.map(|arg| CString::new(arg.as_bytes()).context("argument contained an interior NUL"))
.collect::<Result<_>>()?;
let mut env: Vec<CString> = std::env::vars_os()
.filter(|(key, _)| key != IN_NETNS_ENV)
.map(|(key, value)| {
let mut entry = key.into_vec();
entry.push(b'=');
entry.extend_from_slice(value.as_bytes());
CString::new(entry).context("environment entry contained an interior NUL")
})
.collect::<Result<_>>()?;
env.push(CString::new(format!("{IN_NETNS_ENV}=1")).expect("literal has no NUL"));
let program = CString::new("pasta").expect("literal has no NUL");
tracing::info!("re-executing under pasta to enter a private network namespace");
nix::unistd::execvpe(&program, &argv, &env)
.context("failed to execute pasta; VPN mode cannot continue")
}
/// Ask the kernel to signal this process when pasta exits.
///
/// pasta does not kill the command it spawned when it is itself killed: the
/// namespace loses its tap interface but torad keeps running, orphaned and
/// unreachable through the pid file that named pasta. Since that pid file is
/// the documented way to stop torad, torad has to die with pasta.
///
/// `PR_SET_PDEATHSIG` survives into this process because it was set after the
/// `execvpe`, and it fires across the PID namespace boundary — the signal is
/// tied to the parent task exiting, not to what pasta's PID looks like from in
/// here (from inside the namespace it is not visible at all).
fn die_with_parent() -> Result<()> {
nix::sys::prctl::set_pdeathsig(nix::sys::signal::Signal::SIGTERM)
.context("failed to arrange for torad to exit when pasta does")
}
/// Unshare a mount namespace and make `/var/run/wireguard` writable.
///
/// Returns the directory `wireguard-go` will place its UAPI socket in.
pub fn enter_mount_namespace() -> Result<&'static Path> {
die_with_parent()?;
unshare(CloneFlags::CLONE_NEWNS).context(
"failed to unshare a mount namespace; \
VPN mode needs one to give wireguard-go a writable socket directory",
)?;
// Without this our tmpfs could propagate back into the host's mount tree.
mount(
Some("none"),
"/",
None::<&str>,
MsFlags::MS_REC | MsFlags::MS_PRIVATE,
None::<&str>,
)
.context("failed to make the mount namespace private")?;
mount(
Some("tmpfs"),
"/var",
Some("tmpfs"),
MsFlags::empty(),
Some("mode=0755"),
)
.context("failed to mount a tmpfs over /var")?;
fs::create_dir_all(WIREGUARD_SOCKET_DIR)
.with_context(|| format!("failed to create {WIREGUARD_SOCKET_DIR}"))?;
Ok(Path::new(WIREGUARD_SOCKET_DIR))
}
/// Point the namespace's resolver at the tunnel's DNS servers.
///
/// Must be called *after* the peer endpoint has been resolved: the tunnel DNS
/// is typically only reachable through the tunnel, so installing it first would
/// leave nothing able to resolve the endpoint hostname.
///
/// The file is bind-mounted rather than written in place because `/etc` belongs
/// to root on the host and our uid maps outside the namespace.
pub fn install_resolv_conf(runtime_dir: &Path, nameservers: &[std::net::IpAddr]) -> Result<()> {
if nameservers.is_empty() {
tracing::info!("config has no DNS entry; leaving the host resolver in place");
return Ok(());
}
let contents: String = nameservers
.iter()
.map(|ip| format!("nameserver {ip}\n"))
.collect();
let source = runtime_dir.join("resolv.conf");
fs::write(&source, contents)
.with_context(|| format!("failed to write {}", source.display()))?;
mount(
Some(&source),
"/etc/resolv.conf",
None::<&str>,
MsFlags::MS_BIND,
None::<&str>,
)
.context("failed to bind-mount the tunnel's resolv.conf over /etc/resolv.conf")?;
tracing::info!(count = nameservers.len(), "installed tunnel DNS");
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn accepts_positive_count() {
assert!(evaluate("125445\n").is_ok());
assert!(evaluate("125445").is_ok());
assert!(evaluate("1\n").is_ok());
}
#[test]
fn rejects_zero_count() {
let err = evaluate("0\n").unwrap_err().to_string();
assert!(
err.contains("user.max_user_namespaces"),
"error should name the sysctl and its remediation, got: {err}"
);
}
#[test]
fn rejects_non_numeric() {
assert!(evaluate("not-a-number\n").is_err());
}
/// `evaluate` must agree with the sysctl on this host, where it is present
/// and non-zero. `preflight` itself additionally requires pasta and
/// wireguard-go on PATH, which is asserted separately below.
#[test]
fn evaluate_agrees_with_the_host_sysctl() {
if let Ok(contents) = fs::read_to_string(MAX_USER_NAMESPACES_PROC) {
assert!(evaluate(&contents).is_ok());
}
}
#[test]
fn which_finds_a_binary_that_exists_and_not_one_that_does_not() {
assert!(which("sh").is_ok(), "sh must be on PATH");
assert!(which("torad-definitely-not-a-real-binary").is_err());
}
/// A missing dependency must say which binary and why, not fail later with
/// a bare ENOENT from execvp.
#[test]
fn preflight_error_names_the_missing_binary() {
// Only meaningful when the dependency really is absent; when it is
// installed, preflight should simply pass.
match preflight() {
Ok(()) => {
assert!(which("pasta").is_ok());
assert!(which("wireguard-go").is_ok());
}
Err(err) => {
let text = format!("{err:#}");
assert!(
text.contains("pasta") || text.contains("wireguard-go") || text.contains("ip"),
"error should name the missing binary, got: {text}"
);
assert!(text.contains("PATH"), "got: {text}");
}
}
}
#[test]
fn already_inside_reads_the_marker_env_var() {
// The variable is not set in the test process.
assert!(!already_inside());
}
}
crates/torad/src/vpn/tunnel.rs
+313
View File
@@ -0,0 +1,313 @@
//! Bringing up the WireGuard tunnel inside the namespace.
//!
//! Order matters here and is not arbitrary:
//!
//! 1. Read pasta's default route *before* touching routing, because step 4
//! replaces it and we need the original gateway to pin the endpoint.
//! 2. Resolve the peer endpoint while the host resolver still works.
//! 3. Pin the endpoint to the pasta gateway with a host route. Without this,
//! installing `default dev wg0` in step 4 would send WireGuard's own
//! handshake into the tunnel it is trying to establish. `wg-quick` solves
//! this with fwmark plus policy routing; a host route is the simpler
//! equivalent for a single fixed peer.
//! 4. Only then bring up wg0 and move the default route onto it.
//! 5. Install tunnel DNS last, since it is typically only reachable through
//! the tunnel that now exists.
use std::net::{IpAddr, ToSocketAddrs};
use std::path::Path;
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};
use anyhow::{Context, Result, bail};
use super::config::WireguardConfig;
use super::{namespace, uapi};
/// Interface name for the tunnel. Fixed rather than configurable: it is also
/// what librqbit is told to bind every socket to, and nothing outside this
/// process needs to name it.
pub const INTERFACE: &str = "wg0";
/// How long to wait for `wireguard-go` to create its UAPI socket.
const UAPI_TIMEOUT: Duration = Duration::from_secs(10);
const UAPI_POLL_INTERVAL: Duration = Duration::from_millis(100);
/// A running tunnel. Dropping this kills `wireguard-go` and takes the
/// interface with it, which is the intended behaviour: no interface means
/// librqbit's bound sockets fail rather than falling back to the host route.
pub struct Tunnel {
process: Child,
}
impl Tunnel {
/// Wait for the datapath to exit. Returns the status it exited with.
///
/// A live torad with a dead tunnel is the failure mode that leaks, so the
/// caller is expected to shut down when this returns.
pub fn wait(&mut self) -> Result<std::process::ExitStatus> {
self.process
.wait()
.context("failed to wait on the wireguard-go process")
}
}
impl Drop for Tunnel {
fn drop(&mut self) {
let _ = self.process.kill();
let _ = self.process.wait();
}
}
/// Bring up the tunnel. Returns once traffic can flow through [`INTERFACE`].
pub fn up(config: &WireguardConfig, socket_dir: &Path) -> Result<Tunnel> {
let gateway = default_route().context(
"failed to read the namespace's default route; \
pasta should have installed one with --config-net",
)?;
tracing::info!(gateway = %gateway.via, device = %gateway.device, "namespace default route");
let endpoints = resolve_endpoint(config)?;
let tunnel = start_datapath(socket_dir)?;
uapi::configure(&socket_dir.join(format!("{INTERFACE}.sock")), config)?;
// Pin the peer outside the tunnel before the tunnel can capture it.
for endpoint in &endpoints {
let host_route = match endpoint {
IpAddr::V4(addr) => format!("{addr}/32"),
IpAddr::V6(addr) => format!("{addr}/128"),
};
ip(&[
"route",
"add",
&host_route,
"via",
&gateway.via,
"dev",
&gateway.device,
])
.with_context(|| format!("failed to pin peer endpoint {endpoint} to the host route"))?;
tracing::info!(%endpoint, "pinned peer endpoint outside the tunnel");
}
for address in &config.interface.addresses {
ip(&["address", "add", &address.to_string(), "dev", INTERFACE])
.with_context(|| format!("failed to assign {address} to {INTERFACE}"))?;
}
ip(&["link", "set", INTERFACE, "up"])
.with_context(|| format!("failed to bring {INTERFACE} up"))?;
for allowed in &config.peer.allowed_ips {
let destination = allowed.to_string();
// A default route must replace pasta's, not collide with it. Narrower
// routes are added, so a malformed config surfaces as an error rather
// than silently clobbering something.
let is_default = allowed.prefix_len == 0;
let verb = if is_default { "replace" } else { "add" };
let target = if is_default { "default" } else { &destination };
let family = if allowed.addr.is_ipv4() { "-4" } else { "-6" };
ip(&[family, "route", verb, target, "dev", INTERFACE])
.with_context(|| format!("failed to route {destination} through {INTERFACE}"))?;
tracing::info!(route = %destination, "routed through the tunnel");
}
namespace::install_resolv_conf(socket_dir, &config.interface.dns)?;
tracing::info!(interface = INTERFACE, "tunnel is up");
Ok(tunnel)
}
/// Spawn `wireguard-go` and wait for its UAPI socket to appear.
fn start_datapath(socket_dir: &Path) -> Result<Tunnel> {
let mut process = Command::new("wireguard-go")
// Stay in the foreground so the process is ours to supervise; the
// default is to daemonise, which would leave nothing to wait on.
.arg("-f")
.arg(INTERFACE)
.env("WG_PROCESS_FOREGROUND", "1")
.stdin(Stdio::null())
.spawn()
.context("failed to spawn wireguard-go")?;
let socket = socket_dir.join(format!("{INTERFACE}.sock"));
let deadline = Instant::now() + UAPI_TIMEOUT;
while Instant::now() < deadline {
if socket.exists() {
// Prove the device answers, not merely that the socket file exists.
if let Err(err) = uapi::get(&socket) {
let _ = process.kill();
let _ = process.wait();
return Err(err).context("wireguard-go created its socket but did not respond");
}
return Ok(Tunnel { process });
}
if let Some(status) = process
.try_wait()
.context("failed to poll the wireguard-go process")?
{
bail!("wireguard-go exited before creating its UAPI socket (status {status})");
}
std::thread::sleep(UAPI_POLL_INTERVAL);
}
let _ = process.kill();
let _ = process.wait();
bail!(
"wireguard-go did not create {} within {:?}",
socket.display(),
UAPI_TIMEOUT
)
}
#[derive(Debug)]
struct DefaultRoute {
via: String,
device: String,
}
/// Read the namespace's IPv4 default route, as installed by `pasta --config-net`.
fn default_route() -> Result<DefaultRoute> {
let output = Command::new("ip")
.args(["-4", "route", "show", "default"])
.output()
.context("failed to run `ip route show default`")?;
if !output.status.success() {
bail!(
"`ip route show default` failed: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
parse_default_route(&String::from_utf8_lossy(&output.stdout))
}
/// Pull the gateway and device out of `ip route show default` output.
fn parse_default_route(text: &str) -> Result<DefaultRoute> {
let line = text
.lines()
.find(|line| line.starts_with("default "))
.context("no default route in the namespace")?;
let fields: Vec<&str> = line.split_whitespace().collect();
let via = field_after(&fields, "via")
.context("default route has no `via` gateway")?
.to_owned();
let device = field_after(&fields, "dev")
.context("default route has no `dev` interface")?
.to_owned();
Ok(DefaultRoute { via, device })
}
fn field_after<'a>(fields: &[&'a str], key: &str) -> Option<&'a str> {
fields
.iter()
.position(|field| *field == key)
.and_then(|index| fields.get(index + 1))
.copied()
}
/// Resolve the peer endpoint to concrete addresses.
///
/// Done once, at setup. WireGuard would normally re-resolve a roaming endpoint
/// on its own, but the host route pinned in [`up`] is what keeps the handshake
/// out of the tunnel, and that route is per-address — so an endpoint that
/// changes address mid-session is not followed. Documented as a limitation
/// rather than solved with policy routing.
fn resolve_endpoint(config: &WireguardConfig) -> Result<Vec<IpAddr>> {
let endpoint = &config.peer.endpoint;
if let Ok(addr) = endpoint.host.parse::<IpAddr>() {
return Ok(vec![addr]);
}
let addresses: Vec<IpAddr> = (endpoint.host.as_str(), endpoint.port)
.to_socket_addrs()
.with_context(|| format!("failed to resolve peer endpoint host {:?}", endpoint.host))?
.map(|socket| socket.ip())
.collect();
if addresses.is_empty() {
bail!(
"peer endpoint host {:?} resolved to no addresses",
endpoint.host
);
}
tracing::info!(host = %endpoint.host, count = addresses.len(), "resolved peer endpoint");
Ok(addresses)
}
/// Run `ip` with the given arguments, failing with its stderr attached.
fn ip(args: &[&str]) -> Result<()> {
let output = Command::new("ip")
.args(args)
.output()
.with_context(|| format!("failed to run `ip {}`", args.join(" ")))?;
if !output.status.success() {
bail!(
"`ip {}` failed: {}",
args.join(" "),
String::from_utf8_lossy(&output.stderr).trim()
);
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_a_pasta_default_route() {
// Exactly what `pasta --config-net` installs, as observed in the probe.
let text = "default via 192.168.100.1 dev enp7s0f3u1u2 proto dhcp metric 100 \n";
let route = parse_default_route(text).unwrap();
assert_eq!(route.via, "192.168.100.1");
assert_eq!(route.device, "enp7s0f3u1u2");
}
#[test]
fn ignores_non_default_routes() {
let text = "192.168.100.0/24 dev eth0 proto kernel scope link\n\
default via 10.0.0.1 dev eth0\n";
let route = parse_default_route(text).unwrap();
assert_eq!(route.via, "10.0.0.1");
assert_eq!(route.device, "eth0");
}
#[test]
fn rejects_missing_or_incomplete_default_routes() {
assert!(parse_default_route("").is_err());
assert!(parse_default_route("192.168.1.0/24 dev eth0\n").is_err());
// An on-link default has no gateway to pin the endpoint to.
let err = parse_default_route("default dev eth0\n")
.unwrap_err()
.to_string();
assert!(err.contains("via"), "got: {err}");
}
#[test]
fn field_after_finds_values_and_handles_edges() {
let fields = vec!["default", "via", "10.0.0.1", "dev", "eth0"];
assert_eq!(field_after(&fields, "via"), Some("10.0.0.1"));
assert_eq!(field_after(&fields, "dev"), Some("eth0"));
assert_eq!(field_after(&fields, "proto"), None);
// Key present but nothing follows it.
assert_eq!(field_after(&["default", "via"], "via"), None);
}
#[test]
fn resolves_a_literal_endpoint_without_dns() {
let config = crate::vpn::config::parse(
"[Interface]\n\
PrivateKey = AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=\n\
Address = 10.2.0.2/32\n\
[Peer]\n\
PublicKey = ICEiIyQlJicoKSorLC0uLzAxMjM0NTY3ODk6Ozw9Pj8=\n\
AllowedIPs = 0.0.0.0/0\n\
Endpoint = 156.146.50.5:51820\n",
)
.unwrap();
let resolved = resolve_endpoint(&config).unwrap();
assert_eq!(resolved, vec![IpAddr::from([156, 146, 50, 5])]);
}
}
crates/torad/src/vpn/uapi.rs
+249
View File
@@ -0,0 +1,249 @@
//! Client for WireGuard's UAPI protocol.
//!
//! `wireguard-go` exposes a unix socket speaking a line-oriented text protocol
//! instead of the netlink interface the kernel module uses. Talking to it
//! directly means torad needs no `wg` binary at all — `wireguard-tools` is not
//! a dependency of VPN mode.
//!
//! The protocol: send an operation line (`set=1`), then `key=value` lines, then
//! a blank line. The device replies with its own `key=value` lines (empty for
//! `set`), an `errno=N` line, and a blank line. `errno=0` means success.
//! Keys are lowercase hex, not the base64 used in config files.
use std::io::{BufRead, BufReader, Write};
use std::os::unix::net::UnixStream;
use std::path::Path;
use anyhow::{Context, Result, bail};
use super::config::WireguardConfig;
/// Push a parsed config into a running `wireguard-go` device.
///
/// `replace_peers` and `replace_allowed_ips` make this idempotent: applying the
/// same config twice leaves exactly one peer rather than accumulating them.
pub fn configure(socket_path: &Path, config: &WireguardConfig) -> Result<()> {
let mut request = String::from("set=1\n");
request.push_str(&format!(
"private_key={}\n",
config.interface.private_key.to_hex()
));
request.push_str("replace_peers=true\n");
request.push_str(&format!("public_key={}\n", config.peer.public_key.to_hex()));
request.push_str(&format!("endpoint={}\n", config.peer.endpoint));
if let Some(keepalive) = config.peer.persistent_keepalive {
request.push_str(&format!("persistent_keepalive_interval={keepalive}\n"));
}
request.push_str("replace_allowed_ips=true\n");
for cidr in &config.peer.allowed_ips {
request.push_str(&format!("allowed_ip={cidr}\n"));
}
request.push('\n');
let response = round_trip(socket_path, &request)
// The request body contains the private key in hex, so it must never
// appear in an error. Only the socket path does.
.with_context(|| format!("UAPI set=1 failed on {}", socket_path.display()))?;
match parse_errno(&response)? {
0 => Ok(()),
errno => bail!(
"wireguard-go rejected the configuration (UAPI errno={errno}). \
The config parsed cleanly, so this is most likely a key the device \
would not accept."
),
}
}
/// Read back the device state. Used by the readiness check and by tests.
pub fn get(socket_path: &Path) -> Result<String> {
round_trip(socket_path, "get=1\n\n")
.with_context(|| format!("UAPI get=1 failed on {}", socket_path.display()))
}
fn round_trip(socket_path: &Path, request: &str) -> Result<String> {
let mut stream =
UnixStream::connect(socket_path).context("failed to connect to UAPI socket")?;
stream
.write_all(request.as_bytes())
.context("failed to write UAPI request")?;
stream.flush().context("failed to flush UAPI request")?;
// The device answers with lines terminated by a blank line. Reading to EOF
// would work too, but wireguard-go keeps the connection open after `set`,
// so stop at the terminator instead of blocking.
let mut reader = BufReader::new(&stream);
let mut response = String::new();
loop {
let mut line = String::new();
let read = reader
.read_line(&mut line)
.context("failed to read UAPI response")?;
if read == 0 {
break;
}
if line == "\n" {
break;
}
response.push_str(&line);
}
Ok(response)
}
/// Extract the `errno=N` line the device terminates every response with.
fn parse_errno(response: &str) -> Result<i64> {
response
.lines()
.find_map(|line| line.strip_prefix("errno="))
.context("UAPI response contained no errno line")?
.trim()
.parse()
.context("UAPI errno was not a number")
}
#[cfg(test)]
mod tests {
use super::*;
use crate::vpn::config;
use std::io::Read;
use std::os::unix::net::UnixListener;
const SAMPLE: &str = "[Interface]\n\
PrivateKey = AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=\n\
Address = 10.2.0.2/32\n\
DNS = 10.2.0.1\n\
\n\
[Peer]\n\
PublicKey = ICEiIyQlJicoKSorLC0uLzAxMjM0NTY3ODk6Ozw9Pj8=\n\
AllowedIPs = 0.0.0.0/0, ::/0\n\
Endpoint = 156.146.50.5:51820\n\
PersistentKeepalive = 25\n";
/// Stands in for wireguard-go: accepts one connection, records the request,
/// and replies with the given canned response.
fn spawn_fake_device(
dir: &Path,
reply: &'static str,
) -> (std::path::PathBuf, std::thread::JoinHandle<String>) {
let path = dir.join("wg0.sock");
let listener = UnixListener::bind(&path).unwrap();
let handle = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut request = String::new();
// The client writes a blank-line-terminated block and then waits,
// so read until that terminator rather than to EOF.
let mut byte = [0u8; 1];
while stream.read(&mut byte).unwrap() == 1 {
request.push(byte[0] as char);
if request.ends_with("\n\n") {
break;
}
}
stream.write_all(reply.as_bytes()).unwrap();
stream.flush().unwrap();
request
});
(path, handle)
}
/// Per-test directory. These tests bind a unix socket at a fixed name
/// inside it, and the test harness runs them concurrently, so sharing one
/// directory would race on both `bind` and the teardown `remove_dir_all`.
fn tempdir(test_name: &str) -> std::path::PathBuf {
let dir =
std::env::temp_dir().join(format!("torad-uapi-{}-{test_name}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
dir
}
#[test]
fn sends_a_wellformed_set_request() {
let dir = tempdir("set-request");
let (path, handle) = spawn_fake_device(&dir, "errno=0\n\n");
let config = config::parse(SAMPLE).unwrap();
configure(&path, &config).expect("errno=0 must be accepted");
let request = handle.join().unwrap();
assert!(request.starts_with("set=1\n"), "got: {request}");
assert!(request.ends_with("\n\n"), "must end with a blank line");
// Hex, not the base64 from the file: PRIVATE_KEY is bytes 0x00..=0x1f.
let private_hex: String = (0u8..32).map(|b| format!("{b:02x}")).collect();
assert!(
request.contains(&format!("private_key={private_hex}\n")),
"got: {request}"
);
let public_hex: String = (32u8..64).map(|b| format!("{b:02x}")).collect();
assert!(
request.contains(&format!("public_key={public_hex}\n")),
"got: {request}"
);
assert!(
request.contains("endpoint=156.146.50.5:51820\n"),
"got: {request}"
);
assert!(
request.contains("persistent_keepalive_interval=25\n"),
"got: {request}"
);
assert!(request.contains("allowed_ip=0.0.0.0/0\n"), "got: {request}");
assert!(request.contains("allowed_ip=::/0\n"), "got: {request}");
// Without these, re-applying a config would stack duplicate peers.
assert!(request.contains("replace_peers=true\n"), "got: {request}");
assert!(
request.contains("replace_allowed_ips=true\n"),
"got: {request}"
);
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn rejects_a_nonzero_errno() {
let dir = tempdir("nonzero-errno");
let (path, handle) = spawn_fake_device(&dir, "errno=22\n\n");
let config = config::parse(SAMPLE).unwrap();
let err = configure(&path, &config).unwrap_err();
handle.join().unwrap();
let text = format!("{err:#}");
assert!(text.contains("errno=22"), "got: {text}");
std::fs::remove_dir_all(&dir).ok();
}
/// The request carries the private key in hex; a failure must not echo it.
#[test]
fn errors_never_contain_the_private_key() {
let dir = tempdir("no-key-leak");
let (path, handle) = spawn_fake_device(&dir, "errno=22\n\n");
let config = config::parse(SAMPLE).unwrap();
let err = configure(&path, &config).unwrap_err();
handle.join().unwrap();
let private_hex: String = (0u8..32).map(|b| format!("{b:02x}")).collect();
let text = format!("{err:#}");
assert!(!text.contains(&private_hex), "private key leaked: {text}");
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn parses_errno_from_a_full_response() {
let response = "private_key=00\nlisten_port=51820\npublic_key=11\nerrno=0\n";
assert_eq!(parse_errno(response).unwrap(), 0);
assert_eq!(parse_errno("errno=95\n").unwrap(), 95);
assert!(parse_errno("no errno here\n").is_err());
assert!(parse_errno("errno=banana\n").is_err());
}
#[test]
fn connect_failure_names_the_socket() {
let config = config::parse(SAMPLE).unwrap();
let missing = Path::new("/nonexistent/torad/wg0.sock");
let err = format!("{:#}", configure(missing, &config).unwrap_err());
assert!(err.contains("wg0.sock"), "got: {err}");
}
}
db/schema.sql
+11 -2
View File
@@ -3,7 +3,8 @@ CREATE TYPE torrent_state AS ENUM (
'downloading',
'paused',
'finished',
'error'
'error',
'stale'
);
CREATE TABLE torrents (
@@ -11,7 +12,15 @@ CREATE TABLE torrents (
info_hash TEXT NOT NULL UNIQUE,
name TEXT,
source TEXT NOT NULL,
output_path TEXT NOT NULL,
-- Set by the poller once librqbit resolves metadata and creates the
-- output directory. NULL until then (magnet links take time to resolve).
output_path TEXT,
-- Per-torrent override of the download location. NULL = use the daemon
-- default (<download_dir>/<torrent_name>/ for multi-file torrents).
download_folder TEXT,
-- Per-torrent override of the post-completion move target. NULL = use
-- the daemon default (<download_dir>/completed/<torrent_name>/).
move_after_completion_folder TEXT,
total_bytes BIGINT,
downloaded_bytes BIGINT NOT NULL DEFAULT 0,
state torrent_state NOT NULL DEFAULT 'pending',
devenv.nix
+24
View File
@@ -55,6 +55,18 @@ in
buf
grpcurl
# VPN mode (torad --wireguard-config). Both are runtime dependencies torad
# execs, not build inputs.
# passt -> `pasta`, which creates the unprivileged user+network
# namespace torad re-executes itself into.
# wireguard-go -> the userspace WireGuard datapath.
# `wireguard-tools` is deliberately absent: torad configures the device by
# speaking the UAPI protocol over its socket, so no `wg` binary is needed.
# `iproute2` provides the `ip` used for the tunnel address and routes.
passt
wireguard-go
iproute2
# Protobuf generators
protoc-gen-prost-crate
protoc-gen-prost-serde
@@ -80,14 +92,26 @@ in
env.TORAD_PID_FILE = "${config.env.DEVENV_RUNTIME}/torad.pid";
processes.torad = {
# VPN mode is opt-in so the default `devenv up` is unchanged: set
# TORAD_WIREGUARD_CONFIG to a wg-quick config path and torad will re-exec
# itself into a namespace and route all torrent traffic through the tunnel.
# torad reads that variable itself (clap `env`), so nothing is needed here
# beyond leaving it unset by default.
exec = ''
${config.outputs.torad}/bin/torad --socket "$TORAD_SOCKET" --pid-file "$TORAD_PID_FILE"
'';
# Waits for postgres's own readiness probe (pg_isready + SELECT 1), not just process start.
after = [ "devenv:processes:postgres" ];
# torad speaks gRPC over a Unix socket and does not register reflection,
# so grpcurl would need proto files. Instead use the `tora health` CLI
# (already knows the proto, reads TORAD_SOCKET env), which exits 0 iff
# torad reports SERVING. torad's poll_db_health flips the status to
# NotServing when postgres ping fails, so this also gates on a live DB.
ready.exec = "${config.outputs.tora}/bin/tora health";
};
outputs = {
torad = cargoNix.workspaceMembers.torad.build;
tora = cargoNix.workspaceMembers.tora.build;
};
}
proto/torrent.proto
+109 -2
View File
@@ -15,12 +15,25 @@ service Torrents {
rpc Pause(PauseRequest) returns (PauseResponse);
// Resume a paused torrent.
rpc Resume(ResumeRequest) returns (ResumeResponse);
// Subscribe to a live stream of torrent notifications. Events are delivered
// from the subscribe point onward — no history is replayed. New subscribers
// only see events that fire after they connect.
rpc Notifications(NotificationsRequest) returns (stream Notification);
// List the files inside a torrent with their download progress. Returns
// NOT_FOUND if the torrent is unknown or its metadata has not resolved yet
// (e.g. a magnet link still fetching peers).
rpc Files(FilesRequest) returns (FilesResponse);
}
message AddRequest {
string magnet = 1;
// Optional output directory override. Empty means use torad's default.
string output_dir = 2;
// Folder to download into. Empty = daemon default
// (<download_dir>/<torrent_name>/ for multi-file torrents).
string download_folder = 2;
// Folder to move the torrent into after completion. Empty = daemon default
// (<download_dir>/completed/<torrent_name>/).
string move_after_completion_folder = 3;
}
message AddResponse {
@@ -56,6 +69,94 @@ message ResumeRequest {
message ResumeResponse {}
message NotificationsRequest {
// If non-empty, only events for these torrent ids (or id prefixes, as
// accepted by `Status`/`Remove`) are delivered.
repeated string ids = 1;
// If non-empty, only these event kinds are delivered. An empty list means
// all kinds.
repeated NotificationKind kinds = 2;
}
message FilesRequest {
string id = 1;
}
enum FileState {
FILE_STATE_UNSPECIFIED = 0;
// File is partially downloaded.
IN_PROGRESS = 1;
// File's full byte range has been verified.
READY = 2;
}
message TorrentFile {
// Path relative to the torrent root, as declared in the .torrent metadata.
// May contain forward slashes for multi-file torrents.
string name = 1;
uint64 length = 2;
uint64 downloaded_bytes = 3;
FileState state = 4;
}
message FilesResponse {
repeated TorrentFile files = 1;
}
message Notification {
// Wall-clock time the daemon observed the change, in milliseconds since the
// Unix epoch.
uint64 observed_at_unix_millis = 1;
// What changed. Use this to dispatch; use the `oneof detail` for
// kind-specific payload.
NotificationKind kind = 2;
// Full snapshot of the torrent *after* the change. Always populated so
// consumers don't need to fan out to `Status` for context.
TorrentStatus torrent = 3;
oneof detail {
StateChanged state_changed = 10;
ProgressMark progress = 11;
AddedInfo added = 12;
RemovedInfo removed = 13;
}
}
enum NotificationKind {
NOTIFICATION_KIND_UNSPECIFIED = 0;
// Emitted once when a torrent is added via `Add`.
TORRENT_ADDED = 1;
// Emitted once when a torrent is removed via `Remove`.
TORRENT_REMOVED = 2;
// Emitted on any `State` transition, including PAUSED <-> DOWNLOADING,
// DOWNLOADING <-> STALE, * -> FINISHED, * -> ERROR.
STATE_CHANGED = 3;
// Emitted when the integer download percentage changes (rate-limited).
// No event is sent if the percentage did not advance since the last tick.
PROGRESS = 4;
}
message StateChanged {
State previous = 1;
State current = 2;
}
message ProgressMark {
// Current integer percentage 0..=100.
uint32 percent = 1;
// Bytes downloaded since the previous PROGRESS event for this torrent.
uint64 bytes_delta = 2;
}
message AddedInfo {
// Source string as provided to `Add` (magnet, URL, or file path).
string source = 1;
}
message RemovedInfo {
// True if `Remove` was called with `delete_files = true`.
bool files_deleted = 1;
}
enum State {
STATE_UNSPECIFIED = 0;
PENDING = 1;
@@ -63,6 +164,12 @@ enum State {
PAUSED = 3;
FINISHED = 4;
ERROR = 5;
// Live per librqbit, but the daemon has detected a stall: no byte progress
// for several ticks AND zero connected peers. The torrent is still
// considered live by the underlying engine; this state is an application-
// level overlay. Recovers to DOWNLOADING automatically when bytes advance
// or a peer connects.
STALE = 6;
}
message TorrentStatus {
rustfmt.toml
+5
View File
@@ -0,0 +1,5 @@
# Declares the edition for `rustfmt` when it parses buffers via stdin
# (e.g. emacs `rust-format-buffer` / DOOM `format-all`). rustfmt does not
# consult Cargo.toml for stdin input, so without this it defaults to 2015
# and rejects `async fn` with E0670. Must match edition in Cargo.toml.
edition = "2024"