Preserve SOCKS stream data and make control startup cancellable

This commit is contained in:
evgeny committed 2026-10-25 23:25:16 +01:00
1 parent 6660f59b3f
commit 36900780f2
4 files changed
+257 -77

No files matched your search

+136 -66
View File
@@ -13,6 +13,11 @@
#define SOCKS_UDP_RECONNECT_MIN_TB 10000 // 1 с в 0.1 мс
#define SOCKS_UDP_RECONNECT_MAX_TB 300000 // 30 с
#define SOCKS_UDP_MAX_PAYLOAD 2048
#ifdef _WIN32
#define SOCKS_IN_PROGRESS WSAEINPROGRESS
#else
#define SOCKS_IN_PROGRESS EINPROGRESS
#endif
// ── Общий control-канал: TCP к прокси + greeting + auth ───────────────
// Обе фазы (dial и udp) используют эту базу; после greeting+auth вызывается on_ready(c),
@@ -27,6 +32,7 @@ struct socks_ctrl {
uint8_t done; // dial: guard против повторного finish
uint8_t rbuf[SOCKS_CTRL_RBUF_SIZE];
size_t rbuf_len;
size_t rbuf_need; // читаем только текущую control-рамку, не забирая данные готового TCP-потока
uint8_t sbuf[512];
size_t sbuf_len, sbuf_off;
void (*on_ready)(struct socks_ctrl *c); // greeting+auth OK → фаза 2
@@ -46,10 +52,14 @@ static void socks_ctrl_timeout_cb(void *arg);
static void socks_ctrl_flush(struct socks_ctrl *c) {
while (c->sbuf_off < c->sbuf_len) {
ssize_t sent = send(c->sock, c->sbuf + c->sbuf_off, c->sbuf_len - c->sbuf_off, 0);
if (sent < 0) {
int err = socket_get_error();
if (err == ERR_AGAIN || err == ERR_WOULDBLOCK) { uasync_set_socket_write(c->ua, c->socket_id, 1); return; }
ssize_t sent = socket_send(c->sock, c->sbuf + c->sbuf_off, c->sbuf_len - c->sbuf_off);
if (sent <= 0) {
int err = sent < 0 ? socket_get_error() : EPIPE;
if (err == ERR_INTR) continue;
if (err == ERR_AGAIN || err == ERR_WOULDBLOCK) {
if (uasync_set_socket_write(c->ua, c->socket_id, 1) == ERR_OK) return;
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: cannot enable write callback"); err = EIO;
}
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: send failed err=%d", err);
c->on_error(c, err);
return;
@@ -58,10 +68,16 @@ static void socks_ctrl_flush(struct socks_ctrl *c) {
}
c->sbuf_len = 0;
c->sbuf_off = 0;
uasync_set_socket_write(c->ua, c->socket_id, 0);
if (uasync_set_socket_write(c->ua, c->socket_id, 0) != ERR_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: cannot disable write callback"); c->on_error(c, EIO);
}
}
static void socks_ctrl_send(struct socks_ctrl *c, const uint8_t *data, size_t len) {
if (len > sizeof(c->sbuf) || c->sbuf_off < c->sbuf_len) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: invalid control write len=%zu pending=%zu", len, c->sbuf_len - c->sbuf_off);
c->on_error(c, ENOBUFS); return;
}
memcpy(c->sbuf, data, len);
c->sbuf_len = len;
c->sbuf_off = 0;
@@ -71,11 +87,14 @@ static void socks_ctrl_send(struct socks_ctrl *c, const uint8_t *data, size_t le
static void socks_ctrl_read_cb(socket_t sock, void *arg) {
struct socks_ctrl *c = (struct socks_ctrl *)arg;
(void)sock;
if (c->rbuf_len >= sizeof(c->rbuf)) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: rbuf overflow"); c->on_error(c, ENOBUFS); return; }
ssize_t n = recv(c->sock, c->rbuf + c->rbuf_len, sizeof(c->rbuf) - c->rbuf_len, 0);
if (c->rbuf_need > sizeof(c->rbuf) || c->rbuf_need <= c->rbuf_len) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: invalid read state have=%zu need=%zu", c->rbuf_len, c->rbuf_need);
c->on_error(c, ENOBUFS); return;
}
ssize_t n = recv(c->sock, c->rbuf + c->rbuf_len, c->rbuf_need - c->rbuf_len, 0);
if (n < 0) {
int err = socket_get_error();
if (err == ERR_AGAIN || err == ERR_WOULDBLOCK) return;
if (err == ERR_AGAIN || err == ERR_WOULDBLOCK || err == ERR_INTR) return;
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: recv failed err=%d", err);
c->on_error(c, err);
return;
@@ -100,7 +119,7 @@ static void socks_ctrl_process(struct socks_ctrl *c) {
if (method == 0x00) { c->on_ready(c); return; }
if (method == 0x02 && c->have_auth) {
size_t ulen = strlen(c->cfg.user), plen = strlen(c->cfg.pass);
if (ulen > 255 || plen > 255) { c->on_error(c, EINVAL); return; }
if (ulen > 255 || plen > 255) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: invalid credential lengths"); c->on_error(c, EINVAL); return; }
uint8_t ab[2 + 255 + 1 + 255];
size_t n = 0;
ab[n++] = 0x01; ab[n++] = (uint8_t)ulen;
@@ -108,6 +127,7 @@ static void socks_ctrl_process(struct socks_ctrl *c) {
ab[n++] = (uint8_t)plen;
memcpy(ab + n, c->cfg.pass, plen); n += plen;
c->process = socks_ctrl_auth_process;
c->rbuf_need = 2;
socks_ctrl_send(c, ab, n);
return;
}
@@ -131,6 +151,7 @@ static void socks_ctrl_after_connect(struct socks_ctrl *c) {
if (c->have_auth) { g[1] = 0x02; g[2] = 0x00; g[3] = 0x02; n = 4; }
else { g[1] = 0x01; g[2] = 0x00; n = 3; }
c->process = socks_ctrl_process;
c->rbuf_need = 2;
DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "socks: connected to proxy %s:%u, greeting (auth=%d)", c->cfg.host, c->cfg.port, c->have_auth);
socks_ctrl_send(c, g, n);
}
@@ -163,40 +184,44 @@ static void socks_ctrl_timeout_cb(void *arg) {
static int socks_ctrl_start(struct socks_ctrl *c, int timeout_tb) {
char port_str[16];
snprintf(port_str, sizeof(port_str), "%u", c->cfg.port);
struct addrinfo hints;
memset(&hints, 0, sizeof(hints));
hints.ai_family = AF_UNSPEC;
hints.ai_socktype = SOCK_STREAM;
struct addrinfo *res = NULL;
if (getaddrinfo(c->cfg.host, port_str, &hints, &res) != 0 || !res) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: getaddrinfo(%s) failed", c->cfg.host);
struct addrinfo hints = {0}, *res = NULL;
hints.ai_family = AF_UNSPEC; hints.ai_socktype = SOCK_STREAM;
int result = getaddrinfo(c->cfg.host, port_str, &hints, &res);
if (result || !res) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: proxy resolve failed host=%s code=%d", c->cfg.host, result);
if (res) freeaddrinfo(res);
return -1;
}
struct sockaddr_storage pa;
struct sockaddr_storage pa = {0};
socklen_t palen = (socklen_t)res->ai_addrlen;
memcpy(&pa, res->ai_addr, res->ai_addrlen);
freeaddrinfo(res);
c->sock = socket(pa.ss_family, SOCK_STREAM, 0);
if (c->sock == SOCKET_INVALID) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: socket() failed err=%d", socket_get_error()); return -1; }
socket_set_nonblocking(c->sock);
int cr = connect(c->sock, (struct sockaddr *)&pa, palen);
if (cr < 0) {
int err = socket_get_error();
if (err != EINPROGRESS && err != ERR_WOULDBLOCK) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: connect to %s:%u failed err=%d", c->cfg.host, c->cfg.port, err);
socket_close_wrapper(c->sock); c->sock = SOCKET_INVALID;
return -1;
}
c->socket_id = uasync_add_socket_t(c->ua, c->sock, NULL, socks_ctrl_connect_wait_cb, NULL, "socks_ctrl_connect", c);
} else {
c->socket_id = uasync_add_socket_t(c->ua, c->sock, socks_ctrl_read_cb, socks_ctrl_write_cb, NULL, "socks_ctrl", c);
if (c->socket_id) socks_ctrl_after_connect(c);
if ((res->ai_family != AF_INET && res->ai_family != AF_INET6) || res->ai_addrlen > sizeof(pa)) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: unsupported proxy address family=%d len=%zu", res->ai_family, (size_t)res->ai_addrlen);
freeaddrinfo(res); return -1;
}
if (!c->socket_id) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: add_socket failed"); socket_close_wrapper(c->sock); c->sock = SOCKET_INVALID; return -1; }
memcpy(&pa, res->ai_addr, res->ai_addrlen); freeaddrinfo(res);
c->sock = socket(pa.ss_family, SOCK_STREAM, 0);
if (c->sock == SOCKET_INVALID) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: socket creation failed err=%d", socket_get_error()); return -1;
}
if (socket_set_nonblocking(c->sock) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: cannot enable nonblocking err=%d", socket_get_error()); goto fail;
}
result = connect(c->sock, (struct sockaddr*)&pa, palen);
int error = result < 0 ? socket_get_error() : 0;
if (result < 0 && error != SOCKS_IN_PROGRESS && error != ERR_WOULDBLOCK) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: connect to %s:%u failed err=%d", c->cfg.host, c->cfg.port, error); goto fail;
}
/* Даже готовый connect продолжится после возврата конструктора: send error может освободить dial. */
c->socket_id = uasync_add_socket_t(c->ua, c->sock, NULL, socks_ctrl_connect_wait_cb, NULL, "socks_ctrl_connect", c);
if (!c->socket_id) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: connect registration failed"); goto fail; }
c->rbuf_need = 2;
c->timer = uasync_set_timeout(c->ua, timeout_tb > 0 ? timeout_tb : SOCKS_CTRL_TIMEOUT_TB, c, socks_ctrl_timeout_cb, "socks_ctrl");
if (!c->timer) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: handshake timer allocation failed"); goto fail; }
return 0;
fail:
if (c->socket_id) { uasync_remove_socket_t(c->ua, c->sock); c->socket_id = NULL; }
socket_close_wrapper(c->sock); c->sock = SOCKET_INVALID;
return -1;
}
// ── SOCKS5 CONNECT (dial) ─────────────────────────────────────────────
@@ -240,12 +265,13 @@ static void dial_on_ready(struct socks_ctrl *c) {
buf[n++] = 0x04; memcpy(buf + n, &a6, 16); n += 16;
} else {
size_t hl = strlen(d->target_host);
if (hl > 255) { c->on_error(c, EINVAL); return; }
if (hl > 255) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks: target hostname too long"); c->on_error(c, EINVAL); return; }
buf[n++] = 0x03; buf[n++] = (uint8_t)hl; memcpy(buf + n, d->target_host, hl); n += hl;
}
uint16_t pb = htons(d->target_port);
memcpy(buf + n, &pb, 2); n += 2;
c->process = dial_reply_process;
c->rbuf_need = 4;
DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "socks_dial: CONNECT %s:%u", d->target_host, d->target_port);
socks_ctrl_send(c, buf, n);
}
@@ -253,14 +279,20 @@ static void dial_on_ready(struct socks_ctrl *c) {
static void dial_reply_process(struct socks_ctrl *c) {
if (c->rbuf_len < 4) return;
uint8_t ver = c->rbuf[0], rep = c->rbuf[1], atyp = c->rbuf[3];
if (ver != 5 || c->rbuf[2] || rep) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_dial: invalid/rejected reply ver=%u rsv=%u rep=%u", ver, c->rbuf[2], rep);
c->on_error(c, rep ? ECONNREFUSED : EPROTO); return;
}
size_t need = 4;
if (atyp == 0x01) need += 4 + 2;
else if (atyp == 0x04) need += 16 + 2;
else if (atyp == 0x03) { if (c->rbuf_len < 5) return; need += 1 + (size_t)c->rbuf[4] + 2; }
else if (atyp == 0x03) {
if (c->rbuf_len < 5) { c->rbuf_need = 5; return; }
if (!c->rbuf[4]) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_dial: empty BND domain"); c->on_error(c, EPROTO); return; }
need += 1 + (size_t)c->rbuf[4] + 2;
}
else { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_dial: bad reply atyp=%d", atyp); c->on_error(c, EPROTO); return; }
if (c->rbuf_len < need) return;
if (ver != 0x05) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_dial: bad reply ver=%d", ver); c->on_error(c, EPROTO); return; }
if (rep != 0x00) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_dial: CONNECT rejected rep=%d", rep); c->on_error(c, ECONNREFUSED); return; }
if (c->rbuf_len < need) { c->rbuf_need = need; return; }
struct socks_dial *d = (struct socks_dial *)c;
DEBUG_INFO(DEBUG_CATEGORY_PROXY, "socks_dial: CONNECT %s:%u established via %s:%u",
d->target_host, d->target_port, c->cfg.host, c->cfg.port);
@@ -274,8 +306,11 @@ static void dial_on_error(struct socks_ctrl *c, int err) {
void *socks_dial(struct UASYNC *ua, const struct socks_cfg *cfg,
const char *target_host, uint16_t target_port,
socks_dial_cb cb, void *arg, int timeout_tb) {
if (!ua || !cfg || !cfg->host[0] || !cfg->port || !target_host || !target_host[0] || !cb)
return NULL;
if (!ua || !cfg || !cfg->host[0] || !cfg->port || !target_host || !target_host[0] || !cb || !target_port ||
!memchr(cfg->host, 0, sizeof(cfg->host)) || !memchr(cfg->user, 0, sizeof(cfg->user)) ||
!memchr(cfg->pass, 0, sizeof(cfg->pass)) || strnlen(target_host, 256) > 255) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_dial: invalid arguments/config/target length"); return NULL;
}
struct socks_dial *d = u_calloc(1, sizeof(*d));
if (!d) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_dial: alloc failed"); return NULL; }
d->ctrl.ua = ua;
@@ -324,8 +359,8 @@ static void udp_reply_process(struct socks_ctrl *c);
static void udp_idle_process(struct socks_ctrl *c);
static void udp_reconnect_cb(void *arg);
static void udp_schedule_reconnect(struct socks_udp *s) {
if (s->closing) return;
static int udp_schedule_reconnect(struct socks_udp *s) {
if (s->closing) return -1;
s->ready = 0;
if (s->ctrl.socket_id) { uasync_remove_socket_t(s->ctrl.ua, s->ctrl.sock); s->ctrl.socket_id = NULL; }
if (s->ctrl.timer) { uasync_cancel_timeout(s->ctrl.ua, s->ctrl.timer); s->ctrl.timer = NULL; }
@@ -335,18 +370,22 @@ static void udp_schedule_reconnect(struct socks_udp *s) {
s->ctrl.sbuf_len = 0;
s->ctrl.sbuf_off = 0;
s->ctrl.process = socks_ctrl_process;
s->ctrl.rbuf_need = 2;
if (s->reconnect_delay_tb == 0) s->reconnect_delay_tb = SOCKS_UDP_RECONNECT_MIN_TB;
DEBUG_WARN(DEBUG_CATEGORY_PROXY, "socks_udp: reconnect in %u ms", s->reconnect_delay_tb / 10);
s->reconnect_timer = uasync_set_timeout(s->ctrl.ua, (int)s->reconnect_delay_tb, s, udp_reconnect_cb, "socks_udp_rc");
if (!s->reconnect_timer) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: reconnect timer allocation failed"); return -1; }
s->reconnect_delay_tb *= 2;
if (s->reconnect_delay_tb > SOCKS_UDP_RECONNECT_MAX_TB) s->reconnect_delay_tb = SOCKS_UDP_RECONNECT_MAX_TB;
return 0;
}
static void udp_reconnect_cb(void *arg) {
struct socks_udp *s = (struct socks_udp *)arg;
s->reconnect_timer = NULL;
if (s->closing) return;
if (socks_ctrl_start(&s->ctrl, 0) < 0) udp_schedule_reconnect(s);
if (socks_ctrl_start(&s->ctrl, 0) < 0 && udp_schedule_reconnect(s) < 0 && s->ready_cb)
s->ready_cb(s, ENOMEM, s->ready_arg);
}
static void udp_on_ready(struct socks_ctrl *c) {
@@ -355,6 +394,7 @@ static void udp_on_ready(struct socks_ctrl *c) {
buf[3] = 0x01; // ATYP=IPv4, DST=0.0.0.0:0
memset(buf + 4, 0, 6);
c->process = udp_reply_process;
c->rbuf_need = 4;
DEBUG_DEBUG(DEBUG_CATEGORY_PROXY, "socks_udp: sending UDP ASSOCIATE");
socks_ctrl_send(c, buf, 10);
}
@@ -363,14 +403,16 @@ static void udp_reply_process(struct socks_ctrl *c) {
struct socks_udp *s = (struct socks_udp *)c;
if (c->rbuf_len < 4) return;
uint8_t ver = c->rbuf[0], rep = c->rbuf[1], atyp = c->rbuf[3];
if (ver != 5 || c->rbuf[2] || rep) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: invalid/rejected reply ver=%u rsv=%u rep=%u", ver, c->rbuf[2], rep);
c->on_error(c, rep ? ECONNREFUSED : EPROTO); return;
}
size_t need = 4;
if (atyp == 0x01) need += 4 + 2;
else if (atyp == 0x04) need += 16 + 2;
else if (atyp == 0x03) { if (c->rbuf_len < 5) return; need += 1 + (size_t)c->rbuf[4] + 2; }
else if (atyp == 0x03) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: domain relay unsupported"); c->on_error(c, EAFNOSUPPORT); return; }
else { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: bad reply atyp=%d", atyp); c->on_error(c, EPROTO); return; }
if (c->rbuf_len < need) return;
if (ver != 0x05) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: bad reply ver=%d", ver); c->on_error(c, EPROTO); return; }
if (rep != 0x00) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: ASSOCIATE rejected rep=%d", rep); c->on_error(c, ECONNREFUSED); return; }
if (c->rbuf_len < need) { c->rbuf_need = need; return; }
memset(&s->relay, 0, sizeof(s->relay));
if (atyp == 0x01) {
@@ -385,27 +427,49 @@ static void udp_reply_process(struct socks_ctrl *c) {
memcpy(&sa6->sin6_port, c->rbuf + 20, 2);
} else { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: unsupported relay family"); c->on_error(c, EAFNOSUPPORT); return; }
uint16_t relay_port = ss_get_port(&s->relay);
if (!relay_port) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: relay has zero port"); c->on_error(c, EPROTO); return; }
int wildcard = atyp == 1 ? !((struct sockaddr_in*)&s->relay)->sin_addr.s_addr :
IN6_IS_ADDR_UNSPECIFIED(&((struct sockaddr_in6*)&s->relay)->sin6_addr);
if (wildcard) {
struct sockaddr_storage peer = {0}; socklen_t size = sizeof(peer);
if (getpeername(c->sock, (struct sockaddr*)&peer, &size) != 0 || peer.ss_family != s->relay.ss_family) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: cannot resolve wildcard relay from proxy peer");
c->on_error(c, EAFNOSUPPORT); return;
}
s->relay = peer;
if (atyp == 1) ((struct sockaddr_in*)&s->relay)->sin_port = htons(relay_port);
else ((struct sockaddr_in6*)&s->relay)->sin6_port = htons(relay_port);
}
memmove(c->rbuf, c->rbuf + need, c->rbuf_len - need); c->rbuf_len -= need;
s->ready = 1;
s->reconnect_delay_tb = 0;
if (c->timer) { uasync_cancel_timeout(c->ua, c->timer); c->timer = NULL; }
c->process = udp_idle_process;
c->rbuf_need = sizeof(c->rbuf);
DEBUG_INFO(DEBUG_CATEGORY_PROXY, "socks_udp: ASSOCIATE ready, relay=%s", sockaddr_storage_to_str(&s->relay).str);
if (s->ready_cb) s->ready_cb(s, 0, s->ready_arg);
}
static void udp_idle_process(struct socks_ctrl *c) {
DEBUG_WARN(DEBUG_CATEGORY_PROXY, "socks_udp: unexpected control data after READY bytes=%zu", c->rbuf_len);
c->rbuf_len = 0; // неожиданные данные на control-канале после READY — игнорируем
}
static void udp_on_error(struct socks_ctrl *c, int err) {
DEBUG_WARN(DEBUG_CATEGORY_PROXY, "socks_udp: control error err=%d → reconnect", err);
udp_schedule_reconnect((struct socks_udp *)c);
struct socks_udp* s = (struct socks_udp*)c;
if (udp_schedule_reconnect(s) < 0 && s->ready_cb) s->ready_cb(s, ENOMEM, s->ready_arg);
}
struct socks_udp *socks_udp_associate(struct UASYNC *ua, const struct socks_cfg *cfg,
socket_t local_udp_fd, socks_udp_ready_cb cb, void *arg) {
if (!ua || !cfg || !cfg->host[0] || !cfg->port || local_udp_fd == SOCKET_INVALID) return NULL;
if (!ua || !cfg || !cfg->host[0] || !cfg->port || local_udp_fd == SOCKET_INVALID ||
!memchr(cfg->host, 0, sizeof(cfg->host)) || !memchr(cfg->user, 0, sizeof(cfg->user)) ||
!memchr(cfg->pass, 0, sizeof(cfg->pass))) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: invalid arguments/config"); return NULL;
}
struct socks_udp *s = u_calloc(1, sizeof(*s));
if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: alloc failed"); return NULL; }
s->ctrl.ua = ua;
@@ -418,7 +482,7 @@ struct socks_udp *socks_udp_associate(struct UASYNC *ua, const struct socks_cfg
s->local_udp_fd = local_udp_fd;
s->ready_cb = cb;
s->ready_arg = arg;
if (socks_ctrl_start(&s->ctrl, 0) < 0) udp_schedule_reconnect(s);
if (socks_ctrl_start(&s->ctrl, 0) < 0 && udp_schedule_reconnect(s) < 0) { socks_udp_destroy(s); return NULL; }
return s;
}
@@ -434,8 +498,11 @@ void socks_udp_destroy(struct socks_udp *s) {
ssize_t socks_udp_sendto(struct socks_udp *s, const uint8_t *data, size_t len,
const struct sockaddr_storage *dst) {
if (!s || !data || !dst || !s->ready) return -1;
if (dst->ss_family != AF_INET && dst->ss_family != AF_INET6) return -1;
if (!s || (!data && len) || !dst || (dst->ss_family != AF_INET && dst->ss_family != AF_INET6)) {
DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: invalid send arguments"); return -1;
}
if (!s->ready) { DEBUG_TRACE(DEBUG_CATEGORY_PROXY, "socks_udp: send deferred until relay ready"); return -1; }
if (len > SOCKS_UDP_MAX_PAYLOAD) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: payload too large %zu", len); return -1; }
uint8_t buf[SOCKS_UDP_MAX_PAYLOAD + 32];
size_t off = 0;
buf[off++] = 0; buf[off++] = 0; // RSV
@@ -451,26 +518,31 @@ ssize_t socks_udp_sendto(struct socks_udp *s, const uint8_t *data, size_t len,
memcpy(buf + off, &sa6->sin6_addr, 16); off += 16;
memcpy(buf + off, &sa6->sin6_port, 2); off += 2;
}
if (off + len > sizeof(buf)) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: payload too large %zu", len); return -1; }
memcpy(buf + off, data, len); off += len;
if (len) memcpy(buf + off, data, len);
off += len;
socklen_t rl = (s->relay.ss_family == AF_INET6) ? sizeof(struct sockaddr_in6) : sizeof(struct sockaddr_in);
ssize_t sent = socket_sendto(s->local_udp_fd, buf, off, (struct sockaddr *)&s->relay, rl);
if (sent < 0) DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: sendto relay failed err=%d", socket_get_error());
return sent;
if (sent < 0) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: sendto relay failed err=%d", socket_get_error()); return -1; }
if ((size_t)sent != off) { DEBUG_ERROR(DEBUG_CATEGORY_PROXY, "socks_udp: partial datagram sent=%zd expected=%zu", sent, off); return -1; }
return (ssize_t)len;
}
ssize_t socks_udp_unwrap(const struct socks_udp *s, const uint8_t *raw, size_t raw_len,
uint8_t *payload, size_t cap, struct sockaddr_storage *src) {
(void)s;
if (!raw || raw_len < 10 || !src) return -1;
if (!raw || raw_len < 10 || !src) { DEBUG_WARN(DEBUG_CATEGORY_PROXY, "socks_udp: missing/short input len=%zu", raw_len); return -1; }
if (raw[0] != 0 || raw[1] != 0 || raw[2] != 0) { DEBUG_WARN(DEBUG_CATEGORY_PROXY, "socks_udp: bad RSV/FRAG"); return -1; }
uint8_t atyp = raw[3];
size_t hlen;
if (atyp == 0x01) hlen = 4 + 4 + 2;
else if (atyp == 0x04) hlen = 4 + 16 + 2;
else if (atyp == 0x03) { if (raw_len < 5) return -1; hlen = 4 + 1 + (size_t)raw[4] + 2; }
else if (atyp == 0x03) { DEBUG_WARN(DEBUG_CATEGORY_PROXY, "socks_udp: domain source unsupported"); return -1; }
else { DEBUG_WARN(DEBUG_CATEGORY_PROXY, "socks_udp: bad atyp=%d", atyp); return -1; }
if (raw_len < hlen) return -1;
if (raw_len < hlen) { DEBUG_WARN(DEBUG_CATEGORY_PROXY, "socks_udp: short address len=%zu required=%zu", raw_len, hlen); return -1; }
size_t plen = raw_len - hlen;
if (plen > cap || (plen && !payload)) {
DEBUG_WARN(DEBUG_CATEGORY_PROXY, "socks_udp: insufficient output capacity=%zu payload=%zu", cap, plen); return -1;
}
memset(src, 0, sizeof(*src));
if (atyp == 0x01) {
@@ -486,8 +558,6 @@ ssize_t socks_udp_unwrap(const struct socks_udp *s, const uint8_t *raw, size_t r
} else {
return -1; // domain — для ETCP-пиров не встречается
}
size_t plen = raw_len - hlen;
if (plen > cap) return -1;
if (plen > 0) memcpy(payload, raw + hlen, plen);
if (plen > 0) memmove(payload, raw + hlen, plen);
return (ssize_t)plen;
}
+25 -7
View File
@@ -4,7 +4,8 @@
// - SOCKS5 CONNECT (TCP/STCP-линки) — socks_dial();
// - SOCKS5 UDP ASSOCIATE (UDP-линки) — socks_udp_associate() и далее.
//
// Модуль не знает про ETCP — только транспорт.
// Модуль не знает про ETCP — только транспорт. API/callbacks — uasync-поток.
// cfg копируется, cb/arg заимствованы. Диагностика — категория proxy, пароли в лог не попадают.
#ifndef SOCKS_CLIENT_H
#define SOCKS_CLIENT_H
@@ -17,7 +18,8 @@ extern "C" {
struct UASYNC;
// Конфигурация прокси. host — ip или домен; user/pass пустые = no-auth (метод 0x00).
// Все строки NUL-терминированы, максимум 63 байта. host — IP/домен, port ненулевой.
// Пустой user предлагает только no-auth; непустой предлагает no-auth и username/password.
struct socks_cfg {
char host[64];
uint16_t port;
@@ -29,12 +31,18 @@ struct socks_cfg {
// Коллбэк завершения подключения:
// sock — готовый сокет (на нём можно делать STCP-хендшейк), err == 0;
// sock == SOCKET_INVALID при ошибке, err — код ошибки (errno).
// sock == SOCKET_INVALID при ошибке, err — errno/WSA-код или протокольная ошибка.
// Состояние dial освобождено до callback, handle больше недействителен; отменять его из callback нельзя.
typedef void (*socks_dial_cb)(socket_t sock, int err, void *arg);
// Асинхронно подключиться к прокси и выполнить SOCKS5 CONNECT к target_host:target_port.
// target_host может быть числовым IP или доменом (домен резолвит сам прокси).
// Возвращает handle (NULL при немедленной ошибке — тогда cb НЕ вызывается, ошибка залогирована).
// target_host — 1..255 байт, target_port ненулевой; длина не обрезается.
// Разрешение host самого прокси синхронно через getaddrinfo, используется первый IPv4/IPv6 адрес;
// connect/handshake и callback — после возврата. timeout_tb в 0.1 мс, <=0 означает 10 секунд.
// При успехе удаляет свою uasync-регистрацию и передаёт nonblocking socket владельцу.
// Первые байты TCP-потока после SOCKS-ответа остаются в сокете, control-буфер их не поглощает.
void *socks_dial(struct UASYNC *ua, const struct socks_cfg *cfg,
const char *target_host, uint16_t target_port,
socks_dial_cb cb, void *arg, int timeout_tb);
@@ -46,23 +54,33 @@ void socks_dial_cancel(void *h);
struct socks_udp;
// Коллбэк готовности ассоциации: err==0 → relay получен, можно слать через socks_udp_sendto.
// err==0: relay готов (повторяется после reconnect), можно слать через socks_udp_sendto.
// ENOMEM: после запуска не удалось поставить retry, handle остаётся unready до destroy/recreate.
// В callback допускается socks_udp_destroy(s); после этого s/arg больше не используются модулем.
typedef void (*socks_udp_ready_cb)(struct socks_udp *s, int err, void *arg);
// Открыть UDP ASSOCIATE к прокси (persistent TCP-control + relay). local_udp_fd — уже
// забинденный локальный UDP-сокет (reserved port), через него шлём на relay.
// При обрыве control-канала сессия автоматически пересоздаёт ассоциацию (backoff).
// Возвращает сессию (NULL только при ошибке аллокации/аргументов).
// Возвращает сессию; NULL при неверных аргументах/ресурсах без callback.
// Начальная ошибка connect может вернуть сессию с ожидающим reconnect; 10с на попытку,
// backoff 1/2/4/.../30с. При отказе обоих начальных таймеров возвращает NULL.
// local_udp_fd заимствован до destroy, callback необязателен. Релей только IPv4/IPv6;
// wildcard BND.ADDR заменяется IP TCP-прокси с сохранением BND.PORT, который должен быть ненулевым.
struct socks_udp *socks_udp_associate(struct UASYNC *ua, const struct socks_cfg *cfg,
socket_t local_udp_fd, socks_udp_ready_cb cb, void *arg);
// Отправить датаграмму пиру dst: обёртка в SOCKS-UDP заголовок + sendto(relay).
// -1 пока ассоциация не готова (или ошибка).
// len<=2048, data может быть NULL при len=0. Успех — len без SOCKS-заголовка, включая 0;
// -1 пока ассоциация не готова или при ошибке. Удалённая доставка не гарантируется.
ssize_t socks_udp_sendto(struct socks_udp *s, const uint8_t *data, size_t len,
const struct sockaddr_storage *dst);
// Снять SOCKS-UDP заголовок с входящей датаграммы: вернуть payload и реальный src пира.
// Возвращает длину payload (<=0 при битом/неподходящем пакете).
// s не используется, может быть NULL; raw/src обязательны. Поддержаны IPv4/IPv6, FRAG=0.
// Возвращает длину payload (0 допустим), -1 при отказе; при ошибке src/payload не меняются.
// payload может быть NULL для пустых данных, иначе cap должен вмещать их целиком.
// Допускается снятие заголовка в том же буфере; src должен быть отдельным от raw/payload.
ssize_t socks_udp_unwrap(const struct socks_udp *s, const uint8_t *raw, size_t raw_len,
uint8_t *payload, size_t cap, struct sockaddr_storage *src);
+1 -1
View File
@@ -463,7 +463,7 @@ test_stcp_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
test_socks_client_SOURCES = test_socks_client.c $(test_clock_sources)
test_socks_client_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/transport_layer -I$(top_srcdir)/lib
test_socks_client_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
test_socks_client_LDFLAGS = $(test_clock_ldflags)
test_socks_client_LDFLAGS = $(test_clock_ldflags) -Wl,--wrap=uasync_set_timeout -Wl,--wrap=connect -Wl,--wrap=socket_send
test_etcp_two_instances_SOURCES = test_etcp_two_instances.c
test_etcp_two_instances_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
+95 -3
View File
@@ -16,7 +16,25 @@ static int test_failed = 0;
if (!(cond)) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, " FAIL: %s", #cond); test_failed = 1; return test_failed; } \
} while(0)
enum { MOCK_NOAUTH = 0, MOCK_AUTH = 1, MOCK_REJECT = 2 };
enum { MOCK_NOAUTH = 0, MOCK_AUTH = 1, MOCK_REJECT = 2, MOCK_BANNER = 3, MOCK_BAD_RSV = 4 };
static int fail_control_timer, fail_reconnect_timer, force_immediate_connect, zero_send;
extern void* __real_uasync_set_timeout(struct UASYNC*, int, void*, timeout_callback_t, const char*);
void* __wrap_uasync_set_timeout(struct UASYNC* ua, int delay, void* arg, timeout_callback_t callback, const char* name) {
if (fail_control_timer && !strcmp(name, "socks_ctrl")) { fail_control_timer = 0; return NULL; }
if (fail_reconnect_timer && !strcmp(name, "socks_udp_rc")) { fail_reconnect_timer = 0; return NULL; }
return __real_uasync_set_timeout(ua, delay, arg, callback, name);
}
extern int __real_connect(socket_t, const struct sockaddr*, socklen_t);
int __wrap_connect(socket_t fd, const struct sockaddr* address, socklen_t size) {
int result = __real_connect(fd, address, size);
if (force_immediate_connect && result < 0 && (socket_get_error() == EINPROGRESS || socket_get_error() == ERR_WOULDBLOCK)) return 0;
return result;
}
extern ssize_t __real_socket_send(socket_t, const void*, size_t);
ssize_t __wrap_socket_send(socket_t fd, const void* data, size_t size) {
if (zero_send) { zero_send = 0; return 0; }
return __real_socket_send(fd, data, size);
}
struct mock {
struct UASYNC *ua;
@@ -68,7 +86,11 @@ static void mock_process(struct mock *m) {
else return;
if (m->buf_len < need) return;
if (m->mode == MOCK_REJECT) { uint8_t r[] = {0x05, 0x05, 0x00, 0x01, 0,0,0,0, 0,0}; mock_send(m, r, 10); }
else { uint8_t r[] = {0x05, 0x00, 0x00, 0x01, 0,0,0,0, 0,0}; mock_send(m, r, 10); m->got_connect = 1; }
else {
uint8_t r[] = {0x05, 0x00, 0x00, 0x01, 0,0,0,0, 0,0, 'h','e','l','l','o'};
if (m->mode == MOCK_BAD_RSV) r[2] = 1;
mock_send(m, r, m->mode == MOCK_BANNER ? sizeof(r) : 10); m->got_connect = 1;
}
m->state = 3;
memmove(m->buf, m->buf + need, m->buf_len - need); m->buf_len -= need;
continue;
@@ -411,7 +433,8 @@ static int test4_udp_associate(void) {
p->sin_family = AF_INET; p->sin_addr.s_addr = htonl(INADDR_LOOPBACK); p->sin_port = htons(9999);
ssize_t snd = socks_udp_sendto(su, (const uint8_t *)"hello", 5, &peer);
TASSERT(snd > 0);
TASSERT(snd == 5);
TASSERT(socks_udp_sendto(su, (const uint8_t*)"x", SIZE_MAX, &peer) == -1);
uint8_t raw[2048];
struct sockaddr_storage from; socklen_t fl = sizeof(from);
@@ -431,6 +454,11 @@ static int test4_udp_associate(void) {
TASSERT(memcmp(payload, "hello", 5) == 0);
TASSERT(src.ss_family == AF_INET);
TASSERT(((struct sockaddr_in *)&src)->sin_port == p->sin_port);
struct sockaddr_storage unchanged = src;
TASSERT(socks_udp_unwrap(NULL, raw, (size_t)got, NULL, 256, &src) == -1);
TASSERT(memcmp(&src, &unchanged, sizeof(src)) == 0);
TASSERT(socks_udp_unwrap(NULL, raw, (size_t)got, payload, 4, &src) == -1);
TASSERT(memcmp(&src, &unchanged, sizeof(src)) == 0);
socks_udp_destroy(su);
socket_close_wrapper(lfd);
@@ -502,6 +530,67 @@ static int test5_udp_reassociate(void) {
return 0;
}
static int test6_banner_and_reply_validation(void) {
for (int bad = 0; bad <= 1; bad++) {
struct UASYNC* ua = uasync_create(); TASSERT(ua);
struct mock m; TASSERT(mock_start(&m, ua, 0, bad ? MOCK_BAD_RSV : MOCK_BANNER) == 0);
struct socks_cfg cfg = {0}; strcpy(cfg.host, "127.0.0.1"); cfg.port = test_bound_port(m.listen_sock);
int error = run_dial(ua, &cfg, "127.0.0.1", 22);
if (bad) TASSERT(error == EPROTO && g_sock == SOCKET_INVALID);
else {
TASSERT(error == 0);
uint8_t banner[5]; ssize_t n = recv(g_sock, banner, sizeof(banner), 0);
TASSERT(n == 5 && !memcmp(banner, "hello", 5));
socket_close_wrapper(g_sock);
}
mock_stop(&m); uasync_destroy(ua, 1);
}
return 0;
}
static int test7_start_failures_and_deferred_callback(void) {
struct UASYNC* ua = uasync_create(); TASSERT(ua);
struct mock m; TASSERT(mock_start(&m, ua, 0, MOCK_NOAUTH) == 0);
struct socks_cfg cfg = {0}; strcpy(cfg.host, "127.0.0.1"); cfg.port = test_bound_port(m.listen_sock);
g_done = 0; fail_control_timer = 1;
TASSERT(!socks_dial(ua, &cfg, "127.0.0.1", 22, dial_cb, NULL, 0) && !g_done);
mock_stop(&m); uasync_poll(ua, 0);
TASSERT(mock_start(&m, ua, 0, MOCK_NOAUTH) == 0); cfg.port = test_bound_port(m.listen_sock);
force_immediate_connect = 1; zero_send = 1;
void* handle = socks_dial(ua, &cfg, "127.0.0.1", 22, dial_cb, NULL, 0);
TASSERT(handle && !g_done);
for (int i = 0; !g_done && i < 20000; i++) test_clock_poll(ua, 10);
TASSERT(g_done && g_err == EPIPE && g_sock == SOCKET_INVALID && !zero_send);
force_immediate_connect = 0;
mock_stop(&m); uasync_destroy(ua, 1);
ua = uasync_create(); TASSERT(ua);
socket_t fd = socket(AF_INET, SOCK_DGRAM, 0); TASSERT(fd != SOCKET_INVALID);
g_su_ready = 0; fail_control_timer = fail_reconnect_timer = 1;
TASSERT(!socks_udp_associate(ua, &cfg, fd, su_ready_cb, NULL));
TASSERT(!g_su_ready && !fail_control_timer && !fail_reconnect_timer);
socket_close_wrapper(fd); uasync_destroy(ua, 1);
return 0;
}
static void destroy_ready_cb(struct socks_udp* s, int err, void* arg) {
(void)arg; g_su_err = err; g_su_ready++; socks_udp_destroy(s);
}
static int test8_destroy_in_udp_ready(void) {
struct UASYNC* ua = uasync_create(); TASSERT(ua);
struct udp_mock m; TASSERT(umock_start(&m, ua, 0) == 0);
struct socks_cfg cfg = {0}; strcpy(cfg.host, "127.0.0.1"); cfg.port = test_bound_port(m.listen_sock);
socket_t fd = socket(AF_INET, SOCK_DGRAM, 0); TASSERT(fd != SOCKET_INVALID);
g_su_ready = 0; g_su_err = 0;
TASSERT(socks_udp_associate(ua, &cfg, fd, destroy_ready_cb, NULL));
for (int i = 0; !g_su_ready && i < 20000; i++) test_clock_poll(ua, 10);
TASSERT(g_su_ready == 1 && !g_su_err);
struct sockaddr_storage address; socklen_t size = sizeof(address);
TASSERT(getsockname(fd, (struct sockaddr*)&address, &size) == 0);
socket_close_wrapper(fd); umock_stop(&m); uasync_destroy(ua, 1);
return 0;
}
int main(void) {
debug_config_init();
debug_set_level(DEBUG_LEVEL_ERROR);
@@ -514,6 +603,9 @@ int main(void) {
{"test3_reject", test3_reject},
{"test4_udp_associate", test4_udp_associate},
{"test5_udp_reassociate", test5_udp_reassociate},
{"test6_banner_and_reply_validation", test6_banner_and_reply_validation},
{"test7_start_failures_and_deferred_callback", test7_start_failures_and_deferred_callback},
{"test8_destroy_in_udp_ready", test8_destroy_in_udp_ready},
};
int n = (int)(sizeof(tests) / sizeof(tests[0]));
int passed = 0;