From 892b45d1e522b4a2d8b48625469fc321bd6e154d Mon Sep 17 00:00:00 2001 From: Max Lv Date: Wed, 14 Oct 2026 15:22:58 +0800 Subject: [PATCH] Add the C++ TCP relay engine and run ss-tunnel on it One engine (src/relay/tcp) for the four TCP relays: connections live in a fixed pool owned through PoolPtr, handshake, cipher state and buffers are inline, timeouts are intrusive deadline lists on one timer and DNS answers come back through generation-checked handles. A full pool refuses new connections and pauses accepting until a slot frees. ss-tunnel is the first user; its hand-written relay callbacks are gone. tests/tunnel_interop.py covers every method against ss-server and shadowsocks-rust. --- scripts/no_heap_baseline.txt | 2 +- src/CMakeLists.txt | 21 +- src/crypto/aead.hpp | 2 + src/crypto/aead2022.cpp | 96 ++- src/crypto/aead2022.hpp | 15 +- src/jconf.h | 2 + src/relay/codec.cpp | 188 +++++ src/relay/codec.hpp | 118 +++ src/relay/socket.cpp | 347 +++++++++ src/relay/socket.hpp | 83 ++ src/relay/tcp.cpp | 1157 ++++++++++++++++++++++++++++ src/relay/tcp.hpp | 232 ++++++ src/tunnel.cpp | 773 ++----------------- src/tunnel.h | 95 --- tests/CMakeLists.txt | 9 + tests/test_crypto_aead2022_cpp.cpp | 6 +- tests/tunnel_interop.py | 102 +++ 17 files changed, 2391 insertions(+), 857 deletions(-) create mode 100644 src/relay/codec.cpp create mode 100644 src/relay/codec.hpp create mode 100644 src/relay/socket.cpp create mode 100644 src/relay/socket.hpp create mode 100644 src/relay/tcp.cpp create mode 100644 src/relay/tcp.hpp delete mode 100644 src/tunnel.h create mode 100644 tests/tunnel_interop.py diff --git a/scripts/no_heap_baseline.txt b/scripts/no_heap_baseline.txt index 28ef5186..7a6db978 100644 --- a/scripts/no_heap_baseline.txt +++ b/scripts/no_heap_baseline.txt @@ -19,6 +19,6 @@ src/server.cpp 17 src/ss_event.cpp 5 src/ssurl.cpp 3 src/stream.cpp 3 -src/tunnel.cpp 14 +src/tunnel.cpp 2 src/udprelay.cpp 7 src/utils.cpp 10 diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index c5a98565..69ba0dd3 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -48,6 +48,19 @@ else() set(SS_PLUGIN_SOURCE plugin_disabled.cpp) endif() +# The C++ relay engine (src/relay) and the codecs it runs on. +set(SS_RELAY_SOURCE + relay/codec.cpp + relay/socket.cpp + relay/tcp.cpp + relay/resolver.cpp + crypto/aead.cpp + crypto/aead2022.cpp + ) +if(SS_ENABLE_LEGACY) + list(APPEND SS_RELAY_SOURCE crypto/stream.cpp) +endif() + set(SS_SHARED_SOURCES ppbloom.cpp utils.cpp @@ -65,8 +78,8 @@ set(LIBSHADOWSOCKS_LIBEV_SOURCE cache.cpp local.cpp resolv.cpp - relay/resolver.cpp ssurl.cpp + ${SS_RELAY_SOURCE} ${SS_CRYPTO_SOURCE} ${SS_PLUGIN_SOURCE} ${SS_ACL_SOURCE} @@ -82,6 +95,7 @@ set(SS_TUNNEL_SOURCE udprelay.cpp cache.cpp tunnel.cpp + ${SS_RELAY_SOURCE} ${SS_CRYPTO_SOURCE} ${SS_PLUGIN_SOURCE} ) @@ -91,8 +105,8 @@ set(SS_SERVER_SOURCE udprelay.cpp cache.cpp resolv.cpp - relay/resolver.cpp server.cpp + ${SS_RELAY_SOURCE} ${SS_CRYPTO_SOURCE} ${SS_PLUGIN_SOURCE} ${SS_ACL_SOURCE} @@ -108,6 +122,7 @@ set(SS_REDIR_SOURCE udprelay.cpp cache.cpp redir.cpp + ${SS_RELAY_SOURCE} ${SS_CRYPTO_SOURCE} ${SS_PLUGIN_SOURCE} ) @@ -126,7 +141,7 @@ if(ENABLE_NFTABLES) endif() # Modules rewritten in C++ compile with the stricter warning set. -ss_strict_warnings(cache.cpp resolv.cpp relay/resolver.cpp) +ss_strict_warnings(cache.cpp resolv.cpp ${SS_RELAY_SOURCE}) function(ss_program name module sources) add_executable(${name} ${sources}) diff --git a/src/crypto/aead.hpp b/src/crypto/aead.hpp index 64905cdc..825fc7f5 100644 --- a/src/crypto/aead.hpp +++ b/src/crypto/aead.hpp @@ -184,6 +184,8 @@ public: // Bytes received but not yet decrypted. std::size_t pending() const noexcept { return buffer_.size(); } + // Input bytes the next decrypt() takes in full. + std::size_t accept_limit() const noexcept { return buffer_.capacity() - buffer_.size(); } bool salt_received() const noexcept { return engine_ready_; } private: diff --git a/src/crypto/aead2022.cpp b/src/crypto/aead2022.cpp index 32bb5a24..434eaf9f 100644 --- a/src/crypto/aead2022.cpp +++ b/src/crypto/aead2022.cpp @@ -368,6 +368,8 @@ Status Aead2022TcpDecryptor::init(const Psk &psk, Role role, std::span Aead2022TcpDecryptor::decrypt(std::span Aead2022TcpDecryptor::decrypt(std::span(reinterpret_cast(r.data()), r.size()); + return std::span(reinterpret_cast(r.data()), r.size()); }; auto broken = [this](Errc code) { failed_ = true; return fail(code); }; - auto open = [this](std::uint8_t *dst, std::span chunk) { - bool ok = static_cast(engine_.open(dst, chunk, nonce_.data())); + // Chunks are opened in place: the plaintext ends up at the front of the + // buffer, followed by the tag (and for a request header the padding), + // which is skipped once the plaintext has been handed out. + auto open = [this](std::span chunk) { + bool ok = static_cast(engine_.open(chunk.data(), chunk, nonce_.data())); if (ok) { sodium_increment(nonce_.data(), psk_.spec.aead.nonce_len); } @@ -415,6 +417,23 @@ Result Aead2022TcpDecryptor::decrypt(std::span Aead2022TcpDecryptor::decrypt(std::span Aead2022TcpDecryptor::decrypt(std::span hdr(dst, want_); - std::size_t addr_len = socks_addr_len(hdr); - if (addr_len == 0 || want_ < addr_len + kLenSize) { - return broken(Errc::protocol); - } - std::size_t pad = load16_be(dst + addr_len); - std::size_t off = addr_len + kLenSize + pad; - if (pad > k2022MaxPad || off > want_) { - return broken(Errc::protocol); - } - std::size_t payload = want_ - off; - if (payload == 0 && pad == 0) { - return broken(Errc::protocol); // SIP022: neither payload nor padding - } - std::memmove(dst + addr_len, dst + off, payload); - step.produced += addr_len + payload; - } stage_ = Stage::length; + if (role_ == Role::client) { + plain_ = want_; // the first response payload + skip_ = kTag; + break; + } + std::uint8_t *hdr = data.data(); + std::size_t addr_len = socks_addr_len(data.first(want_)); + if (addr_len == 0 || want_ < addr_len + kLenSize) { + return broken(Errc::protocol); + } + std::size_t pad = load16_be(hdr + addr_len); + std::size_t off = addr_len + kLenSize + pad; + if (pad > k2022MaxPad || off > want_) { + return broken(Errc::protocol); + } + std::size_t payload = want_ - off; + if (payload == 0 && pad == 0) { + return broken(Errc::protocol); // SIP022: neither payload nor padding + } + std::memmove(hdr + addr_len, hdr + off, payload); + plain_ = addr_len + payload; + skip_ = want_ - plain_ + kTag; break; } case Stage::length: { if (data.size() < kLenSize + kTag) { return step; } - std::uint8_t len[kLenSize]; - if (!open(len, data.first(kLenSize + kTag))) { + if (!open(data.first(kLenSize + kTag))) { return broken(Errc::crypto); } - want_ = load16_be(len); + want_ = load16_be(data.data()); if (want_ == 0) { return broken(Errc::protocol); } @@ -508,14 +526,14 @@ Result Aead2022TcpDecryptor::decrypt(std::span decrypt(std::span in, std::span out) noexcept; + std::size_t accept_limit() const noexcept { return buffer_.capacity() - buffer_.size(); } bool header_received() const noexcept { return stage_ != Stage::fixed; } std::span salt() const noexcept { return { salt_.data(), psk_.size() }; } std::size_t pending() const noexcept { return buffer_.size(); } @@ -143,8 +146,10 @@ private: std::array nonce_{}; // The largest unit is a full 64 KiB chunk plus its tag. FixedBuffer buffer_; - std::size_t want_ = 0; // plaintext length of the next variable/payload chunk - Stage stage_ = Stage::fixed; + std::size_t want_ = 0; // plaintext length of the next variable/payload chunk + std::size_t plain_ = 0; // decrypted bytes at the front of buffer_ not yet handed out + std::size_t skip_ = 0; // bytes after them to drop (tag, padding) once they are + Stage stage_ = Stage::fixed; bool failed_ = false; }; diff --git a/src/jconf.h b/src/jconf.h index ce1c86e8..807b6bb2 100644 --- a/src/jconf.h +++ b/src/jconf.h @@ -33,6 +33,8 @@ extern "C" { #define MAX_CONNECT_TIMEOUT 10 #define MIN_TCP_IDLE_TIMEOUT (24 * 3600) #define MIN_UDP_TIMEOUT 10 +/* Connections a TCP relay serves at once; more are refused. */ +#define DEFAULT_MAX_CONNECTIONS 1024 #define DSCP_EF 0x2E #define DSCP_MIN 0x0 diff --git a/src/relay/codec.cpp b/src/relay/codec.cpp new file mode 100644 index 00000000..327c5b0a --- /dev/null +++ b/src/relay/codec.cpp @@ -0,0 +1,188 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later */ +#include "relay/codec.hpp" + +#include +#include + +#include "crypto.h" + +// crypto.h defines min/max macros for the C sources. +#undef min +#undef max + +namespace ss::relay { + +Status Method::init(const char *method, const char *password, const char *key, Role role) noexcept +{ + if (method == nullptr) { + return fail(Errc::config); + } + role_ = role; + if (auto spec = crypto::find_aead2022(method)) { + // The password field may carry the base64 PSK too. + SS_TRY_ASSIGN(psk, crypto::Psk::from_base64(*spec, key != nullptr ? key : password)); + key_ = psk; + return {}; + } + if (auto spec = crypto::find_aead(method)) { + if (key != nullptr) { + SS_TRY_ASSIGN(master, crypto::MasterKey::from_base64(*spec, key)); + key_ = master; + } else { + SS_TRY_ASSIGN(master, crypto::MasterKey::from_password(*spec, password)); + key_ = master; + } + return {}; + } +#if SS_ENABLE_LEGACY + if (auto spec = crypto::find_stream(method)) { + if (key != nullptr) { + crypto::StreamKey stream{ *spec, {} }; + if (crypto_parse_key(key, stream.bytes.data(), spec->key_len) != spec->key_len) { + return fail(Errc::config); + } + key_ = stream; + } else { + SS_TRY_ASSIGN(stream, crypto::StreamKey::from_password(*spec, password)); + key_ = stream; + } + return {}; + } +#endif + return fail(Errc::config); +} + +Status Encoder::init(Method &method, bool plain, const Decoder *peer, bool printable_salt) noexcept +{ + if (plain) { + impl_.emplace(); + return {}; + } + const crypto::ReplayFilter filter = crypto::ReplayFilter::ppbloom(); + if (auto *master = std::get_if(&method.key_)) { + return impl_.emplace().init(*master, filter, printable_salt); + } + if (auto *psk = std::get_if(&method.key_)) { + const crypto::Aead2022TcpDecryptor *request = peer != nullptr ? peer->as_2022() : nullptr; + return impl_.emplace().init(*psk, method.role_, request); + } +#if SS_ENABLE_LEGACY + if (auto *stream = std::get_if(&method.key_)) { + return impl_.emplace().init(*stream, filter); + } +#endif + return fail(Errc::config); +} + +// Explicit dispatch: std::visit can report bad_variant_access through the +// C++ runtime, which the programs do not link. +#if SS_ENABLE_LEGACY +#define SS_STREAM_CASE(E, expr) \ + if (auto *e = std::get_if(&impl_)) { \ + return expr; \ + } +#else +#define SS_STREAM_CASE(E, expr) +#endif + +std::size_t Encoder::sealed_size(std::size_t plaintext_len) const noexcept +{ + if (auto *e = std::get_if(&impl_)) { + return e->sealed_size(plaintext_len); + } + if (auto *e = std::get_if(&impl_)) { + return e->sealed_size(plaintext_len); + } + SS_STREAM_CASE(TcpEncryptor, e->sealed_size(plaintext_len)) + return plaintext_len; +} + +Result Encoder::seal(std::span plaintext, std::span out) noexcept +{ + if (auto *e = std::get_if(&impl_)) { + return e->seal(plaintext, out); + } + if (auto *e = std::get_if(&impl_)) { + return e->seal(plaintext, out); + } + SS_STREAM_CASE(TcpEncryptor, e->seal(plaintext, out)) + if (out.size() < plaintext.size()) { + return fail(Errc::invalid); + } + if (!plaintext.empty()) { + std::memcpy(out.data(), plaintext.data(), plaintext.size()); + } + return plaintext.size(); +} + +std::span Encoder::salt_2022() const noexcept +{ + if (auto *e = std::get_if(&impl_)) { + return e->salt(); + } + return {}; +} + +Status Decoder::init(Method &method, bool plain, const Encoder *peer) noexcept +{ + if (plain) { + impl_.emplace(); + return {}; + } + const crypto::ReplayFilter filter = crypto::ReplayFilter::ppbloom(); + if (auto *master = std::get_if(&method.key_)) { + return impl_.emplace().init(*master, filter); + } + if (auto *psk = std::get_if(&method.key_)) { + if (method.role_ == Role::client) { + if (peer == nullptr) { + return fail(Errc::invalid); + } + return impl_.emplace().init(*psk, Role::client, peer->salt_2022()); + } + return impl_.emplace().init(*psk, Role::server, {}, &method.salt_pool_); + } +#if SS_ENABLE_LEGACY + if (auto *stream = std::get_if(&method.key_)) { + return impl_.emplace().init(*stream, filter); + } +#endif + return fail(Errc::config); +} + +std::size_t Decoder::accept_limit() const noexcept +{ + if (auto *d = std::get_if(&impl_)) { + return d->accept_limit(); + } + if (auto *d = std::get_if(&impl_)) { + return d->accept_limit(); + } + return SIZE_MAX; // plain and stream decode as much as there is room for +} + +Result Decoder::decrypt(std::span in, std::span out) noexcept +{ + auto convert = [](auto step) -> Result { + if (!step) { + return fail(step.error()); + } + return Step{ step->consumed, step->produced }; + }; + if (auto *d = std::get_if(&impl_)) { + return convert(d->decrypt(in, out)); + } + if (auto *d = std::get_if(&impl_)) { + return convert(d->decrypt(in, out)); + } + SS_STREAM_CASE(TcpDecryptor, convert(e->decrypt(in, out))) + std::size_t n = std::min(in.size(), out.size()); + if (n != 0) { + std::memcpy(out.data(), in.data(), n); + } + return Step{ n, n }; +} + +#undef SS_STREAM_CASE + +} // namespace ss::relay diff --git a/src/relay/codec.hpp b/src/relay/codec.hpp new file mode 100644 index 00000000..ffd3ced5 --- /dev/null +++ b/src/relay/codec.hpp @@ -0,0 +1,118 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later */ +#ifndef SS_RELAY_CODEC_HPP +#define SS_RELAY_CODEC_HPP + +// The cipher a relay is configured with, and the per-connection encoder and +// decoder built from it. Every connection holds both by value; which kind is +// live is chosen per connection, because a connection that bypasses the +// proxy (ACL) is plain. + +#include +#include +#include +#include + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include "core/result.hpp" +#include "crypto/aead.hpp" +#include "crypto/aead2022.hpp" +#if SS_ENABLE_LEGACY +#include "crypto/stream.hpp" +#endif + +namespace ss::relay { + +using crypto::Role; + +class Method { +public: + // The same inputs as crypto_init(): a method name and either a password + // or a base64 key (AEAD-2022 needs the key). Replay protection uses the + // process-wide ppbloom filter, which crypto_init() sets up. + Status init(const char *method, const char *password, const char *key, Role role) noexcept; + + Role role() const noexcept { return role_; } + bool is_2022() const noexcept { return std::holds_alternative(key_); } + // Room a first seal() needs on top of the plaintext. + static constexpr std::size_t kMaxOverhead = 2048; + + // AEAD-2022 servers remember every request salt for a minute. + Status reserve_salt_pool(std::uint32_t capacity, std::uint64_t seed) noexcept + { + return salt_pool_.reserve(capacity, seed); + } + +private: + friend class Encoder; + friend class Decoder; + +#if SS_ENABLE_LEGACY + using Key = std::variant; +#else + using Key = std::variant; +#endif + Key key_{}; + Role role_ = Role::client; + crypto::SaltPool salt_pool_; +}; + +class Decoder; + +class Encoder { +public: + // `plain` skips encryption (a direct connection). An AEAD-2022 server + // encoder answers the request `peer` decoded. + Status init(Method &method, bool plain, const Decoder *peer, bool printable_salt = false) noexcept; + + std::size_t sealed_size(std::size_t plaintext_len) const noexcept; + Result seal(std::span plaintext, std::span out) noexcept; + std::span salt_2022() const noexcept; + +private: + struct Plain {}; +#if SS_ENABLE_LEGACY + std::variant impl_; +#else + std::variant impl_; +#endif +}; + +class Decoder { +public: + struct Step { + std::size_t consumed; + std::size_t produced; + }; + + // An AEAD-2022 client decoder checks the response against `peer`'s salt. + Status init(Method &method, bool plain, const Encoder *peer) noexcept; + + // Input bytes the next decrypt() consumes in full. + std::size_t accept_limit() const noexcept; + // Decrypt into `out`; a decoder may keep plaintext back when `out` is + // small, so call again with an empty input to drain it. `out` must hold + // at least kMinOutput bytes. + Result decrypt(std::span in, std::span out) noexcept; + bool is_2022() const noexcept { return std::holds_alternative(impl_); } + const crypto::Aead2022TcpDecryptor *as_2022() const noexcept + { + return std::get_if(&impl_); + } + + static constexpr std::size_t kMinOutput = crypto::kMaxChunk; + +private: + struct Plain {}; +#if SS_ENABLE_LEGACY + std::variant impl_; +#else + std::variant impl_; +#endif +}; + +} // namespace ss::relay + +#endif diff --git a/src/relay/socket.cpp b/src/relay/socket.cpp new file mode 100644 index 00000000..1ab45a97 --- /dev/null +++ b/src/relay/socket.cpp @@ -0,0 +1,347 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later */ +#include "relay/socket.hpp" + +#include + +#ifndef _WIN32 +#include +#endif + +#include "utils.h" + +#ifndef SOL_TCP +#define SOL_TCP IPPROTO_TCP +#endif + +namespace ss::relay { + +namespace { + +bool would_block(int err) noexcept +{ + return err == EAGAIN || err == EWOULDBLOCK; +} + +} // namespace + +Status set_nonblocking(ss_socket_t fd) noexcept +{ +#ifdef _WIN32 + if (setnonblocking(fd) != 0) { + return fail(Errc::io, ss_socket_error()); + } +#else + int flags = fcntl(fd, F_GETFL, 0); + if (flags == -1) { + flags = 0; + } + if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) { + return fail(Errc::io, errno); + } +#endif + return {}; +} + +void set_nodelay(ss_socket_t fd, bool on) noexcept +{ + int opt = on ? 1 : 0; + (void)ss_setsockopt(fd, SOL_TCP, TCP_NODELAY, &opt, sizeof(opt)); +} + +void set_nosigpipe(ss_socket_t fd) noexcept +{ +#ifdef SO_NOSIGPIPE + int opt = 1; + (void)ss_setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &opt, sizeof(opt)); +#else + (void)fd; +#endif +} + +socklen_t sockaddr_size(const sockaddr *addr) noexcept +{ + if (addr->sa_family == AF_INET) { + return sizeof(sockaddr_in); + } + if (addr->sa_family == AF_INET6) { + return sizeof(sockaddr_in6); + } + return 0; +} + +Result open_upstream(int family, SocketTuning &t, bool protect) noexcept +{ + int protocol = t.mptcp < 0 ? IPPROTO_MPTCP : IPPROTO_TCP; + ss_socket_t fd = ss_socket(family, SOCK_STREAM, protocol); +#ifdef _WIN32 + if (fd == INVALID_SOCKET) { +#else + if (fd < 0) { +#endif + int err = ss_socket_error(); + ERROR("socket"); + return fail(Errc::io, err); + } + if (protect && t.protect != nullptr && t.protect(static_cast(fd)) == -1) { + ERROR("protect_socket"); + ss_socket_close(fd); + return fail(Errc::io); + } + int opt = 1; + set_nodelay(fd, true); + set_nosigpipe(fd); + if (t.keepalive || t.keepalive_tuned) { + (void)ss_setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &opt, sizeof(opt)); + } +#if defined(TCP_KEEPIDLE) && defined(TCP_KEEPINTVL) && defined(TCP_KEEPCNT) + if (t.keepalive_tuned) { + int idle = 40, interval = 20, count = 5; + (void)ss_setsockopt(fd, SOL_TCP, TCP_KEEPIDLE, &idle, sizeof(idle)); + (void)ss_setsockopt(fd, SOL_TCP, TCP_KEEPINTVL, &interval, sizeof(interval)); + (void)ss_setsockopt(fd, SOL_TCP, TCP_KEEPCNT, &count, sizeof(count)); + } +#endif + if (t.tos >= 0) { + if (ss_setsockopt(fd, IPPROTO_IP, IP_TOS, &t.tos, sizeof(t.tos)) < 0 && ss_socket_error() != ENOPROTOOPT) { + LOGE("setting ipv4 dscp failed: %d", ss_socket_error()); + } +#ifdef IPV6_TCLASS + if (ss_setsockopt(fd, IPPROTO_IPV6, IPV6_TCLASS, &t.tos, sizeof(t.tos)) < 0 + && ss_socket_error() != ENOPROTOOPT) { + LOGE("setting ipv6 dscp failed: %d", ss_socket_error()); + } +#endif + } + if (t.mptcp > 1) { + if (ss_setsockopt(fd, SOL_TCP, t.mptcp, &opt, sizeof(opt)) == -1) { + ERROR("failed to enable out-of-tree multipath TCP"); + } + } else if (t.mptcp == 1) { + int i = 0; + while ((t.mptcp = mptcp_enabled_values[i]) > 0) { + if (ss_setsockopt(fd, SOL_TCP, t.mptcp, &opt, sizeof(opt)) != -1) { + break; + } + ++i; + } + if (t.mptcp == 0) { + ERROR("failed to enable out-of-tree multipath TCP"); + } + } + if (t.sndbuf > 0) { + (void)ss_setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &t.sndbuf, sizeof(int)); + } + if (t.rcvbuf > 0) { + (void)ss_setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &t.rcvbuf, sizeof(int)); + } + if (!set_nonblocking(fd)) { + ERROR("setnonblocking"); + } + const sockaddr_storage *source = family == AF_INET ? t.bind_v4 : t.bind_v6; + if (source != nullptr && source->ss_family == family + && bind_to_addr(const_cast(source), static_cast(fd)) == -1) { + ERROR("bind_to_addr"); + ss_socket_close(fd); + return fail(Errc::io); + } +#ifdef SET_INTERFACE + if (t.iface != nullptr && setinterface(static_cast(fd), t.iface) == -1) { + ERROR("setinterface"); + } +#endif + return fd; +} + +Result start_connect(ss_socket_t fd, const sockaddr *addr, std::span data, + bool fast_open) noexcept +{ + socklen_t len = sockaddr_size(addr); + if (!fast_open || data.empty()) { + if (connect(fd, addr, len) == -1 && ss_socket_error() != CONNECT_IN_PROGRESS) { + return fail(Errc::io, ss_socket_error()); + } + return std::size_t{ 0 }; + } + ssize_t s = -1; +#if defined(MSG_FASTOPEN) && !defined(TCP_FASTOPEN_CONNECT) + s = sendto(fd, reinterpret_cast(data.data()), data.size(), MSG_FASTOPEN, addr, len); +#elif defined(CONNECT_DATA_IDEMPOTENT) + sockaddr_storage copy{}; + std::memcpy(©, addr, len); + if (copy.ss_family == AF_INET) { + reinterpret_cast(©)->sin_len = sizeof(sockaddr_in); + } + sa_endpoints_t endpoints{}; + endpoints.sae_dstaddr = reinterpret_cast(©); + endpoints.sae_dstaddrlen = len; + s = connectx(fd, &endpoints, SAE_ASSOCID_ANY, CONNECT_RESUME_ON_READ_WRITE | CONNECT_DATA_IDEMPOTENT, nullptr, + 0, nullptr, nullptr); + if (s == 0) { + s = send(fd, data.data(), data.size(), 0); + } +#elif defined(TCP_FASTOPEN_CONNECT) + int optval = 1; + if (ss_setsockopt(fd, IPPROTO_TCP, TCP_FASTOPEN_CONNECT, &optval, sizeof(optval)) < 0) { + return fail(Errc::invalid, ss_socket_error()); + } + s = connect(fd, addr, len); + if (s == 0) { + s = send(fd, data.data(), data.size(), 0); + } +#else + return fail(Errc::invalid, ENOPROTOOPT); // not on this platform +#endif + if (s >= 0) { + return static_cast(s); + } + int err = ss_socket_error(); + if (err == CONNECT_IN_PROGRESS || would_block(err)) { + return std::size_t{ 0 }; + } + if (err == EOPNOTSUPP || err == EPROTONOSUPPORT || err == ENOPROTOOPT) { + return fail(Errc::invalid, err); + } + return fail(Errc::io, err); +} + +bool is_connected(ss_socket_t fd) noexcept +{ + sockaddr_storage addr{}; + socklen_t len = sizeof(addr); + return getpeername(fd, reinterpret_cast(&addr), &len) == 0; +} + +Result recv_some(ss_socket_t fd, std::span buf) noexcept +{ + ssize_t r = recv(fd, reinterpret_cast(buf.data()), buf.size(), 0); + if (r > 0) { + return static_cast(r); + } + if (r == 0) { + return fail(Errc::closed); + } + int err = ss_socket_error(); + if (would_block(err) || err == EINTR) { + return fail(Errc::again); + } + return fail(Errc::io, err); +} + +Result send_some(ss_socket_t fd, std::span buf) noexcept +{ + ssize_t s = send(fd, reinterpret_cast(buf.data()), buf.size(), 0); + if (s >= 0) { + return static_cast(s); + } + int err = ss_socket_error(); + if (would_block(err) || err == EINTR) { + return fail(Errc::again); + } + return fail(Errc::io, err); +} + +std::size_t encode_socks_addr(const sockaddr *addr, std::uint8_t *out) noexcept +{ + if (addr->sa_family == AF_INET) { + const auto *in = reinterpret_cast(addr); + out[0] = 1; + std::memcpy(out + 1, &in->sin_addr, 4); + std::memcpy(out + 5, &in->sin_port, 2); + return 7; + } + if (addr->sa_family == AF_INET6) { + const auto *in = reinterpret_cast(addr); + out[0] = 4; + std::memcpy(out + 1, &in->sin6_addr, 16); + std::memcpy(out + 17, &in->sin6_port, 2); + return 19; + } + return 0; +} + +std::size_t encode_socks_host(const char *host, std::uint16_t port, std::uint8_t *out) noexcept +{ + std::uint16_t be = htons(port); + in_addr v4{}; + in6_addr v6{}; + std::size_t at = 0; + if (inet_pton(AF_INET, host, &v4) == 1) { + out[at++] = 1; + std::memcpy(out + at, &v4, 4); + at += 4; + } else if (inet_pton(AF_INET6, host, &v6) == 1) { + out[at++] = 4; + std::memcpy(out + at, &v6, 16); + at += 16; + } else { + std::size_t n = std::strlen(host); + if (n == 0 || n > 255) { + return 0; + } + out[at++] = 3; + out[at++] = static_cast(n); + std::memcpy(out + at, host, n); + at += n; + } + std::memcpy(out + at, &be, 2); + return at + 2; +} + +ParseResult parse_socks_addr(std::span in, SocksTarget &out) noexcept +{ + if (in.empty()) { + return ParseResult::need_more; + } + out.type = in[0] & 0x0F; + out.ip = {}; + std::size_t at = 1; + switch (out.type) { + case 1: { + if (in.size() < 1 + 4 + 2) { + return ParseResult::need_more; + } + auto *sa = reinterpret_cast(&out.ip); + sa->sin_family = AF_INET; + std::memcpy(&sa->sin_addr, in.data() + 1, 4); + std::memcpy(&sa->sin_port, in.data() + 5, 2); + ss_inet_ntop(AF_INET, &sa->sin_addr, out.host, sizeof(out.host)); + at += 4; + break; + } + case 4: { + if (in.size() < 1 + 16 + 2) { + return ParseResult::need_more; + } + auto *sa = reinterpret_cast(&out.ip); + sa->sin6_family = AF_INET6; + std::memcpy(&sa->sin6_addr, in.data() + 1, 16); + std::memcpy(&sa->sin6_port, in.data() + 17, 2); + ss_inet_ntop(AF_INET6, &sa->sin6_addr, out.host, sizeof(out.host)); + at += 16; + break; + } + case 3: { + if (in.size() < 2) { + return ParseResult::need_more; + } + std::size_t n = in[1]; + if (n == 0) { + return ParseResult::invalid; + } + if (in.size() < 2 + n + 2) { + return ParseResult::need_more; + } + std::memcpy(out.host, in.data() + 2, n); + out.host[n] = '\0'; + at += 1 + n; + break; + } + default: + return ParseResult::invalid; + } + out.port = static_cast((in[at] << 8) | in[at + 1]); + out.length = at + 2; + return ParseResult::complete; +} + +} // namespace ss::relay diff --git a/src/relay/socket.hpp b/src/relay/socket.hpp new file mode 100644 index 00000000..e76c6376 --- /dev/null +++ b/src/relay/socket.hpp @@ -0,0 +1,83 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later */ +#ifndef SS_RELAY_SOCKET_HPP +#define SS_RELAY_SOCKET_HPP + +// Socket plumbing shared by the relays: option tuning, non-blocking I/O +// with Result errors, TCP Fast Open and SOCKS5 address encoding. + +#include +#include +#include +#include + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include "core/result.hpp" +#include "netutils.h" +#include "platform.h" + +namespace ss::relay { + +// Options applied to every outgoing (upstream) socket. +struct SocketTuning { + int sndbuf = 0; + int rcvbuf = 0; + // As the C relays: < 0 opens IPPROTO_MPTCP sockets, 1 probes the + // out-of-tree option (and remembers the one that worked), > 1 is that + // option. + int mptcp = 0; + const char *iface = nullptr; + bool keepalive = false; + bool keepalive_tuned = false; // ss-redir: idle 40 s, interval 20 s, 5 probes + int tos = -1; + // ss-server --bind-address: source addresses for each family. + const sockaddr_storage *bind_v4 = nullptr; + const sockaddr_storage *bind_v6 = nullptr; + int (*protect)(int fd) = nullptr; // Android VpnService +}; + +Status set_nonblocking(ss_socket_t fd) noexcept; +void set_nodelay(ss_socket_t fd, bool on) noexcept; +void set_nosigpipe(ss_socket_t fd) noexcept; +socklen_t sockaddr_size(const sockaddr *addr) noexcept; + +// A non-blocking stream socket for `family`, tuned for upstream use. +Result open_upstream(int family, SocketTuning &tuning, bool protect) noexcept; + +// Start connecting. With `fast_open` the first bytes of `data` travel in the +// SYN; returns how many were accepted (0 when nothing was sent yet). Fails +// with Errc::invalid when the platform turned fast open down; the caller then +// falls back to a plain connect. +Result start_connect(ss_socket_t fd, const sockaddr *addr, std::span data, + bool fast_open) noexcept; +// After the socket became writable: did the connect succeed? +bool is_connected(ss_socket_t fd) noexcept; + +// recv/send with EAGAIN as Errc::again and an orderly EOF as Errc::closed. +Result recv_some(ss_socket_t fd, std::span buf) noexcept; +Result send_some(ss_socket_t fd, std::span buf) noexcept; + +// SOCKS5 address (type, address, port) for an IP endpoint; returns its length. +inline constexpr std::size_t kMaxSocksAddr = 1 + 1 + 255 + 2; +std::size_t encode_socks_addr(const sockaddr *addr, std::uint8_t *out) noexcept; +// For a host name or a literal IP plus a port in host byte order. Returns 0 +// when the name does not fit. +std::size_t encode_socks_host(const char *host, std::uint16_t port, std::uint8_t *out) noexcept; + +struct SocksTarget { + std::size_t length = 0; // bytes of the encoded address + std::uint8_t type = 0; // 1, 3 or 4 + char host[256] = {}; // dotted/colon form for IPs, the name for domains + std::uint16_t port = 0; // host byte order + sockaddr_storage ip{}; // filled for IP addresses +}; +enum class ParseResult : std::uint8_t { complete, need_more, invalid }; +// Parse a SOCKS5 address at the head of `in` (the shadowsocks request +// header). The address type's upper bits are ignored, as ss-server does. +ParseResult parse_socks_addr(std::span in, SocksTarget &out) noexcept; + +} // namespace ss::relay + +#endif diff --git a/src/relay/tcp.cpp b/src/relay/tcp.cpp new file mode 100644 index 00000000..bde52131 --- /dev/null +++ b/src/relay/tcp.cpp @@ -0,0 +1,1157 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later */ +#include "relay/tcp.hpp" + +#include +#include +#include +#include + +#include + +#ifdef __linux__ +#include +#endif + +#include "acl.h" // BLACK_LIST, WHITE_LIST +#include "jconf.h" +#include "utils.h" + +extern "C" int verbose; + +#ifndef SOL_IPV6 +#define SOL_IPV6 IPPROTO_IPV6 +#endif +#ifndef IP6T_SO_ORIGINAL_DST +#define IP6T_SO_ORIGINAL_DST 80 +#endif + +namespace ss::relay { + +using event::Flow; + +namespace { + +constexpr std::uint64_t kDelayMs = 50; // wait for a first payload before connecting +constexpr int kAcceptBurst = 64; // connections accepted per wakeup +constexpr int kPumpRounds = 16; // receives per direction per wakeup +constexpr std::uint8_t kSocksVersion = 5; + +std::uint64_t pack(Handle h) noexcept +{ + return (static_cast(h.index) << 32) | h.gen; +} + +Handle unpack(std::uint64_t v) noexcept +{ + return Handle{ static_cast(v >> 32), static_cast(v) }; +} + +std::span bytes_of(std::span b) noexcept +{ + return { reinterpret_cast(b.data()), b.size() }; +} + +bool loopback_v4(const sockaddr *addr) noexcept +{ + return addr->sa_family == AF_INET + && reinterpret_cast(addr)->sin_addr.s_addr == htonl(INADDR_LOOPBACK); +} + +void send_reply(ss_socket_t fd, const std::uint8_t *data, std::size_t len) noexcept +{ + // Handshake replies are tiny; like the C relays, a short send is fatal. + (void)send_some(fd, { data, len }); +} + +#ifdef __linux__ +bool original_destination(ss_socket_t fd, bool tproxy, sockaddr_storage &out) noexcept +{ + socklen_t len = sizeof(out); + if (tproxy) { + return getsockname(fd, reinterpret_cast(&out), &len) == 0; + } + if (getsockopt(fd, SOL_IPV6, IP6T_SO_ORIGINAL_DST, &out, &len) == 0) { + return true; + } + len = sizeof(out); + return getsockopt(fd, SOL_IP, SO_ORIGINAL_DST, &out, &len) == 0; +} +#else +bool original_destination(ss_socket_t, bool, sockaddr_storage &) noexcept +{ + return false; +} +#endif + +void peer_name(ss_socket_t fd, char *out, std::size_t size) noexcept +{ + out[0] = '\0'; + sockaddr_storage addr{}; + socklen_t len = sizeof(addr); + if (getpeername(fd, reinterpret_cast(&addr), &len) != 0) { + return; + } + if (addr.ss_family == AF_INET) { + ss_inet_ntop(AF_INET, &reinterpret_cast(&addr)->sin_addr, out, static_cast(size)); + } else if (addr.ss_family == AF_INET6) { + ss_inet_ntop(AF_INET6, &reinterpret_cast(&addr)->sin6_addr, out, + static_cast(size)); + } +} + +} // namespace + +// ------------------------------------------------------------------ relay + +TcpRelay::~TcpRelay() +{ + for (auto &owner : owners_) { + std::destroy_at(&owner); + } +} + +Status TcpRelay::init(event::Loop &loop, Method &method, const TcpConfig &config, Resolver *resolver) noexcept +{ + if (loop_ != nullptr || config.max_connections == 0) { + return fail(Errc::invalid); + } + if (config.kind != Kind::server && config.remotes.empty()) { + return fail(Errc::config); + } + loop_ = &loop; + method_ = &method; + resolver_ = resolver; + config_ = config; + const AclHooks &acl = config_.acl_hooks; + if (acl.match_host == nullptr || acl.mode == nullptr || acl.outbound_blocked == nullptr) { + config_.acl = false; + } + SS_TRY(pool_.reserve(config.max_connections)); + using Owner = PoolPtr; + SS_TRY_ASSIGN(region, Region::reserve(config.max_connections * sizeof(Owner), alignof(Owner))); + owners_region_ = std::move(region); + auto *first = reinterpret_cast(owners_region_.data()); + for (std::uint32_t i = 0; i < config.max_connections; ++i) { + std::construct_at(first + i); + } + owners_ = { first, config.max_connections }; + + const std::uint64_t timeout = static_cast(std::max(config.timeout_secs, 1)); + delay_list_.set_timeout(kDelayMs); + connect_list_.set_timeout(std::min(MAX_CONNECT_TIMEOUT, timeout) * 1000); + idle_list_.set_timeout(std::max(MIN_TCP_IDLE_TIMEOUT, timeout) * 1000); + SS_TRY(timer_.open(loop, *this)); + own_closed_ = false; + return {}; +} + +Status TcpRelay::listen(ss_socket_t fd) noexcept +{ + if (listener_count_ == kMaxListeners || loop_ == nullptr) { + return fail(Errc::exhausted); + } + SS_TRY(set_nonblocking(fd)); + Listener &l = listeners_[listener_count_]; + l.relay = this; + SS_TRY(l.poll.open(*loop_, l, fd)); + ++listener_count_; + return l.poll.want(event::readable); +} + +Flow TcpRelay::Listener::on_ready(int) noexcept +{ + relay->accept_from(*this); + return Flow::proceed; +} + +void TcpRelay::accept_from(Listener &listener) noexcept +{ + for (int i = 0; i < kAcceptBurst && !shutting_down_; ++i) { + ss_socket_t fd = ss_accept(listener.poll.fd(), nullptr, nullptr); +#ifdef _WIN32 + if (fd == INVALID_SOCKET) { +#else + if (fd < 0) { +#endif + int err = ss_socket_error(); + if (err != EAGAIN && err != EWOULDBLOCK && err != EINTR) { + ERROR("accept"); + } + return; + } + if (pool_.full()) { + ss_socket_close(fd); + ++refused_; + std::uint64_t now = loop_->now(); + if (now - last_refusal_log_ >= 1000 || last_refusal_log_ == 0) { + LOGE("connection limit of %u reached, %" PRIu64 " connection(s) refused so far", + pool_.capacity(), refused_); + last_refusal_log_ = now; + } + pause_accept(); + return; + } + if (config_.kind == Kind::server && config_.acl) { + char peer[INET6_ADDRSTRLEN]; + peer_name(fd, peer, sizeof(peer)); + if (peer[0] != '\0' + && ((config_.acl_hooks.mode() == BLACK_LIST && config_.acl_hooks.match_host(peer) == 1) + || (config_.acl_hooks.mode() == WHITE_LIST && config_.acl_hooks.match_host(peer) >= 0))) { + LOGE("Access denied from %s", peer); + ss_socket_close(fd); + continue; + } + } + auto conn = pool_.acquire(*this); + if (!conn) { + ss_socket_close(fd); + return; + } + Connection &c = **conn; + std::uint32_t at = pool_.handle_of(&c).index; + owners_[at] = std::move(*conn); + (void)c.start(fd); + } +} + +void TcpRelay::pause_accept() noexcept +{ + if (accept_paused_) { + return; + } + accept_paused_ = true; + for (std::size_t i = 0; i < listener_count_; ++i) { + (void)listeners_[i].poll.want(0); + } +} + +void TcpRelay::resume_accept() noexcept +{ + accept_paused_ = false; + for (std::size_t i = 0; i < listener_count_; ++i) { + (void)listeners_[i].poll.want(event::readable); + } +} + +void TcpRelay::slot_freed() noexcept +{ + if (accept_paused_ && !shutting_down_ && !pool_.full()) { + resume_accept(); + } +} + +void TcpRelay::close(Connection &conn) noexcept +{ + clear_deadline(conn); + std::uint32_t at = pool_.handle_of(&conn).index; + owners_[at].reset(); // begins the asynchronous close +} + +const sockaddr_storage &TcpRelay::pick_remote() noexcept +{ + auto n = static_cast(config_.remotes.size()); + return config_.remotes[n > 1 ? randombytes_uniform(n) : 0]; +} + +void TcpRelay::set_deadline(Connection &conn, Connection::Timed which) noexcept +{ + if (conn.timed_ != which) { + clear_deadline(conn); + } + event::DeadlineList *list = which == Connection::Timed::delay ? &delay_list_ + : which == Connection::Timed::connect ? &connect_list_ + : &idle_list_; + std::uint64_t now = loop_->now(); + list->touch(conn, now); + conn.timed_ = which; + std::uint64_t due = conn.deadline(); + if (timer_due_ == 0 || due < timer_due_) { + timer_due_ = due; + timer_.start(due > now ? due - now : 1); + } +} + +void TcpRelay::clear_deadline(Connection &conn) noexcept +{ + if (conn.timed_ != Connection::Timed::none) { + event::DeadlineList::remove(conn); + conn.timed_ = Connection::Timed::none; + } +} + +void TcpRelay::on_timer() noexcept +{ + timer_due_ = 0; + std::uint64_t now = loop_->now(); + auto fire = [](Connection::Timed which) { + return [which](Connection &c) { + c.timed_ = Connection::Timed::none; + if (which == Connection::Timed::delay) { + c.on_deadline(); + } else { + if (verbose) { + LOGI("TCP connection timeout"); + } + (void)c.close(); + } + }; + }; + delay_list_.expire(now, fire(Connection::Timed::delay)); + connect_list_.expire(now, fire(Connection::Timed::connect)); + idle_list_.expire(now, fire(Connection::Timed::idle)); + schedule(); +} + +void TcpRelay::schedule() noexcept +{ + std::uint64_t due = 0; + for (auto *list : { &delay_list_, &connect_list_, &idle_list_ }) { + if (auto next = list->next_due()) { + due = due == 0 ? *next : std::min(due, *next); + } + } + if (due == 0) { + timer_.stop(); + timer_due_ = 0; + return; + } + std::uint64_t now = loop_->now(); + timer_due_ = due; + timer_.start(due > now ? due - now : 1); +} + +Status TcpRelay::resolve(Connection &conn, const char *host, std::uint16_t port_be) noexcept +{ + if (resolver_ == nullptr) { + return fail(Errc::invalid); + } + ResolveCallback cb; + cb.fn = &TcpRelay::resolved; + cb.ctx = this; + cb.token = pack(pool_.handle_of(&conn)); + return resolver_->resolve(host, port_be, cb); +} + +void TcpRelay::resolved(const ResolveCallback &cb, const sockaddr *address) noexcept +{ + auto *self = static_cast(cb.ctx); + if (Connection *conn = self->pool_.resolve(unpack(cb.token))) { + conn->on_resolved(address); + } +} + +void TcpRelay::shutdown() noexcept +{ + if (shutting_down_ || loop_ == nullptr) { + return; + } + shutting_down_ = true; + for (auto &owner : owners_) { + if (owner) { + close(*owner); + } + } + closer_.arm(CloseCompletion([](void *self, std::uint32_t) { static_cast(self)->own_closed_ = true; }, + this, 0)); + for (std::size_t i = 0; i < listener_count_; ++i) { + listeners_[i].poll.close(closer_); + } + timer_.close(closer_); + closer_.seal(); +} + +bool TcpRelay::closed() const noexcept +{ + return own_closed_ && pool_.live() == 0 && pool_.closing() == 0; +} + +// ------------------------------------------------------------- connection + +bool Connection::client_kind() const noexcept +{ + return relay_->config_.kind != Kind::server; +} + +void Connection::begin_close(CloseCompletion done) noexcept +{ + done_ = done; + closer_.arm(CloseCompletion( + [](void *self, std::uint32_t) { + auto *c = static_cast(self); + TcpRelay *relay = c->relay_; + CloseCompletion d = c->done_; + d(); // destroys *c + relay->slot_freed(); + }, + this, 0)); + client_.close(closer_); + upstream_.close(closer_); + closer_.seal(); +} + +Flow Connection::close() noexcept +{ + relay_->close(*this); + return Flow::closed; +} + +Flow Connection::start(ss_socket_t fd) noexcept +{ + (void)set_nonblocking(fd); + set_nodelay(fd, true); + set_nosigpipe(fd); + if (!client_.open(*relay_->loop_, *this, fd)) { + ss_socket_close(fd); + return close(); + } + const TcpConfig &cfg = relay_->config_; + switch (cfg.kind) { + case Kind::local: + if (verbose) { + char peer[INET6_ADDRSTRLEN]; + peer_name(fd, peer, sizeof(peer)); + LOGI("connection from %s", peer); + } + stage_ = Stage::socks_method; + update_interest(); + return Flow::proceed; + case Kind::redir: { + sockaddr_storage dest{}; + if (!original_destination(fd, cfg.tproxy, dest)) { + ERROR("getdestaddr"); + return close(); + } + target_header_len_ = encode_socks_addr(reinterpret_cast(&dest), target_header_.data()); + if (target_header_len_ == 0) { + return close(); + } + break; + } + case Kind::tunnel: + std::memcpy(target_header_.data(), cfg.tunnel_target.data(), cfg.tunnel_target_len); + target_header_len_ = cfg.tunnel_target_len; + break; + case Kind::server: { + if (!dec_.init(*relay_->method_, false, nullptr) || !enc_.init(*relay_->method_, false, &dec_)) { + return close(); + } + stage_ = Stage::header; + relay_->set_deadline(*this, Timed::idle); + update_interest(); + return Flow::proceed; + } + } + // ss-tunnel and ss-redir: the target is known; with fast open, wait a + // moment so the first payload can ride in the SYN. + upstream_addr_ = relay_->pick_remote(); + if (cfg.fast_open) { + stage_ = Stage::delay; + relay_->set_deadline(*this, Timed::delay); + update_interest(); + return Flow::proceed; + } + return start_upstream(reinterpret_cast(&upstream_addr_), false); +} + +Flow Connection::on_client(int events) noexcept +{ + switch (stage_) { + case Stage::socks_method: + case Stage::socks_request: + return socks_input(); + case Stage::delay: + return delay_input(); + case Stage::udp_assoc: + case Stage::stopped: { + // Read and drop until the client goes away (or the timeout). + for (int i = 0; i < kPumpRounds; ++i) { + auto r = recv_some(client_.fd(), relay_->scratch_); + if (!r) { + if (r.error().code == Errc::again) { + return Flow::proceed; + } + return close(); + } + if (relay_->config_.tx != nullptr) { + *relay_->config_.tx += *r; + } + } + return Flow::proceed; + } + case Stage::header: { + Pull p = pull(client_.fd(), true, to_upstream_, relay_->config_.tx); + switch (p) { + case Pull::data: + return server_header(); + case Pull::nothing: + case Pull::again: + return Flow::proceed; + case Pull::eof: + case Pull::error: + return close(); + } + return Flow::proceed; + } + case Stage::stream: + if ((events & event::writable) && pump_to_client() == Flow::closed) { + return Flow::closed; + } + if ((events & (event::readable | event::failed)) && pump_to_upstream() == Flow::closed) { + return Flow::closed; + } + update_interest(); + return Flow::proceed; + case Stage::acl_resolve: + case Stage::resolving: + case Stage::connecting: + return Flow::proceed; // not reading the client meanwhile + } + return Flow::proceed; +} + +Flow Connection::on_upstream(int events) noexcept +{ + if (stage_ == Stage::connecting) { + if (!is_connected(upstream_.fd())) { + if (verbose || client_kind()) { + ERROR("getpeername"); + } + return close(); + } + return connected(); + } + if (stage_ != Stage::stream) { + return Flow::proceed; + } + if ((events & event::writable) && pump_to_upstream() == Flow::closed) { + return Flow::closed; + } + if ((events & (event::readable | event::failed)) && pump_to_client() == Flow::closed) { + return Flow::closed; + } + update_interest(); + return Flow::proceed; +} + +void Connection::on_deadline() noexcept +{ + // Only the first-payload delay ends anywhere but in a close. + if (stage_ == Stage::delay) { + (void)start_upstream(reinterpret_cast(&upstream_addr_), direct_); + } +} + +void Connection::on_resolved(const sockaddr *address) noexcept +{ + if (stage_ == Stage::acl_resolve) { + dns_done_ = true; + dns_ok_ = address != nullptr; + if (address != nullptr) { + std::memcpy(&dns_answer_, address, sockaddr_size(address)); + } + (void)socks_route(); + return; + } + if (stage_ == Stage::resolving) { + if (address == nullptr) { + LOGE("unable to resolve %s", target_.host); + (void)close(); + return; + } + if (verbose) { + LOGI("successfully resolved %s", target_.host); + } + (void)server_connect(address); + } +} + +// ----------------------------------------------------------- ss-local + +Flow Connection::socks_input() noexcept +{ + auto room = bytes_of(handshake_.writable()); + if (room.empty()) { + return close(); // no SOCKS5 message is this long + } + auto r = recv_some(client_.fd(), room); + if (!r) { + return r.error().code == Errc::again ? Flow::proceed : close(); + } + (void)handshake_.commit(*r); + + if (stage_ == Stage::socks_method) { + auto in = bytes_of(handshake_.readable()); + if (in[0] != kSocksVersion) { + return close(); + } + if (in.size() < 2 || in.size() < 2U + in[1]) { + return Flow::proceed; + } + std::uint8_t reply[2] = { kSocksVersion, 0xFF }; + for (std::size_t i = 0; i < in[1]; ++i) { + if (in[2 + i] == 0x00) { + reply[1] = 0x00; // no authentication + break; + } + } + send_reply(client_.fd(), reply, sizeof(reply)); + if (reply[1] != 0x00) { + return close(); + } + (void)handshake_.consume(2U + in[1]); + stage_ = Stage::socks_request; + if (handshake_.empty()) { + return Flow::proceed; + } + } + return socks_request(); +} + +Flow Connection::socks_request() noexcept +{ + auto in = bytes_of(handshake_.readable()); + if (in.size() < 4) { + return Flow::proceed; + } + constexpr std::uint8_t kCmdConnect = 1, kCmdUdp = 3; + if (in[1] == kCmdUdp) { + if (verbose) { + LOGI("udp assc request accepted"); + } + sockaddr_storage local{}; + socklen_t len = sizeof(local); + std::uint8_t reply[3 + kMaxSocksAddr] = { kSocksVersion, 0, 0 }; + std::size_t n = 0; + if (getsockname(client_.fd(), reinterpret_cast(&local), &len) == 0) { + n = encode_socks_addr(reinterpret_cast(&local), reply + 3); + } + if (n == 0) { + reply[1] = 5; // connection refused + send_reply(client_.fd(), reply, 3); + return close(); + } + send_reply(client_.fd(), reply, 3 + n); + handshake_.clear(); + stage_ = Stage::udp_assoc; + update_interest(); + return Flow::proceed; + } + if (in[1] != kCmdConnect) { + LOGE("unsupported command: %d", in[1]); + std::uint8_t reply[4] = { kSocksVersion, 7, 0, 1 }; // command not supported + send_reply(client_.fd(), reply, sizeof(reply)); + return close(); + } + std::uint8_t atyp = in[3]; + if (atyp != 1 && atyp != 3 && atyp != 4) { + LOGE("unsupported addrtype: %d", atyp); + std::uint8_t reply[4] = { kSocksVersion, 8, 0, 1 }; // address type not supported + send_reply(client_.fd(), reply, sizeof(reply)); + return close(); + } + ParseResult pr = parse_socks_addr(in.subspan(3), target_); + if (pr == ParseResult::need_more) { + return Flow::proceed; + } + if (pr == ParseResult::invalid) { + return close(); + } + std::memcpy(target_header_.data(), in.data() + 3, target_.length); + target_header_len_ = target_.length; + (void)handshake_.consume(3 + target_.length); // what remains is the first payload + if (verbose) { + if (target_.type == 4) { + LOGI("connect to [%s]:%u", target_.host, target_.port); + } else { + LOGI("connect to %s:%u", target_.host, target_.port); + } + } + return socks_route(); +} + +Flow Connection::socks_route() noexcept +{ + const TcpConfig &cfg = relay_->config_; + // An ACL on host names may need the address before deciding. + if (cfg.acl && target_.type == 3 && !dns_done_ && !cfg.vpn && cfg.acl_hooks.match_host(target_.host) >= 0) { + stage_ = Stage::acl_resolve; + update_interest(); + if (relay_->resolve(*this, target_.host, htons(target_.port))) { + return Flow::proceed; + } + dns_done_ = true; // cannot ask: decide without the address + } + + // The reply goes out before the upstream is connected, as in the C + // implementation (clients send their first bytes sooner). + std::uint8_t reply[10] = { kSocksVersion, 0, 0, 1, 0, 0, 0, 0, 0, 0 }; + send_reply(client_.fd(), reply, sizeof(reply)); + + bool bypass = false; + sockaddr_storage direct{}; + bool have_direct = false; + if (cfg.acl && !(cfg.vpn && target_.port == 53)) { + int host_match = target_.type == 3 ? cfg.acl_hooks.match_host(target_.host) : 0; + if (host_match > 0) { + bypass = true; + } else if (host_match == 0) { + char ip[INET6_ADDRSTRLEN] = {}; + bool known = false; + if (target_.type == 3) { + if (dns_ok_ && !cfg.vpn) { + direct = dns_answer_; + have_direct = true; + const void *raw = direct.ss_family == AF_INET + ? static_cast(&reinterpret_cast(&direct)->sin_addr) + : static_cast(&reinterpret_cast(&direct)->sin6_addr); + known = ss_inet_ntop(direct.ss_family, raw, ip, sizeof(ip)) != nullptr; + } + } else { + std::strncpy(ip, target_.host, sizeof(ip) - 1); + direct = target_.ip; + have_direct = true; + known = true; + } + if (known) { + int ip_match = cfg.acl_hooks.match_host(ip); + if (cfg.acl_hooks.mode() == BLACK_LIST) { + bypass = ip_match > 0; + } else { + bypass = ip_match >= 0; + } + } + } + if (bypass && !have_direct) { + if (target_.type == 3 && dns_ok_ && !cfg.vpn) { + direct = dns_answer_; + have_direct = true; + } else { + bypass = false; // nowhere to go directly + } + } + if (bypass && verbose) { + LOGI("bypass %s:%u", target_.host, target_.port); + } + } + + if (bypass && have_direct) { + upstream_addr_ = direct; + direct_ = true; + } else { + upstream_addr_ = relay_->pick_remote(); + direct_ = false; + } + if (!handshake_.empty()) { + return start_upstream(reinterpret_cast(&upstream_addr_), direct_); + } + // No payload with the request: give the client a moment to send some, + // so it can travel with the address (and in the SYN with fast open). + stage_ = Stage::delay; + relay_->set_deadline(*this, Timed::delay); + update_interest(); + return Flow::proceed; +} + +Flow Connection::delay_input() noexcept +{ + auto room = bytes_of(handshake_.writable()); + if (!room.empty()) { + auto r = recv_some(client_.fd(), room); + if (!r) { + return r.error().code == Errc::again ? Flow::proceed : close(); + } + (void)handshake_.commit(*r); + } + relay_->clear_deadline(*this); + return start_upstream(reinterpret_cast(&upstream_addr_), direct_); +} + +Flow Connection::start_upstream(const sockaddr *target, bool direct) noexcept +{ + TcpConfig &cfg = relay_->config_; + direct_ = direct; + if (!enc_.init(*relay_->method_, direct, nullptr, cfg.printable_salt && !direct) + || !dec_.init(*relay_->method_, direct, &enc_)) { + LOGE("invalid password or cipher"); + return close(); + } + // The first write carries the target address and whatever the client + // has sent so far. + std::size_t n = 0; + if (!direct) { + std::memcpy(relay_->scratch_.data(), target_header_.data(), target_header_len_); + n = target_header_len_; + } + auto pending = bytes_of(handshake_.readable()); + if (!pending.empty()) { + std::memcpy(relay_->scratch_.data() + n, pending.data(), pending.size()); + n += pending.size(); + } + handshake_.clear(); + to_upstream_.clear(); + if (n != 0) { + auto w = enc_.seal({ relay_->scratch_.data(), n }, bytes_of(to_upstream_.writable())); + if (!w) { + LOGE("invalid password or cipher"); + return close(); + } + (void)to_upstream_.commit(*w); + } + + bool protect = cfg.vpn && !loopback_v4(target); + auto fd = open_upstream(target->sa_family, cfg.upstream, protect); + if (!fd) { + return close(); + } + if (!upstream_.open(*relay_->loop_, *this, *fd)) { + ss_socket_close(*fd); + return close(); + } + if (verbose) { + char host[INET6_ADDRSTRLEN] = {}; + const void *raw = target->sa_family == AF_INET + ? static_cast(&reinterpret_cast(target)->sin_addr) + : static_cast(&reinterpret_cast(target)->sin6_addr); + ss_inet_ntop(target->sa_family, raw, host, sizeof(host)); + LOGI("remote: %s", host); + } + bool fast = cfg.fast_open && !direct; + auto sent = start_connect(*fd, target, bytes_of(to_upstream_.readable()), fast); + if (!sent && sent.error().code == Errc::invalid) { + LOGE("fast open is not supported on this platform"); + cfg.fast_open = false; + sent = start_connect(*fd, target, {}, false); + } + if (!sent) { + ERROR("connect"); + return close(); + } + (void)to_upstream_.consume(*sent); + stage_ = Stage::connecting; + relay_->set_deadline(*this, Timed::connect); + update_interest(); + return Flow::proceed; +} + +// ---------------------------------------------------------- ss-server + +void Connection::server_reject(const char *reason) noexcept +{ + if (relay_->config_.report_failure != nullptr) { + relay_->config_.report_failure(client_.fd(), reason); + } + stage_ = Stage::stopped; + to_upstream_.clear(); + if (upstream_.is_open()) { + upstream_.close(closer_); + } + update_interest(); +} + +Flow Connection::server_header() noexcept +{ + const TcpConfig &cfg = relay_->config_; + auto in = bytes_of(to_upstream_.readable()); + ParseResult pr = parse_socks_addr(in, target_); + if (pr == ParseResult::need_more) { + if (in.size() > kMaxSocksAddr) { + server_reject("invalid request length"); + } + return Flow::proceed; + } + if (pr == ParseResult::invalid) { + server_reject("invalid address type"); + return Flow::proceed; + } + (void)to_upstream_.consume(target_.length); // the rest is the first payload + bool is_name = target_.type == 3; + if (is_name) { + // A literal address sent as a name. + in_addr v4{}; + in6_addr v6{}; + if (inet_pton(AF_INET, target_.host, &v4) == 1) { + auto *sa = reinterpret_cast(&target_.ip); + sa->sin_family = AF_INET; + sa->sin_addr = v4; + sa->sin_port = htons(target_.port); + is_name = false; + } else if (inet_pton(AF_INET6, target_.host, &v6) == 1) { + auto *sa = reinterpret_cast(&target_.ip); + sa->sin6_family = AF_INET6; + sa->sin6_addr = v6; + sa->sin6_port = htons(target_.port); + is_name = false; + } else { + if (cfg.acl && cfg.acl_hooks.outbound_blocked(target_.host) == 1) { + if (verbose) { + LOGI("outbound blocked %s", target_.host); + } + return close(); + } + if (!validate_hostname(target_.host, static_cast(std::strlen(target_.host)))) { + server_reject("invalid host name"); + return Flow::proceed; + } + } + } + if (verbose) { + if (target_.type == 4) { + LOGI("[%s] connect to [%s]:%u", cfg.label, target_.host, target_.port); + } else { + LOGI("[%s] connect to %s:%u", cfg.label, target_.host, target_.port); + } + } + if (!is_name) { + return server_connect(reinterpret_cast(&target_.ip)); + } + stage_ = Stage::resolving; + update_interest(); + if (!relay_->resolve(*this, target_.host, htons(target_.port))) { + LOGE("unable to resolve %s: too many queries", target_.host); + return close(); + } + return Flow::proceed; +} + +Flow Connection::server_connect(const sockaddr *target) noexcept +{ + TcpConfig &cfg = relay_->config_; + if (cfg.acl) { + char ip[INET6_ADDRSTRLEN] = {}; + const void *raw = target->sa_family == AF_INET + ? static_cast(&reinterpret_cast(target)->sin_addr) + : static_cast(&reinterpret_cast(target)->sin6_addr); + ss_inet_ntop(target->sa_family, raw, ip, sizeof(ip)); + if (cfg.acl_hooks.outbound_blocked(ip) == 1) { + if (verbose) { + LOGI("outbound blocked %s", ip); + } + return close(); + } + } + auto fd = open_upstream(target->sa_family, cfg.upstream, false); + if (!fd) { + LOGE("connect error"); + return close(); + } + if (!upstream_.open(*relay_->loop_, *this, *fd)) { + ss_socket_close(*fd); + return close(); + } + auto sent = start_connect(*fd, target, bytes_of(to_upstream_.readable()), cfg.fast_open); + if (!sent && sent.error().code == Errc::invalid) { + LOGE("fast open is not supported on this platform"); + cfg.fast_open = false; + sent = start_connect(*fd, target, {}, false); + } + if (!sent) { + ERROR("connect"); + return close(); + } + (void)to_upstream_.consume(*sent); + stage_ = Stage::connecting; + update_interest(); + return Flow::proceed; +} + +// -------------------------------------------------------------- stream + +Flow Connection::connected() noexcept +{ + if (client_kind()) { + relay_->clear_deadline(*this); // only the connect was timed + } + stage_ = Stage::stream; + if (pump_to_upstream() == Flow::closed) { + return Flow::closed; + } + update_interest(); + return Flow::proceed; +} + +Connection::Pull Connection::flush(ss_socket_t fd, FixedBuffer &src) noexcept +{ + auto s = send_some(fd, bytes_of(src.readable())); + if (!s) { + if (s.error().code == Errc::again) { + return Pull::again; + } + if (verbose) { + ERROR("send"); + } + return Pull::error; + } + (void)src.consume(*s); + return src.empty() ? Pull::data : Pull::again; +} + +Connection::Pull Connection::pull(ss_socket_t fd, bool decode, FixedBuffer &dst, + std::uint64_t *counter) noexcept +{ + dst.compact(); + auto out = bytes_of(dst.writable()); + std::size_t limit; + if (decode) { + // Hand out plaintext the decoder kept back before reading more. + auto step = dec_.decrypt({}, out); + if (!step) { + return Pull::error; + } + if (step->produced != 0) { + (void)dst.commit(step->produced); + return Pull::data; + } + limit = std::min({ static_cast(SOCKET_BUF_SIZE), dec_.accept_limit(), out.size() }); + } else { + limit = std::min(SOCKET_BUF_SIZE, out.size() - Method::kMaxOverhead); + } + if (limit == 0) { + return Pull::again; + } + auto r = recv_some(fd, { relay_->scratch_.data(), limit }); + if (!r) { + switch (r.error().code) { + case Errc::again: + return Pull::again; + case Errc::closed: + return Pull::eof; + default: + if (verbose) { + ERROR("recv"); + } + return Pull::error; + } + } + const TcpConfig &cfg = relay_->config_; + if (counter != nullptr) { + *counter += *r; + if (cfg.on_traffic != nullptr) { + cfg.on_traffic(); + } + } + if (cfg.kind == Kind::server && stage_ == Stage::stream) { + relay_->set_deadline(*this, Timed::idle); + } + std::span in(relay_->scratch_.data(), *r); + if (decode) { + auto step = dec_.decrypt(in, out); + if (!step || step->consumed != in.size()) { + if (cfg.kind == Kind::server) { + server_reject("authentication error"); + return Pull::nothing; + } + LOGE("invalid password or cipher"); + return Pull::error; + } + (void)dst.commit(step->produced); + return step->produced != 0 ? Pull::data : Pull::nothing; + } + auto w = enc_.seal(in, out); + if (!w) { + LOGE("invalid password or cipher"); + return Pull::error; + } + (void)dst.commit(*w); + return Pull::data; +} + +Flow Connection::pump_to_upstream() noexcept +{ + const bool decode = !client_kind(); + for (int round = 0; round < kPumpRounds; ++round) { + if (!to_upstream_.empty()) { + Pull f = flush(upstream_.fd(), to_upstream_); + if (f == Pull::error) { + return close(); + } + if (f == Pull::again) { + return Flow::proceed; + } + } + Pull p = pull(client_.fd(), decode, to_upstream_, relay_->config_.tx); + if (stage_ == Stage::stopped) { + return Flow::proceed; + } + switch (p) { + case Pull::data: + case Pull::nothing: + continue; + case Pull::again: + return Flow::proceed; + case Pull::eof: + case Pull::error: + return close(); + } + } + return Flow::proceed; +} + +Flow Connection::pump_to_client() noexcept +{ + const bool decode = client_kind(); + for (int round = 0; round < kPumpRounds; ++round) { + if (!to_client_.empty()) { + Pull f = flush(client_.fd(), to_client_); + if (f == Pull::error) { + return close(); + } + if (f == Pull::again) { + return Flow::proceed; + } + } + Pull p = pull(upstream_.fd(), decode, to_client_, relay_->config_.rx); + switch (p) { + case Pull::data: + if (!replied_) { + // Disable TCP_NODELAY once the first response is on its way. + replied_ = true; + if (!relay_->config_.no_delay) { + set_nodelay(client_.fd(), false); + set_nodelay(upstream_.fd(), false); + } + } + continue; + case Pull::nothing: + continue; + case Pull::again: + return Flow::proceed; + case Pull::eof: + case Pull::error: + return close(); + } + } + return Flow::proceed; +} + +void Connection::update_interest() noexcept +{ + int c = 0, u = 0; + switch (stage_) { + case Stage::socks_method: + case Stage::socks_request: + case Stage::udp_assoc: + case Stage::delay: + case Stage::header: + case Stage::stopped: + c = event::readable; + break; + case Stage::acl_resolve: + case Stage::resolving: + break; + case Stage::connecting: + u = event::writable; + break; + case Stage::stream: + c = (to_upstream_.empty() ? event::readable : 0) | (to_client_.empty() ? 0 : event::writable); + u = (to_client_.empty() ? event::readable : 0) | (to_upstream_.empty() ? 0 : event::writable); + break; + } + if (client_.is_open() && !client_.is_closing()) { + (void)client_.want(c); + } + if (upstream_.is_open() && !upstream_.is_closing()) { + (void)upstream_.want(u); + } +} + +} // namespace ss::relay diff --git a/src/relay/tcp.hpp b/src/relay/tcp.hpp new file mode 100644 index 00000000..663f0fb9 --- /dev/null +++ b/src/relay/tcp.hpp @@ -0,0 +1,232 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later */ +#ifndef SS_RELAY_TCP_HPP +#define SS_RELAY_TCP_HPP + +// The TCP relay shared by ss-local, ss-tunnel, ss-redir and ss-server, +// replacing the four hand-copied sets of accept/recv/send/close callbacks. +// +// Ownership +// * TcpRelay owns every Connection through a PoolPtr kept in a table +// indexed by pool slot; closing a connection is resetting that pointer. +// The pool then closes the connection's two sockets (begin_close) and +// recycles the slot once libuv has let go of them. +// * Nothing points at a Connection from outside: DNS answers come back +// with a generation-checked Handle, timeouts are intrusive list links +// the connection removes itself from when it closes. +// * Buffers, cipher state and the handshake state live inside the +// connection, so accepting a connection allocates nothing; when the pool +// is full the new connection is refused. +// +// Data moves in lock step, as in the C relays: one direction reads its +// source only while its buffer towards the other side is empty. + +#include +#include +#include +#include + +#include "core/buffer.hpp" +#include "core/pool.hpp" +#include "core/region.hpp" +#include "core/result.hpp" +#include "event/loop.hpp" +#include "relay/codec.hpp" +#include "relay/resolver.hpp" +#include "relay/socket.hpp" + +namespace ss::relay { + +enum class Kind : std::uint8_t { local, tunnel, redir, server }; + +// The ACL module (acl.cpp), for the binaries that link it. +struct AclHooks { + int (*match_host)(const char *host) = nullptr; // > 0 listed as bypass/block, < 0 as proxy, 0 neither + int (*mode)() = nullptr; // BLACK_LIST or WHITE_LIST + int (*outbound_blocked)(const char *host) = nullptr; +}; + +struct TcpConfig { + Kind kind = Kind::local; + // Client kinds: the ss-server endpoints (or the plugin's local port). + std::span remotes; + // ss-tunnel: the encoded target address. + std::array tunnel_target{}; + std::size_t tunnel_target_len = 0; + int timeout_secs = 60; + std::uint32_t max_connections = 1024; + bool fast_open = false; + bool no_delay = false; + bool printable_salt = false; // ss-local experiment + bool acl = false; // with `acl_hooks` set + AclHooks acl_hooks; + bool tproxy = false; // ss-redir: the target is the local address + bool vpn = false; // Android: protect sockets, no local DNS for ACL + const char *label = ""; // ss-server: the port, for logs + SocketTuning upstream; + std::uint64_t *tx = nullptr; // bytes received from clients + std::uint64_t *rx = nullptr; // bytes received from upstreams + void (*on_traffic)() = nullptr; + // ss-server: a client failed the handshake (logging, nftables). + void (*report_failure)(ss_socket_t client, const char *reason) = nullptr; +}; + +class TcpRelay; + +// Buffers towards each side hold one received chunk plus cipher overhead. +inline constexpr std::size_t kRelayBuffer = SOCKET_BUF_SIZE + Method::kMaxOverhead; +inline constexpr std::size_t kHandshakeBuffer = 1024; + +class Connection : public event::Deadline { +public: + explicit Connection(TcpRelay &relay) noexcept : relay_(&relay) {} + Connection(const Connection &) = delete; + Connection &operator=(const Connection &) = delete; + + void begin_close(CloseCompletion done) noexcept; + + event::Flow on_client(int events) noexcept; + event::Flow on_upstream(int events) noexcept; + void on_deadline() noexcept; + void on_resolved(const sockaddr *address) noexcept; + +private: + friend class TcpRelay; + + enum class Stage : std::uint8_t { + socks_method, // local: waiting for the method selection + socks_request, // local: waiting for the CONNECT request + acl_resolve, // local: resolving a name for the ACL + udp_assoc, // local: UDP ASSOCIATE held open until the client leaves + delay, // client kinds: waiting briefly for the first payload + header, // server: waiting for the request header + resolving, // server: resolving the target + connecting, + stream, + stopped, // server: failed the handshake; drain until timeout + }; + enum class Timed : std::uint8_t { none, delay, connect, idle }; + + event::Flow start(ss_socket_t fd) noexcept; + event::Flow close() noexcept; + + // Client side handshakes. + event::Flow socks_input() noexcept; + event::Flow socks_request() noexcept; + event::Flow socks_route() noexcept; + event::Flow delay_input() noexcept; + event::Flow start_upstream(const sockaddr *target, bool direct) noexcept; + // Server side. + event::Flow server_header() noexcept; + event::Flow server_connect(const sockaddr *target) noexcept; + void server_reject(const char *reason) noexcept; + + event::Flow connected() noexcept; + event::Flow pump_to_upstream() noexcept; + event::Flow pump_to_client() noexcept; + void update_interest() noexcept; + + enum class Pull : std::uint8_t { data, nothing, again, eof, error }; + Pull pull(ss_socket_t fd, bool decode, FixedBuffer &dst, std::uint64_t *counter) noexcept; + Pull flush(ss_socket_t fd, FixedBuffer &src) noexcept; + + bool client_kind() const noexcept; + + TcpRelay *relay_; + CloseCompletion done_; + event::Poll client_; + event::Poll upstream_; + event::CloseGroup closer_; + + FixedBuffer to_upstream_; + FixedBuffer to_client_; + FixedBuffer handshake_; + Encoder enc_; + Decoder dec_; + + std::array target_header_{}; + std::size_t target_header_len_ = 0; + SocksTarget target_{}; + sockaddr_storage upstream_addr_{}; + + Stage stage_ = Stage::socks_method; + Timed timed_ = Timed::none; + bool direct_ = false; + bool dns_done_ = false; + bool dns_ok_ = false; + bool replied_ = false; // first upstream bytes reached the client + sockaddr_storage dns_answer_{}; +}; + +class TcpRelay { +public: + TcpRelay() noexcept = default; + TcpRelay(const TcpRelay &) = delete; + TcpRelay &operator=(const TcpRelay &) = delete; + ~TcpRelay(); + + // `resolver` is needed by ss-server and by ss-local with an ACL. + Status init(event::Loop &loop, Method &method, const TcpConfig &config, Resolver *resolver) noexcept; + // Start accepting on a bound, listening socket; the relay owns it now. + Status listen(ss_socket_t fd) noexcept; + + // Close every connection and listener. Run the loop until closed(). + void shutdown() noexcept; + bool closed() const noexcept; + + std::uint32_t live() const noexcept { return pool_.live(); } + std::uint64_t refused() const noexcept { return refused_; } + +private: + friend class Connection; + + struct Listener { + TcpRelay *relay = nullptr; + event::Flow on_ready(int events) noexcept; + event::Poll poll; + }; + static constexpr std::size_t kMaxListeners = 16; + + void accept_from(Listener &listener) noexcept; + void close(Connection &conn) noexcept; + void slot_freed() noexcept; + void set_deadline(Connection &conn, Connection::Timed which) noexcept; + void clear_deadline(Connection &conn) noexcept; + void on_timer() noexcept; + void schedule() noexcept; + Status resolve(Connection &conn, const char *host, std::uint16_t port_be) noexcept; + static void resolved(const ResolveCallback &cb, const sockaddr *address) noexcept; + const sockaddr_storage &pick_remote() noexcept; + void pause_accept() noexcept; + void resume_accept() noexcept; + + event::Loop *loop_ = nullptr; + Method *method_ = nullptr; + Resolver *resolver_ = nullptr; + TcpConfig config_{}; + + ObjectPool pool_; + Region owners_region_; + std::span > owners_; + + std::array listeners_{}; + std::size_t listener_count_ = 0; + bool accept_paused_ = false; + std::uint64_t refused_ = 0; + std::uint64_t last_refusal_log_ = 0; + + event::DeadlineList delay_list_; + event::DeadlineList connect_list_; + event::DeadlineList idle_list_; + event::Timer timer_; + std::uint64_t timer_due_ = 0; // 0 when the timer is idle + event::CloseGroup closer_; + bool shutting_down_ = false; + bool own_closed_ = true; + + // One receive at a time, shared by every connection. + std::array scratch_{}; +}; + +} // namespace ss::relay + +#endif diff --git a/src/tunnel.cpp b/src/tunnel.cpp index 4015a7ee..9b0d249d 100644 --- a/src/tunnel.cpp +++ b/src/tunnel.cpp @@ -52,9 +52,18 @@ #include "netutils.h" #include "utils.h" #include "plugin.h" -#include "tunnel.h" +#include "ss_event.h" +#include "crypto.h" +#include "jconf.h" +#include "common.h" +#include "udprelay.h" #include "ss_windows.h" +#include "relay/tcp.hpp" + +#undef min +#undef max + #ifndef EAGAIN #define EAGAIN EWOULDBLOCK #endif @@ -63,20 +72,6 @@ #define EWOULDBLOCK EAGAIN #endif -static void accept_cb(SS_P_ ss_io *w, int revents); -static void server_recv_cb(SS_P_ ss_io *w, int revents); -static void server_send_cb(SS_P_ ss_io *w, int revents); -static void remote_recv_cb(SS_P_ ss_io *w, int revents); -static void remote_send_cb(SS_P_ ss_io *w, int revents); - -static remote_t *new_remote(int fd, int timeout); -static server_t *new_server(int fd); - -static void free_remote(remote_t *remote); -static void close_and_free_remote(SS_P_ remote_t *remote); -static void free_server(server_t *server); -static void close_and_free_server(SS_P_ server_t *server); - static int validate_tunnel_addr(const ss_addr_t *addr) { @@ -134,19 +129,6 @@ static struct plugin_watcher_t { } plugin_watcher; #endif -#ifndef __MINGW32__ -static int -setnonblocking(int fd) -{ - int flags; - if (-1 == (flags = fcntl(fd, F_GETFL, 0))) { - flags = 0; - } - return fcntl(fd, F_SETFL, flags | O_NONBLOCK); -} - -#endif - int create_and_bind(const char *addr, const char *port) { @@ -214,667 +196,6 @@ create_and_bind(const char *addr, const char *port) return listen_sock; } -static void -server_recv_cb(SS_P_ ss_io *w, int revents) -{ - server_ctx_t *server_recv_ctx = (server_ctx_t *)w; - server_t *server = server_recv_ctx->server; - remote_t *remote = server->remote; - - if (remote == NULL) { - close_and_free_server(SS_A_ server); - return; - } - - ssize_t r = recv(server->fd, remote->buf->data, SOCKET_BUF_SIZE, 0); - - if (r == 0) { - // connection closed - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } else if (r == -1) { - if (ss_socket_error() == EAGAIN || ss_socket_error() == EWOULDBLOCK) { - // no data - // continue to wait for recv - return; - } else { - ERROR("server recv"); - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } - } - - remote->buf->len = r; - - int err = crypto->encrypt(remote->buf, server->e_ctx, SOCKET_BUF_SIZE); - - if (err) { - LOGE("invalid password or cipher"); - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } - - int s = send(remote->fd, remote->buf->data, remote->buf->len, 0); - - if (s == -1) { - if (ss_socket_error() == EAGAIN || ss_socket_error() == EWOULDBLOCK) { - // no data, wait for send - remote->buf->idx = 0; - ss_io_stop(SS_A_ & server_recv_ctx->io); - ss_io_start(SS_A_ & remote->send_ctx->io); - return; - } else { - ERROR("send"); - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } - } else if (s < remote->buf->len) { - remote->buf->len -= s; - remote->buf->idx = s; - ss_io_stop(SS_A_ & server_recv_ctx->io); - ss_io_start(SS_A_ & remote->send_ctx->io); - return; - } -} - -static void -server_send_cb(SS_P_ ss_io *w, int revents) -{ - server_ctx_t *server_send_ctx = (server_ctx_t *)w; - server_t *server = server_send_ctx->server; - remote_t *remote = server->remote; - if (server->buf->len == 0) { - // close and free - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } else { - // has data to send - ssize_t s = send(server->fd, server->buf->data + server->buf->idx, - server->buf->len, 0); - if (s == -1) { - if (ss_socket_error() != EAGAIN && ss_socket_error() != EWOULDBLOCK) { - ERROR("send"); - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - } - return; - } else if (s < server->buf->len) { - // partly sent, move memory, wait for the next time to send - server->buf->len -= s; - server->buf->idx += s; - return; - } else { - // all sent out, wait for reading - server->buf->len = 0; - server->buf->idx = 0; - ss_io_stop(SS_A_ & server_send_ctx->io); - if (remote != NULL) { - ss_io_start(SS_A_ & remote->recv_ctx->io); - } else { - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } - } - } -} - -static void -remote_timeout_cb(SS_P_ ss_timer *watcher, int revents) -{ - remote_ctx_t *remote_ctx - = ss_container_of(watcher, remote_ctx_t, watcher); - - remote_t *remote = remote_ctx->remote; - server_t *server = remote->server; - - if (verbose) { - LOGI("TCP connection timeout"); - } - - ss_timer_stop(SS_A_ watcher); - - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); -} - -static void -remote_recv_cb(SS_P_ ss_io *w, int revents) -{ - remote_ctx_t *remote_recv_ctx = (remote_ctx_t *)w; - remote_t *remote = remote_recv_ctx->remote; - server_t *server = remote->server; - - ssize_t r = recv(remote->fd, server->buf->data, SOCKET_BUF_SIZE, 0); - - if (r == 0) { - // connection closed - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } else if (r == -1) { - if (ss_socket_error() == EAGAIN || ss_socket_error() == EWOULDBLOCK) { - // no data - // continue to wait for recv - return; - } else { - ERROR("remote recv"); - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } - } - - server->buf->len = r; - - int err = crypto->decrypt(server->buf, server->d_ctx, SOCKET_BUF_SIZE); - if (err == CRYPTO_ERROR) { - LOGE("invalid password or cipher"); - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } else if (err == CRYPTO_NEED_MORE) { - return; // Wait for more - } - - int s = send(server->fd, server->buf->data, server->buf->len, 0); - - if (s == -1) { - if (ss_socket_error() == EAGAIN || ss_socket_error() == EWOULDBLOCK) { - // no data, wait for send - server->buf->idx = 0; - ss_io_stop(SS_A_ & remote_recv_ctx->io); - ss_io_start(SS_A_ & server->send_ctx->io); - } else { - ERROR("send"); - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } - } else if (s < server->buf->len) { - server->buf->len -= s; - server->buf->idx = s; - ss_io_stop(SS_A_ & remote_recv_ctx->io); - ss_io_start(SS_A_ & server->send_ctx->io); - } - - // Disable TCP_NODELAY after the first response are sent - if (!remote->recv_ctx->connected && !no_delay) { - int opt = 0; - ss_setsockopt(server->fd, SOL_TCP, TCP_NODELAY, &opt, sizeof(opt)); - ss_setsockopt(remote->fd, SOL_TCP, TCP_NODELAY, &opt, sizeof(opt)); - } - remote->recv_ctx->connected = 1; -} - -static void -remote_send_cb(SS_P_ ss_io *w, int revents) -{ - remote_ctx_t *remote_send_ctx = (remote_ctx_t *)w; - remote_t *remote = remote_send_ctx->remote; - server_t *server = remote->server; - - ss_timer_stop(SS_A_ & remote_send_ctx->watcher); - - if (!remote_send_ctx->connected) { - int r = 0; - - if (remote->addr == NULL) { - struct sockaddr_storage addr; - socklen_t len = sizeof(struct sockaddr_storage); - r = getpeername(remote->fd, (struct sockaddr *)&addr, &len); - } - - if (r == 0) { - remote_send_ctx->connected = 1; - - if (remote->buf->len != 0) { - LOGE("unexpected pending tunnel data"); - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } - buffer_t *abuf = remote->buf; - - ss_addr_t *sa = &server->destaddr; - struct ss_ip ip; - if (ss_ip_init(&ip, sa->host) != -1) { - if (ip.version == 4) { - // send as IPv4 - struct in_addr host; - memset(&host, 0, sizeof(struct in_addr)); - int host_len = sizeof(struct in_addr); - - if (inet_pton(AF_INET, sa->host, &host) == -1) { - FATAL("IP parser error"); - } - abuf->data[abuf->len++] = 1; - memcpy(abuf->data + abuf->len, &host, host_len); - abuf->len += host_len; - } else if (ip.version == 6) { - // send as IPv6 - struct in6_addr host; - memset(&host, 0, sizeof(struct in6_addr)); - int host_len = sizeof(struct in6_addr); - - if (inet_pton(AF_INET6, sa->host, &host) == -1) { - FATAL("IP parser error"); - } - abuf->data[abuf->len++] = 4; - memcpy(abuf->data + abuf->len, &host, host_len); - abuf->len += host_len; - } else { - FATAL("IP parser error"); - } - } else { - // send as domain - int host_len = strlen(sa->host); - - if (host_len > UINT8_MAX) { - LOGE("tunnel host name is too long"); - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } - abuf->data[abuf->len++] = 3; - abuf->data[abuf->len++] = host_len; - memcpy(abuf->data + abuf->len, sa->host, host_len); - abuf->len += host_len; - } - - uint16_t port_num; - if (ss_parse_uint16_port(sa->port, &port_num) == -1) { - LOGE("invalid tunnel port"); - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } - uint16_t port = htons(port_num); - memcpy(abuf->data + abuf->len, &port, 2); - abuf->len += 2; - - int err = crypto->encrypt(abuf, server->e_ctx, SOCKET_BUF_SIZE); - - if (err) { - LOGE("invalid password or cipher"); - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } - - ss_io_start(SS_A_ & remote->recv_ctx->io); - } else { - ERROR("getpeername"); - // not connected - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } - } - - if (remote->buf->len == 0) { - // close and free - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } else { - // has data to send - ssize_t s = -1; - if (remote->addr != NULL) { -#if defined(TCP_FASTOPEN_WINSOCK) - DWORD s = -1; - DWORD err = 0; - do { - int optval = 1; - // Set fast open option - if (ss_setsockopt(remote->fd, IPPROTO_TCP, TCP_FASTOPEN, - &optval, sizeof(optval)) != 0) { - ERROR("setsockopt"); - break; - } - // Load ConnectEx function - LPFN_CONNECTEX ConnectEx = winsock_getconnectex(); - if (ConnectEx == NULL) { - LOGE("Cannot load ConnectEx() function"); - err = WSAENOPROTOOPT; - break; - } - // ConnectEx requires a bound socket - if (winsock_dummybind(remote->fd, - (struct sockaddr *)&(remote->addr)) != 0) { - ERROR("bind"); - break; - } - // Call ConnectEx to send data - memset(&remote->olap, 0, sizeof(remote->olap)); - remote->connect_ex_done = 0; - if (ConnectEx(remote->fd, (const struct sockaddr *)&(remote->addr), - get_sockaddr_len(remote->addr), remote->buf->data, remote->buf->len, - &s, &remote->olap)) { - remote->connect_ex_done = 1; - break; - } - // XXX: ConnectEx pending, check later in remote_send - if (WSAGetLastError() == ERROR_IO_PENDING) { - err = CONNECT_IN_PROGRESS; - break; - } - ERROR("ConnectEx"); - } while (0); - // Set error number - if (err) { - SetLastError(err); - } -#elif defined(CONNECT_DATA_IDEMPOTENT) - ((struct sockaddr_in *)&(remote->addr))->sin_len = sizeof(struct sockaddr_in); - sa_endpoints_t endpoints; - memset((char *)&endpoints, 0, sizeof(endpoints)); - endpoints.sae_dstaddr = (struct sockaddr *)&(remote->addr); - endpoints.sae_dstaddrlen = get_sockaddr_len(remote->addr); - s = connectx(remote->fd, &endpoints, SAE_ASSOCID_ANY, - CONNECT_RESUME_ON_READ_WRITE | CONNECT_DATA_IDEMPOTENT, - NULL, 0, NULL, NULL); -#elif defined(TCP_FASTOPEN_CONNECT) - int optval = 1; - if (ss_setsockopt(remote->fd, IPPROTO_TCP, TCP_FASTOPEN_CONNECT, - (void *)&optval, sizeof(optval)) < 0) - FATAL("failed to set TCP_FASTOPEN_CONNECT"); - s = connect(remote->fd, remote->addr, get_sockaddr_len(remote->addr)); - if (s == 0) - s = send(remote->fd, remote->buf->data, remote->buf->len, 0); -#elif defined(MSG_FASTOPEN) - s = sendto(remote->fd, remote->buf->data + remote->buf->idx, - remote->buf->len, MSG_FASTOPEN, remote->addr, - get_sockaddr_len(remote->addr)); -#else - FATAL("tcp fast open is not supported on this platform"); -#endif - - remote->addr = NULL; - - if (s == -1) { - if (ss_socket_error() == CONNECT_IN_PROGRESS) { - ss_io_start(SS_A_ & remote_send_ctx->io); - ss_timer_start(SS_A_ & remote_send_ctx->watcher); - } else { - fast_open = 0; - if (ss_socket_error() == EOPNOTSUPP || ss_socket_error() == EPROTONOSUPPORT || - ss_socket_error() == ENOPROTOOPT) { - LOGE("fast open is not supported on this platform"); - } else { - ERROR("fast_open_connect"); - } - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - } - return; - } - } else { - s = send(remote->fd, remote->buf->data + remote->buf->idx, - remote->buf->len, 0); - } - - if (s == -1) { - if (ss_socket_error() != EAGAIN && ss_socket_error() != EWOULDBLOCK) { - ERROR("send"); - // close and free - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - } - return; - } else if (s < remote->buf->len) { - // partly sent, move memory, wait for the next time to send - remote->buf->len -= s; - remote->buf->idx += s; - return; - } else { - // all sent out, wait for reading - remote->buf->len = 0; - remote->buf->idx = 0; - ss_io_stop(SS_A_ & remote_send_ctx->io); - ss_io_start(SS_A_ & server->recv_ctx->io); - } - } -} - -static remote_t * -new_remote(int fd, int timeout) -{ - remote_t *remote = (remote_t *)ss_malloc(sizeof(remote_t)); - memset(remote, 0, sizeof(remote_t)); - - remote->recv_ctx = (struct remote_ctx *)ss_malloc(sizeof(remote_ctx_t)); - remote->send_ctx = (struct remote_ctx *)ss_malloc(sizeof(remote_ctx_t)); - remote->buf = (buffer_t *)ss_malloc(sizeof(buffer_t)); - balloc(remote->buf, SOCKET_BUF_SIZE); - memset(remote->recv_ctx, 0, sizeof(remote_ctx_t)); - memset(remote->send_ctx, 0, sizeof(remote_ctx_t)); - remote->fd = fd; - remote->recv_ctx->remote = remote; - remote->recv_ctx->connected = 0; - remote->send_ctx->remote = remote; - remote->send_ctx->connected = 0; - - ss_io_init(&remote->recv_ctx->io, remote_recv_cb, fd, SS_READ); - ss_io_init(&remote->send_ctx->io, remote_send_cb, fd, SS_WRITE); - ss_timer_init(&remote->send_ctx->watcher, remote_timeout_cb, - min(MAX_CONNECT_TIMEOUT, timeout), 0); - - return remote; -} - -static void -free_remote(remote_t *remote) -{ - if (remote->server != NULL) { - remote->server->remote = NULL; - } - if (remote->buf != NULL) { - bfree(remote->buf); - ss_free(remote->buf); - } - ss_free(remote->recv_ctx); - ss_free(remote->send_ctx); - ss_free(remote); -} - -static void -close_and_free_remote(SS_P_ remote_t *remote) -{ - if (remote != NULL) { - ss_timer_stop(SS_A_ & remote->send_ctx->watcher); - ss_io_stop(SS_A_ & remote->send_ctx->io); - ss_io_stop(SS_A_ & remote->recv_ctx->io); - ss_socket_close(remote->fd); - free_remote(remote); - } -} - -static server_t * -new_server(int fd) -{ - server_t *server = (server_t *)ss_malloc(sizeof(server_t)); - memset(server, 0, sizeof(server_t)); - - server->recv_ctx = (struct server_ctx *)ss_malloc(sizeof(server_ctx_t)); - server->send_ctx = (struct server_ctx *)ss_malloc(sizeof(server_ctx_t)); - server->buf = (buffer_t *)ss_malloc(sizeof(buffer_t)); - balloc(server->buf, SOCKET_BUF_SIZE); - memset(server->recv_ctx, 0, sizeof(server_ctx_t)); - memset(server->send_ctx, 0, sizeof(server_ctx_t)); - server->fd = fd; - server->recv_ctx->server = server; - server->recv_ctx->connected = 0; - server->send_ctx->server = server; - server->send_ctx->connected = 0; - - server->e_ctx = (cipher_ctx_t *)ss_malloc(sizeof(cipher_ctx_t)); - server->d_ctx = (cipher_ctx_t *)ss_malloc(sizeof(cipher_ctx_t)); - crypto->ctx_init(crypto->cipher, server->e_ctx, 1); - crypto->ctx_init(crypto->cipher, server->d_ctx, 0); - cipher_ctx_pair(server->e_ctx, server->d_ctx); - - ss_io_init(&server->recv_ctx->io, server_recv_cb, fd, SS_READ); - ss_io_init(&server->send_ctx->io, server_send_cb, fd, SS_WRITE); - - return server; -} - -static void -free_server(server_t *server) -{ - if (server->remote != NULL) { - // NOLINTNEXTLINE(clang-analyzer-unix.Malloc): back-pointers are cleared symmetrically on free - server->remote->server = NULL; - } - if (server->e_ctx != NULL) { - crypto->ctx_release(server->e_ctx); - ss_free(server->e_ctx); - } - if (server->d_ctx != NULL) { - crypto->ctx_release(server->d_ctx); - ss_free(server->d_ctx); - } - if (server->buf != NULL) { - bfree(server->buf); - ss_free(server->buf); - } - ss_free(server->recv_ctx); - ss_free(server->send_ctx); - ss_free(server); -} - -static void -close_and_free_server(SS_P_ server_t *server) -{ - if (server != NULL) { - ss_io_stop(SS_A_ & server->send_ctx->io); - ss_io_stop(SS_A_ & server->recv_ctx->io); - ss_socket_close(server->fd); - free_server(server); - } -} - -static void -accept_cb(SS_P_ ss_io *w, int revents) -{ - struct listen_ctx *listener = (struct listen_ctx *)w; - int serverfd = ss_accept(listener->fd, NULL, NULL); - if (serverfd == -1) { - ERROR("accept"); - return; - } - setnonblocking(serverfd); - int opt = 1; - ss_setsockopt(serverfd, SOL_TCP, TCP_NODELAY, &opt, sizeof(opt)); -#ifdef SO_NOSIGPIPE - ss_setsockopt(serverfd, SOL_SOCKET, SO_NOSIGPIPE, &opt, sizeof(opt)); -#endif - - int index = (int)randombytes_uniform((uint32_t)listener->remote_num); - struct sockaddr *remote_addr = listener->remote_addr[index]; - - int protocol = IPPROTO_TCP; - if (listener->mptcp < 0) { - protocol = IPPROTO_MPTCP; // Enable upstream MPTCP - } - int remotefd = ss_socket(remote_addr->sa_family, SOCK_STREAM, protocol); - if (remotefd == -1) { - ERROR("socket"); - return; - } - -#ifdef __ANDROID__ - if (vpn) { - int not_protect = 0; - if (remote_addr->sa_family == AF_INET) { - struct sockaddr_in *s = (struct sockaddr_in *)remote_addr; - if (s->sin_addr.s_addr == inet_addr("127.0.0.1")) - not_protect = 1; - } - if (!not_protect) { - if (protect_socket(remotefd) == -1) { - ERROR("protect_socket"); - ss_socket_close(remotefd); - return; - } - } - } -#endif - - int keepAlive = 1; - ss_setsockopt(remotefd, SOL_SOCKET, SO_KEEPALIVE, (void *)&keepAlive, sizeof(keepAlive)); - ss_setsockopt(remotefd, SOL_TCP, TCP_NODELAY, &opt, sizeof(opt)); -#ifdef SO_NOSIGPIPE - ss_setsockopt(remotefd, SOL_SOCKET, SO_NOSIGPIPE, &opt, sizeof(opt)); -#endif - - // Enable out-of-tree MPTCP - if (listener->mptcp > 1) { - int err = ss_setsockopt(remotefd, SOL_TCP, listener->mptcp, &opt, sizeof(opt)); - if (err == -1) { - ERROR("failed to enable out-of-tree multipath TCP"); - } - } else if (listener->mptcp == 1) { - int i = 0; - while ((listener->mptcp = mptcp_enabled_values[i]) > 0) { - int err = ss_setsockopt(remotefd, SOL_TCP, listener->mptcp, &opt, sizeof(opt)); - if (err != -1) { - break; - } - i++; - } - if (listener->mptcp == 0) { - ERROR("failed to enable out-of-tree multipath TCP"); - } - } - - if (tcp_outgoing_sndbuf > 0) { - ss_setsockopt(remotefd, SOL_SOCKET, SO_SNDBUF, &tcp_outgoing_sndbuf, sizeof(int)); - } - - if (tcp_outgoing_rcvbuf > 0) { - ss_setsockopt(remotefd, SOL_SOCKET, SO_RCVBUF, &tcp_outgoing_rcvbuf, sizeof(int)); - } - - // Setup - setnonblocking(remotefd); -#ifdef SET_INTERFACE - if (listener->iface) { - if (setinterface(remotefd, listener->iface) == -1) - ERROR("setinterface"); - } -#endif - - server_t *server = new_server(serverfd); - remote_t *remote = new_remote(remotefd, listener->timeout); - server->destaddr = listener->tunnel_addr; - server->remote = remote; - remote->server = server; - - if (fast_open) { - remote->addr = remote_addr; - } else { - int r = connect(remotefd, remote_addr, get_sockaddr_len(remote_addr)); - - if (r == -1 && ss_socket_error() != CONNECT_IN_PROGRESS) { - ERROR("connect"); - close_and_free_remote(SS_A_ remote); - close_and_free_server(SS_A_ server); - return; - } - } - - // listen to remote connected event - ss_io_start(SS_A_ & remote->send_ctx->io); - ss_timer_start(SS_A_ & remote->send_ctx->watcher); -} - static void signal_cb(SS_P_ ss_signal *w, int revents) { @@ -1617,12 +938,9 @@ Same behavior as `-V`; see that option for details. Short alias: `-V`. FATAL("failed to initialize ciphers"); // Setup proxy context - struct listen_ctx listen_ctx; - memset(&listen_ctx, 0, sizeof(struct listen_ctx)); - listen_ctx.tunnel_addr = tunnel_addr; - listen_ctx.remote_num = remote_num; - listen_ctx.remote_addr = (struct sockaddr **)ss_malloc(sizeof(struct sockaddr *) * remote_num); - memset(listen_ctx.remote_addr, 0, sizeof(struct sockaddr *) * remote_num); + int remote_count = 0; + struct sockaddr_storage remote_storage[MAX_REMOTE_NUM]; + memset(remote_storage, 0, sizeof(remote_storage)); for (i = 0; i < remote_num; i++) { char *host = remote_addr[i].host; char *port = remote_addr[i].port == NULL ? remote_port : remote_addr[i].port; @@ -1630,25 +948,58 @@ Same behavior as `-V`; see that option for details. Short alias: `-V`. host = plugin_host; port = plugin_port; } - struct sockaddr_storage *storage = (struct sockaddr_storage *)ss_malloc(sizeof(struct sockaddr_storage)); - memset(storage, 0, sizeof(struct sockaddr_storage)); - if (get_sockaddr(host, port, storage, 1, ipv6first) == -1) { + if (get_sockaddr(host, port, &remote_storage[i], 1, ipv6first) == -1) { FATAL("failed to resolve the provided hostname"); } - listen_ctx.remote_addr[i] = (struct sockaddr *)storage; + remote_count++; if (plugin != NULL) break; } - listen_ctx.timeout = timeout_secs; - listen_ctx.iface = iface; - listen_ctx.mptcp = mptcp; + + ss::relay::TcpConfig tcp_config; + tcp_config.kind = ss::relay::Kind::tunnel; + tcp_config.remotes = { remote_storage, (size_t)remote_count }; + tcp_config.timeout_secs = timeout_secs; + tcp_config.max_connections = DEFAULT_MAX_CONNECTIONS; + tcp_config.fast_open = fast_open != 0; + tcp_config.no_delay = no_delay != 0; + tcp_config.upstream.sndbuf = tcp_outgoing_sndbuf; + tcp_config.upstream.rcvbuf = tcp_outgoing_rcvbuf; + tcp_config.upstream.mptcp = mptcp; + tcp_config.upstream.iface = iface; + tcp_config.upstream.keepalive = true; +#ifdef __ANDROID__ + tcp_config.vpn = vpn != 0; + tcp_config.upstream.protect = protect_socket; +#endif + { + uint16_t tunnel_port = 0; + ss_parse_uint16_port(tunnel_addr.port, &tunnel_port); + tcp_config.tunnel_target_len = ss::relay::encode_socks_host(tunnel_addr.host, tunnel_port, + tcp_config.tunnel_target.data()); + if (tcp_config.tunnel_target_len == 0) { + FATAL("tunnel host name is too long"); + } + } LOGI("listening at %s:%s", local_addr, local_port); struct ss_loop *loop = SS_DEFAULT; + ss::event::Loop tcp_loop; + ss::relay::Method tcp_method; + ss::relay::TcpRelay tcp_relay; + if (!tcp_loop.adopt(ss_native_loop(loop))) { + FATAL("failed to set up the event loop"); + } if (mode != UDP_ONLY) { + if (!tcp_method.init(method, password, key, ss::crypto::Role::client)) { + FATAL("failed to initialize ciphers"); + } + if (!tcp_relay.init(tcp_loop, tcp_method, tcp_config, NULL)) { + FATAL("failed to set up the TCP relay"); + } // Setup socket int listenfd; listenfd = create_and_bind(local_addr, local_port); @@ -1658,12 +1009,9 @@ Same behavior as `-V`; see that option for details. Short alias: `-V`. if (listen(listenfd, SOMAXCONN) == -1) { FATAL("listen() error"); } - setnonblocking(listenfd); - - listen_ctx.fd = listenfd; - - ss_io_init(&listen_ctx.io, accept_cb, listenfd, SS_READ); - ss_io_start(loop, &listen_ctx.io); + if (!tcp_relay.listen(listenfd)) { + FATAL("failed to watch the listening socket"); + } } // Setup UDP @@ -1678,7 +1026,7 @@ Same behavior as `-V`; see that option for details. Short alias: `-V`. } struct sockaddr *addr = (struct sockaddr *)storage; if (init_udprelay(local_addr, local_port, addr, get_sockaddr_len(addr), - tunnel_addr, mtu, crypto, listen_ctx.timeout, iface) == -1) { + tunnel_addr, mtu, crypto, timeout_secs, iface) == -1) { FATAL("failed to initialize UDP relay"); } } @@ -1704,9 +1052,10 @@ Same behavior as `-V`; see that option for details. Short alias: `-V`. stop_plugin(); } - for (i = 0; i < remote_num; i++) - ss_free(listen_ctx.remote_addr[i]); - ss_free(listen_ctx.remote_addr); + tcp_relay.shutdown(); + while (!tcp_relay.closed()) { + tcp_loop.run_nowait(); + } #ifdef __MINGW32__ if (plugin_watcher.valid) { diff --git a/src/tunnel.h b/src/tunnel.h deleted file mode 100644 index 773a6db3..00000000 --- a/src/tunnel.h +++ /dev/null @@ -1,95 +0,0 @@ -/* - * tunnel.h - Define tunnel's buffers and callbacks - * - * Copyright (C) 2013 - 2019, Max Lv - * - * This file is part of the shadowsocks-libev. - * - * shadowsocks-libev is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation; either version 3 of the License, or - * (at your option) any later version. - * - * shadowsocks-libev is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. - * - * You should have received a copy of the GNU General Public License - * along with shadowsocks-libev; see the file COPYING. If not, see - * . - */ - -#ifndef _TUNNEL_H -#define _TUNNEL_H - -#include "ss_event.h" - -#ifdef __MINGW32__ -#include "ss_windows.h" -#endif - -#include "crypto.h" -#include "jconf.h" - -#include "common.h" - -#ifdef __cplusplus -extern "C" { -#endif - -typedef struct listen_ctx { - ss_io io; - ss_addr_t tunnel_addr; - char *iface; - int remote_num; - int timeout; - int fd; - int mptcp; - struct sockaddr **remote_addr; -} listen_ctx_t; - -typedef struct server_ctx { - ss_io io; - int connected; - struct server *server; -} server_ctx_t; - -typedef struct server { - int fd; - - buffer_t *buf; - cipher_ctx_t *e_ctx; - cipher_ctx_t *d_ctx; - struct server_ctx *recv_ctx; - struct server_ctx *send_ctx; - struct remote *remote; - ss_addr_t destaddr; -} server_t; - -typedef struct remote_ctx { - ss_io io; - ss_timer watcher; - int connected; - struct remote *remote; -} remote_ctx_t; - -typedef struct remote { - int fd; -#ifdef TCP_FASTOPEN_WINSOCK - OVERLAPPED olap; - int connect_ex_done; -#endif - buffer_t *buf; - struct remote_ctx *recv_ctx; - struct remote_ctx *send_ctx; - struct server *server; - struct sockaddr *addr; - uint32_t counter; -} remote_t; - -#ifdef __cplusplus -} -#endif - -#endif // _TUNNEL_H diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index ac58d1a8..c61f1ade 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -354,6 +354,15 @@ if(SS_BUILD_EXECUTABLES AND NOT CMAKE_CROSSCOMPILING) --self --printable-salt --bin ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}) set_tests_properties(test_printable_salt_interop PROPERTIES LABELS "integration" TIMEOUT 90) + # ss-tunnel through the C relay engine against ss-server (and + # shadowsocks-rust when installed), every method. + set(SS_TUNNEL_INTEROP_ARGS --bin ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}) + if(SS_ENABLE_LEGACY) + list(APPEND SS_TUNNEL_INTEROP_ARGS --legacy) + endif() + add_test(NAME test_tunnel_interop + COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/tunnel_interop.py ${SS_TUNNEL_INTEROP_ARGS}) + set_tests_properties(test_tunnel_interop PROPERTIES LABELS "integration" TIMEOUT 300) endif() endif() diff --git a/tests/test_crypto_aead2022_cpp.cpp b/tests/test_crypto_aead2022_cpp.cpp index 39610154..f65e5a1a 100644 --- a/tests/test_crypto_aead2022_cpp.cpp +++ b/tests/test_crypto_aead2022_cpp.cpp @@ -92,12 +92,14 @@ const std::uint8_t kAddr[] = { 0x03, 11, 'e', 'x', 'a', 'm', 'p', 'l', 'e', '.', std::vector cpp_decrypt_all(sc::Aead2022TcpDecryptor &dec, const std::vector &wire) { std::vector got; - std::vector out(sc::k2022MaxChunk); + std::vector out(70000); for (std::size_t off = 0; off < wire.size();) { std::size_t n = std::min(1 + randombytes_uniform(9000), wire.size() - off); std::span in(wire.data() + off, n); for (;;) { - auto step = dec.decrypt(in, out); + // Output windows of any size, down to one byte. + std::size_t room = randombytes_uniform(4) == 0 ? 1 + randombytes_uniform(16) : 1 + randombytes_uniform(70000); + auto step = dec.decrypt(in, std::span(out).first(room)); SS_CHECK(step); got.insert(got.end(), out.begin(), out.begin() + step->produced); in = in.subspan(step->consumed); diff --git a/tests/tunnel_interop.py b/tests/tunnel_interop.py new file mode 100644 index 00000000..78c0b275 --- /dev/null +++ b/tests/tunnel_interop.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +"""ss-tunnel against ss-server (and shadowsocks-rust when present): concurrent TCP +transfers and a UDP round trip through a fixed tunnel target, for every method.""" +import argparse +import base64 +import concurrent.futures +import os +from pathlib import Path +import shutil +import socket +import subprocess +import threading + +from interop import TCPHandler, TCPOrigin, UDPHandler, UDPOrigin, free_port, peers, receive + +METHODS = ["2022-blake3-aes-128-gcm", "2022-blake3-aes-256-gcm", "2022-blake3-chacha20-poly1305", + "aes-128-gcm", "aes-256-gcm", "chacha20-ietf-poly1305"] +LEGACY = ["aes-256-cfb", "aes-128-ctr", "camellia-256-cfb", "chacha20-ietf", "salsa20"] + + +def tcp_case(port, size): + payload = os.urandom(size) + with socket.create_connection(("127.0.0.1", port), timeout=15) as sock, \ + concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: + upload = pool.submit(sock.sendall, payload) + assert receive(sock, size) == payload, "TCP payload mismatch" + upload.result(timeout=15) + + +def udp_case(port): + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.settimeout(.5) + for size in (1, 1200): + payload = os.urandom(size) + # The UDP-only tunnel has no TCP port to wait on: retry until up. + for attempt in range(20): + sock.sendto(payload, ("127.0.0.1", port)) + try: + reply, _ = sock.recvfrom(65535) + break + except TimeoutError: + if attempt == 19: + raise + assert reply == payload, "UDP payload mismatch" + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--bin", required=True, type=Path) + parser.add_argument("--legacy", action="store_true", help="also run the stream ciphers") + parser.add_argument("--fast-open", action="store_true") + args = parser.parse_args() + bins = args.bin.resolve() + server, tunnel = str(bins / "ss-server"), str(bins / "ss-tunnel") + rust_server = shutil.which("ssserver") + methods = METHODS + (LEGACY if args.legacy else []) + failures = 0 + with TCPOrigin(("127.0.0.1", 0), TCPHandler) as tcp, UDPOrigin(("127.0.0.1", 0), UDPHandler) as udp: + for origin in (tcp, udp): + threading.Thread(target=origin.serve_forever, daemon=True).start() + try: + for method in methods: + stream = method in LEGACY + for peer in ("C server",) + (("Rust server",) if rust_server and not stream else ()): + key = base64.b64encode(os.urandom(16 if "aes-128" in method else 32)).decode() + secret = ["-k", "password"] if stream else ["-k", key] + sport, tport, uport = free_port(), free_port(), free_port() + if peer == "C server": + srv = [server, "-s", "127.0.0.1", "-p", str(sport), "-m", method, "-u"] + secret + else: + srv = [rust_server, "-s", f"127.0.0.1:{sport}", "-m", method, "-U"] + secret + extra = ["--fast-open"] if args.fast_open else [] + tcp_tunnel = [tunnel, "-s", "127.0.0.1", "-p", str(sport), "-l", str(tport), "-m", method, + "-L", f"127.0.0.1:{tcp.server_address[1]}"] + secret + extra + udp_tunnel = [tunnel, "-s", "127.0.0.1", "-p", str(sport), "-l", str(uport), "-m", method, + "-U", "-L", f"127.0.0.1:{udp.server_address[1]}"] + secret + label = f"{method}: tunnel -> {peer}" + try: + with peers([srv, tcp_tunnel], [sport, tport]): + with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool: + cases = [pool.submit(tcp_case, tport, size) for size in (1, 4096, 65536, 1048576)] + for case in cases: + case.result(timeout=60) + udp_proc = subprocess.Popen(udp_tunnel, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL) + try: + udp_case(uport) + finally: + udp_proc.terminate() + udp_proc.wait(timeout=5) + print("PASS " + label, flush=True) + except (OSError, AssertionError, TimeoutError) as error: + failures += 1 + print(f"FAIL {label}: {error!r}", flush=True) + finally: + tcp.shutdown() + udp.shutdown() + return int(failures != 0) + + +if __name__ == "__main__": + raise SystemExit(main())