commit b2879caf47a23c8f8e0311094666235d93a3fc1e
| author | 斟酌 鵬兄 <tgckpg@gmail.com> |
|---|
| date | 2026-05-28T21:19:10Z |
|---|
| subject | Break stream_route.c into multiple files |
|---|
commit b2879caf47a23c8f8e0311094666235d93a3fc1e
Author: 斟酌 鵬兄 <tgckpg@gmail.com>
Date: 2026-05-28T21:19:10Z
Break stream_route.c into multiple files
---
devtools/cflow.sh | 4 +-
src/compat_socket.h | 1 +
src/datagram_route.c | 274 ++++++++++++++----
src/datagram_route.h | 4 +-
src/route.h | 3 +
src/stream_builtin.c | 128 +++++++++
src/stream_builtin.h | 8 +
src/stream_conn.c | 356 ++++++++++++++++++++++++
src/stream_conn.h | 12 +
src/stream_file.c | 128 +++++++++
src/stream_file.h | 8 +
src/stream_listener.c | 161 +++++++++++
src/stream_listener.h | 9 +
src/stream_route.c | 750 ++++----------------------------------------------
src/stream_route.h | 24 ++
15 files changed, 1114 insertions(+), 756 deletions(-)
diff --git a/devtools/cflow.sh b/devtools/cflow.sh
index 5c28e76..a465d5f 100755
--- a/devtools/cflow.sh
+++ b/devtools/cflow.sh
@@ -60,5 +60,5 @@ IGNORE_RE="$(printf '%s\n' "$IGNORE" \
| sed '/^[[:space:]]*$/d' \
| paste -sd '|' -)"
-cflow --brief --main="$MAIN" src/*.c |
- grep -Ev "^[[:space:]]*(${IGNORE_RE})\\("
+cflow --brief -n --main="$MAIN" src/*.c |
+ grep -Ev "^[[:space:]]+[[:digit:]]+[[:space:]]*(${IGNORE_RE})\\("
diff --git a/src/compat_socket.h b/src/compat_socket.h
index 4c39529..abea3a4 100644
--- a/src/compat_socket.h
+++ b/src/compat_socket.h
@@ -12,6 +12,7 @@
#else
+#include <errno.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
diff --git a/src/datagram_route.c b/src/datagram_route.c
index 47b2aea..d715fc9 100644
--- a/src/datagram_route.c
+++ b/src/datagram_route.c
@@ -23,7 +23,7 @@
struct datagram_route_ctx;
-struct udp_client {
+struct datagram_client {
struct datagram_route_ctx *ctx;
evutil_socket_t fd;
@@ -36,7 +36,7 @@ struct udp_client {
time_t last_seen;
- struct udp_client *next;
+ struct datagram_client *next;
};
static int sockaddr_equal(
@@ -56,7 +56,7 @@ static int sockaddr_equal(
return memcmp(a, b, (size_t)a_len) == 0;
}
-static void free_udp_client(struct udp_client *c)
+static void free_datagram_client(struct datagram_client *c)
{
if (c == NULL) {
return;
@@ -77,13 +77,13 @@ static void free_udp_client(struct udp_client *c)
free(c);
}
-static void cleanup_idle_udp_clients(struct datagram_route_ctx *ctx)
+static void cleanup_idle_datagram_clients(struct datagram_route_ctx *ctx)
{
time_t now = time(NULL);
- struct udp_client **pp = &ctx->clients;
+ struct datagram_client **pp = &ctx->clients;
while (*pp != NULL) {
- struct udp_client *c = *pp;
+ struct datagram_client *c = *pp;
if (now - c->last_seen <= ctx->route->opts.idle_timeout_sec) {
pp = &c->next;
@@ -98,16 +98,16 @@ static void cleanup_idle_udp_clients(struct datagram_route_ctx *ctx)
"client_len", _LOGV(c->client_addr_len)
);
- free_udp_client(c);
+ free_datagram_client(c);
}
}
-static struct udp_client *find_udp_client(
+static struct datagram_client *find_datagram_client(
struct datagram_route_ctx *ctx,
const struct sockaddr_storage *addr,
socklen_t addr_len
) {
- for (struct udp_client *c = ctx->clients; c != NULL; c = c->next) {
+ for (struct datagram_client *c = ctx->clients; c != NULL; c = c->next) {
if (sockaddr_equal(&c->client_addr, c->client_addr_len, addr, addr_len)) {
return c;
}
@@ -120,7 +120,7 @@ static void upstream_read_cb(evutil_socket_t fd, short events, void *arg)
{
(void)events;
- struct udp_client *c = arg;
+ struct datagram_client *c = arg;
struct datagram_route_ctx *ctx = c->ctx;
unsigned char buf[UDP_MAX_PACKET];
@@ -164,8 +164,8 @@ static void upstream_read_cb(evutil_socket_t fd, short events, void *arg)
}
}
-static const char *udp_client_addr_string(
- const struct udp_client *c,
+static const char *datagram_client_addr_string(
+ const struct datagram_client *c,
char *buf,
size_t buf_len
) {
@@ -219,7 +219,7 @@ static const char *udp_client_addr_string(
return buf;
}
-static int start_udp_builtin(struct udp_client *c)
+static int start_datagram_builtin(struct datagram_client *c)
{
struct x_builtin_request req;
struct x_builtin_response res;
@@ -233,7 +233,7 @@ static int start_udp_builtin(struct udp_client *c)
memset(&req, 0, sizeof(req));
req.builtin = c->ctx->route->upstream.builtin;
- req.client_addr = udp_client_addr_string(c, client_addr, sizeof(client_addr));
+ req.client_addr = datagram_client_addr_string(c, client_addr, sizeof(client_addr));
rc = x_builtin_handle(&req, &res);
if (rc < 0) {
@@ -269,7 +269,7 @@ static int start_udp_builtin(struct udp_client *c)
}
}
-static int connect_udp_upstream(struct udp_client *c, const struct endpoint *upstream)
+static int connect_datagram_upstream(struct datagram_client *c, const struct endpoint *upstream)
{
if (upstream == NULL) {
return -EINVAL;
@@ -380,14 +380,14 @@ static int connect_udp_upstream(struct udp_client *c, const struct endpoint *ups
}
}
-static struct udp_client *create_udp_client(
+static struct datagram_client *create_datagram_client(
struct datagram_route_ctx *ctx,
const struct sockaddr_storage *client_addr,
socklen_t client_addr_len
) {
const struct route *r = ctx->route;
- struct udp_client *c = calloc(1, sizeof(*c));
+ struct datagram_client *c = calloc(1, sizeof(*c));
if (c == NULL) {
return NULL;
}
@@ -398,32 +398,32 @@ static struct udp_client *create_udp_client(
c->client_addr_len = client_addr_len;
c->last_seen = time(NULL);
- int rc = connect_udp_upstream(c, &r->upstream);
+ int rc = connect_datagram_upstream(c, &r->upstream);
if (rc < 0) {
LOG_ERROR("udp upstream connect failed",
"upstream", _LOGV_ENDPOINT(&r->upstream),
"err", _LOGV(evutil_socket_error_to_string(-rc))
);
- free_udp_client(c);
+ free_datagram_client(c);
return NULL;
}
if (evutil_make_socket_nonblocking(c->fd) < 0) {
LOG_ERROR("evutil_make_socket_nonblocking failed");
- free_udp_client(c);
+ free_datagram_client(c);
return NULL;
}
c->ev = event_new(ctx->base, c->fd, EV_READ | EV_PERSIST, upstream_read_cb, c);
if (c->ev == NULL) {
LOG_ERROR("event_new failed for udp upstream");
- free_udp_client(c);
+ free_datagram_client(c);
return NULL;
}
if (event_add(c->ev, NULL) < 0) {
LOG_ERROR("event_add failed for udp upstream");
- free_udp_client(c);
+ free_datagram_client(c);
return NULL;
}
@@ -438,8 +438,8 @@ static struct udp_client *create_udp_client(
return c;
}
-static int send_udp_payload_to_upstream(
- struct udp_client *c,
+static int send_datagram_payload_to_upstream(
+ struct datagram_client *c,
const unsigned char *payload,
size_t payload_len
) {
@@ -504,7 +504,7 @@ static void listen_read_cb(evutil_socket_t fd, short events, void *arg)
struct datagram_route_ctx *ctx = arg;
- cleanup_idle_udp_clients(ctx);
+ cleanup_idle_datagram_clients(ctx);
for (;;) {
unsigned char buf[UDP_MAX_PACKET];
@@ -541,7 +541,7 @@ static void listen_read_cb(evutil_socket_t fd, short events, void *arg)
}
if (ctx->route->upstream.kind == ENDPOINT_BUILTIN) {
- struct udp_client tmp;
+ struct datagram_client tmp;
memset(&tmp, 0, sizeof(tmp));
tmp.ctx = ctx;
@@ -550,7 +550,7 @@ static void listen_read_cb(evutil_socket_t fd, short events, void *arg)
tmp.client_addr_len = client_addr_len;
tmp.last_seen = time(NULL);
- int rc = start_udp_builtin(&tmp);
+ int rc = start_datagram_builtin(&tmp);
if (rc < 0) {
LOG_ERROR("udp builtin failed",
"err", _LOGV(evutil_socket_error_to_string(-rc))
@@ -561,9 +561,9 @@ static void listen_read_cb(evutil_socket_t fd, short events, void *arg)
continue;
}
- struct udp_client *c = find_udp_client(ctx, &client_addr, client_addr_len);
+ struct datagram_client *c = find_datagram_client(ctx, &client_addr, client_addr_len);
if (c == NULL) {
- c = create_udp_client(ctx, &client_addr, client_addr_len);
+ c = create_datagram_client(ctx, &client_addr, client_addr_len);
if (c == NULL) {
return;
}
@@ -571,7 +571,7 @@ static void listen_read_cb(evutil_socket_t fd, short events, void *arg)
c->last_seen = time(NULL);
- int rc = send_udp_payload_to_upstream(c, buf, (size_t)n);
+ int rc = send_datagram_payload_to_upstream(c, buf, (size_t)n);
if (rc < 0) {
LOG_ERROR("udp send to upstream failed",
"err", _LOGV(evutil_socket_error_to_string(-rc))
@@ -585,15 +585,6 @@ static int prepare_datagram_route(struct datagram_route_ctx *ctx)
{
const struct route *r = ctx->route;
- if (r->upstream.kind != ENDPOINT_INET &&
- r->upstream.kind != ENDPOINT_BUILTIN &&
- r->upstream.kind != ENDPOINT_UNIX_DGRAM) {
- LOG_ERROR("unsupported datagram upstream endpoint",
- "upstream", _LOGV_ENDPOINT(&r->upstream)
- );
- return -ENOTSUP;
- }
-
if (r->upstream.kind == ENDPOINT_UNIX_DGRAM) {
const char *runtime_dir = tinyproxy_runtime_dir();
@@ -609,6 +600,124 @@ static int prepare_datagram_route(struct datagram_route_ctx *ctx)
return 0;
}
+#ifndef _WIN32
+static int bind_unix_datagram_listener(struct datagram_route_ctx *ctx)
+{
+ const struct route *r = ctx->route;
+ const char *path = r->listen.path;
+ struct sockaddr_un listen_addr;
+
+ if (path[0] == '\0') {
+ LOG_ERROR("empty unix datagram listen path",
+ "line", _LOGV(r->line_no),
+ "listen", _LOGV_ENDPOINT(&r->listen));
+ return -EINVAL;
+ }
+
+ if (strlen(path) >= sizeof(listen_addr.sun_path)) {
+ LOG_ERROR("unix datagram listen path too long",
+ "line", _LOGV(r->line_no),
+ "path", _LOGV(path));
+ return -ENAMETOOLONG;
+ }
+
+ memset(&listen_addr, 0, sizeof(listen_addr));
+ listen_addr.sun_family = AF_UNIX;
+ memcpy(listen_addr.sun_path, path, strlen(path) + 1);
+
+ if (unlink(path) < 0 && errno != ENOENT) {
+ LOG_ERROR("failed to remove existing unix datagram listener socket",
+ "line", _LOGV(r->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(strerror(errno)));
+ return -errno;
+ }
+
+ ctx->listen_fd = socket(AF_UNIX, SOCK_DGRAM, 0);
+ if (ctx->listen_fd < 0) {
+ int err = errno;
+
+ LOG_ERROR("unix datagram listen socket failed",
+ "line", _LOGV(r->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(strerror(err)));
+ return -err;
+ }
+
+ evutil_make_socket_closeonexec(ctx->listen_fd);
+
+ if (evutil_make_socket_nonblocking(ctx->listen_fd) < 0) {
+ int err = errno;
+
+ LOG_ERROR("evutil_make_socket_nonblocking failed for unix datagram listener",
+ "line", _LOGV(r->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(strerror(err)));
+ return err ? -err : -EINVAL;
+ }
+
+ if (bind(
+ ctx->listen_fd,
+ (const struct sockaddr *)&listen_addr,
+ sizeof(listen_addr)
+ ) < 0) {
+ int err = errno;
+
+ LOG_ERROR("unix datagram bind failed",
+ "line", _LOGV(r->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(strerror(err)));
+ return -err;
+ }
+
+ ctx->local_addr_len = sizeof(ctx->local_addr);
+ memset(&ctx->local_addr, 0, sizeof(ctx->local_addr));
+
+ if (getsockname(
+ ctx->listen_fd,
+ (struct sockaddr *)&ctx->local_addr,
+ &ctx->local_addr_len
+ ) < 0) {
+ int err = errno;
+
+ LOG_ERROR("unix datagram getsockname failed",
+ "line", _LOGV(r->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(strerror(err)));
+ return err ? -err : -EINVAL;
+ }
+
+ ctx->listen_ev = event_new(
+ ctx->base,
+ ctx->listen_fd,
+ EV_READ | EV_PERSIST,
+ listen_read_cb,
+ ctx
+ );
+ if (ctx->listen_ev == NULL) {
+ LOG_ERROR("event_new failed for unix datagram listener",
+ "line", _LOGV(r->line_no),
+ "path", _LOGV(path));
+ return -ENOMEM;
+ }
+
+ if (event_add(ctx->listen_ev, NULL) < 0) {
+ LOG_ERROR("event_add failed for unix datagram listener",
+ "line", _LOGV(r->line_no),
+ "path", _LOGV(path));
+ return -EINVAL;
+ }
+
+ return 0;
+}
+#else
+static int bind_unix_datagram_listener(struct datagram_route_ctx *ctx)
+{
+ (void)ctx;
+ return -ENOTSUP;
+}
+#endif
+
static int bind_udp_datagram_listener(struct datagram_route_ctx *ctx)
{
const struct route *r = ctx->route;
@@ -701,6 +810,23 @@ static int bind_udp_datagram_listener(struct datagram_route_ctx *ctx)
return 0;
}
+static int bind_datagram_listener(struct datagram_route_ctx *ctx)
+{
+ switch (ctx->route->listen.kind) {
+ case ENDPOINT_INET:
+ return bind_udp_datagram_listener(ctx);
+
+ case ENDPOINT_UNIX_DGRAM:
+ return bind_unix_datagram_listener(ctx);
+
+ default:
+ LOG_ERROR("datagram listen protocol not implemented yet",
+ "line", _LOGV(ctx->route->line_no),
+ "listen", _LOGV_ENDPOINT(&ctx->route->listen));
+ return -ENOTSUP;
+ }
+}
+
int start_datagram_route(
struct worker *w,
const struct route *r,
@@ -726,19 +852,7 @@ int start_datagram_route(
return rc;
}
- switch (r->listen.kind) {
- case ENDPOINT_INET:
- rc = bind_udp_datagram_listener(ctx);
- break;
-
- default:
- LOG_ERROR("unsupported datagram listen endpoint",
- "listen", _LOGV_ENDPOINT(&r->listen)
- );
- rc = -ENOTSUP;
- break;
- }
-
+ rc = bind_datagram_listener(ctx);
if (rc != 0) {
stop_datagram_route(ctx);
return rc;
@@ -756,6 +870,54 @@ int start_datagram_route(
return 0;
}
+#ifndef _WIN32
+static void stop_unix_datagram_route(struct datagram_route_ctx *ctx)
+{
+ const struct route *r;
+ const char *path;
+ struct stat st;
+
+ if (ctx == NULL || ctx->route == NULL) {
+ return;
+ }
+
+ r = ctx->route;
+
+ if (r->listen.kind != ENDPOINT_UNIX_DGRAM) {
+ return;
+ }
+
+ path = r->listen.path;
+ if (path[0] == '\0') {
+ return;
+ }
+
+ if (lstat(path, &st) < 0) {
+ if (errno != ENOENT) {
+ LOG_WARN("failed to inspect unix datagram listener socket",
+ "line", _LOGV(r->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(strerror(errno)));
+ }
+ return;
+ }
+
+ if (!S_ISSOCK(st.st_mode)) {
+ LOG_WARN("not removing unix datagram listener path because it is not a socket",
+ "line", _LOGV(r->line_no),
+ "path", _LOGV(path));
+ return;
+ }
+
+ if (unlink(path) < 0) {
+ LOG_WARN("failed to remove unix datagram listener socket",
+ "line", _LOGV(r->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(strerror(errno)));
+ }
+}
+#endif
+
void stop_datagram_route(struct datagram_route_ctx *ctx)
{
if (ctx == NULL) {
@@ -763,10 +925,10 @@ void stop_datagram_route(struct datagram_route_ctx *ctx)
}
while (ctx->clients != NULL) {
- struct udp_client *c = ctx->clients;
+ struct datagram_client *c = ctx->clients;
ctx->clients = c->next;
c->next = NULL;
- free_udp_client(c);
+ free_datagram_client(c);
}
if (ctx->listen_ev != NULL) {
@@ -774,6 +936,10 @@ void stop_datagram_route(struct datagram_route_ctx *ctx)
ctx->listen_ev = NULL;
}
+#ifndef _WIN32
+ stop_unix_datagram_route(ctx);
+#endif
+
if (ctx->listen_fd >= 0) {
evutil_closesocket(ctx->listen_fd);
ctx->listen_fd = -1;
diff --git a/src/datagram_route.h b/src/datagram_route.h
index b35d0d7..97ba6da 100644
--- a/src/datagram_route.h
+++ b/src/datagram_route.h
@@ -8,7 +8,7 @@ struct event;
struct event_base;
struct worker;
struct route;
-struct udp_client;
+struct datagram_client;
struct datagram_route_ctx {
struct event_base *base;
@@ -21,7 +21,7 @@ struct datagram_route_ctx {
struct sockaddr_storage local_addr;
socklen_t local_addr_len;
- struct udp_client *clients;
+ struct datagram_client *clients;
};
int start_datagram_route(struct worker *w, const struct route *r,
diff --git a/src/route.h b/src/route.h
index a8e7e99..76b9d0f 100644
--- a/src/route.h
+++ b/src/route.h
@@ -11,6 +11,9 @@
#define ROUTE_DEFAULT_IDLE_TIMEOUT_SEC 60
#define ROUTE_DEFAULT_CONNECT_TIMEOUT_SEC 5
+#define BEV_READ_HIGH_WATER (256 * 1024)
+#define BEV_WRITE_RESUME_WATER (128 * 1024)
+
struct route_options {
bool proxy_v2;
bool keep_alive;
diff --git a/src/stream_builtin.c b/src/stream_builtin.c
new file mode 100644
index 0000000..c1511dd
--- /dev/null
+++ b/src/stream_builtin.c
@@ -0,0 +1,128 @@
+#include <event2/buffer.h>
+
+#include <string.h>
+
+#include "stream_builtin.h"
+#include "stream_conn.h"
+
+static void builtin_client_read_cb(struct bufferevent *bev, void *arg)
+{
+ struct evbuffer *input = bufferevent_get_input(bev);
+
+ (void)arg;
+
+ evbuffer_drain(input, evbuffer_get_length(input));
+}
+
+static void builtin_close_after_write_cb(struct bufferevent *bev, void *arg)
+{
+ conn_t *conn = arg;
+
+ if (evbuffer_get_length(bufferevent_get_output(bev)) == 0) {
+ free_conn(conn);
+ }
+}
+
+static const char *stream_client_addr_string(conn_t *conn, char *buf, size_t buf_len)
+{
+ struct sockaddr_storage ss;
+ socklen_t len = sizeof(ss);
+ evutil_socket_t fd;
+ int port = 0;
+
+ if (conn == NULL || conn->client == NULL || buf == NULL || buf_len == 0) {
+ return NULL;
+ }
+
+ fd = bufferevent_getfd(conn->client);
+ if (fd < 0) {
+ return NULL;
+ }
+
+ if (getpeername(fd, (struct sockaddr *)&ss, &len) < 0) {
+ return NULL;
+ }
+
+ if (ss.ss_family == AF_INET) {
+ struct sockaddr_in *sin = (struct sockaddr_in *)&ss;
+
+ if (inet_ntop(AF_INET, &sin->sin_addr, buf, buf_len) == NULL) {
+ return NULL;
+ }
+
+ port = ntohs(sin->sin_port);
+ } else if (ss.ss_family == AF_INET6) {
+ struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)&ss;
+
+ if (inet_ntop(AF_INET6, &sin6->sin6_addr, buf, buf_len) == NULL) {
+ return NULL;
+ }
+
+ port = ntohs(sin6->sin6_port);
+ } else {
+ snprintf(buf, buf_len, "unknown");
+ return buf;
+ }
+
+ snprintf(buf + strlen(buf), buf_len - strlen(buf), ":%d", port);
+ return buf;
+}
+
+int start_stream_builtin(conn_t *conn)
+{
+ struct x_builtin_request req;
+ struct x_builtin_response res;
+ const struct route *r;
+ char client_addr[128];
+ int rc;
+
+ if (conn == NULL || conn->route == NULL) {
+ return -EINVAL;
+ }
+
+ r = conn->route;
+
+ memset(&req, 0, sizeof(req));
+
+ req.builtin = r->upstream.builtin;
+ req.client_addr = stream_client_addr_string(conn, client_addr, sizeof(client_addr));
+
+ rc = x_builtin_handle(&req, &res);
+ if (rc < 0) {
+ free_conn(conn);
+ return rc;
+ }
+
+ if (conn->upstream != NULL) {
+ bufferevent_free(conn->upstream);
+ conn->upstream = NULL;
+ }
+
+ set_client_idle_timeout(conn, r);
+
+ switch (res.action) {
+ case X_BUILTIN_ACTION_CLOSE:
+ if (res.data_len > 0) {
+ bufferevent_write(conn->client, res.data, res.data_len);
+ bufferevent_setcb(conn->client, NULL, builtin_close_after_write_cb, event_cb, conn);
+ bufferevent_enable(conn->client, EV_WRITE);
+ } else {
+ free_conn(conn);
+ }
+ return 0;
+
+ case X_BUILTIN_ACTION_DISCARD:
+ bufferevent_setcb(conn->client, builtin_client_read_cb, NULL, event_cb, conn);
+ bufferevent_enable(conn->client, EV_READ);
+ return 0;
+
+ case X_BUILTIN_ACTION_HANG:
+ bufferevent_setcb(conn->client, NULL, NULL, event_cb, conn);
+ bufferevent_enable(conn->client, EV_READ);
+ return 0;
+
+ default:
+ free_conn(conn);
+ return -EINVAL;
+ }
+}
diff --git a/src/stream_builtin.h b/src/stream_builtin.h
new file mode 100644
index 0000000..9ad7a72
--- /dev/null
+++ b/src/stream_builtin.h
@@ -0,0 +1,8 @@
+#ifndef STREAM_BUILTIN_H
+#define STREAM_BUILTIN_H
+
+#include "route_runtime.h"
+
+int start_stream_builtin(conn_t *conn);
+
+#endif
diff --git a/src/stream_conn.c b/src/stream_conn.c
new file mode 100644
index 0000000..2beb4fd
--- /dev/null
+++ b/src/stream_conn.c
@@ -0,0 +1,356 @@
+#include <event2/buffer.h>
+#include <event2/util.h>
+
+#include <stdlib.h>
+#include <string.h>
+
+#include "klog.h"
+#include "worker.h"
+#include "route.h"
+#include "stream_conn.h"
+#include "stream_file.h"
+#include "stream_builtin.h"
+#include "proxy_proto_v2.h"
+
+void free_conn(conn_t *conn) {
+ if (conn == NULL) {
+ return;
+ }
+
+ if (conn->client != NULL) {
+ bufferevent_free(conn->client);
+ }
+
+ if (conn->upstream != NULL) {
+ bufferevent_free(conn->upstream);
+ }
+
+ free(conn);
+}
+
+static void set_connect_timeout(conn_t *conn, const struct route *r)
+{
+ struct timeval connect_timeout = {
+ .tv_sec = r->opts.connect_timeout_sec,
+ .tv_usec = 0,
+ };
+
+ bufferevent_set_timeouts(conn->upstream, NULL, &connect_timeout);
+}
+
+static void set_idle_timeouts(conn_t *conn, const struct route *r)
+{
+ struct timeval idle_timeout = {
+ .tv_sec = r->opts.idle_timeout_sec,
+ .tv_usec = 0,
+ };
+
+ bufferevent_set_timeouts(conn->client, &idle_timeout, &idle_timeout);
+ bufferevent_set_timeouts(conn->upstream, &idle_timeout, &idle_timeout);
+}
+
+void set_client_idle_timeout(conn_t *conn, const struct route *r)
+{
+ struct timeval idle_timeout = {
+ .tv_sec = r->opts.idle_timeout_sec,
+ .tv_usec = 0,
+ };
+
+ bufferevent_set_timeouts(conn->client, &idle_timeout, &idle_timeout);
+}
+
+static int set_socket_keepalive(evutil_socket_t fd, const struct route *r)
+{
+ int v = r->opts.keep_alive ? 1 : 0;
+
+ if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, (const char *)&v, sizeof(v)) < 0) {
+ return -errno;
+ }
+
+ return 0;
+}
+
+void event_cb(struct bufferevent *bev, short events, void *arg) {
+ conn_t *conn = arg;
+
+ if (events & BEV_EVENT_CONNECTED) {
+ set_idle_timeouts(conn, conn->route);
+
+ bufferevent_enable(conn->client, EV_READ | EV_WRITE);
+ bufferevent_enable(conn->upstream, EV_READ | EV_WRITE);
+ return;
+ }
+
+ if (events & BEV_EVENT_TIMEOUT) {
+ LOG_WARN("connection timed out");
+ free_conn(conn);
+ return;
+ }
+
+ if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
+ if (events & BEV_EVENT_ERROR) {
+ int err = EVUTIL_SOCKET_ERROR();
+ LOG_ERROR("connection error",
+ "err", _LOGV(evutil_socket_error_to_string(err))
+ );
+ }
+
+ free_conn(conn);
+ }
+
+ (void)bev;
+}
+
+static int connect_upstream(struct bufferevent *bev, const struct endpoint *ep)
+{
+ if (ep == NULL) {
+ return -EINVAL;
+ }
+
+ switch (ep->kind) {
+ case ENDPOINT_INET: {
+ struct sockaddr_in addr;
+
+ memset(&addr, 0, sizeof(addr));
+ addr.sin_family = AF_INET;
+ addr.sin_port = htons(ep->port);
+
+ if (inet_pton(AF_INET, ep->host, &addr.sin_addr) != 1) {
+ return -EINVAL;
+ }
+
+ if (bufferevent_socket_connect(
+ bev,
+ (struct sockaddr *)&addr,
+ sizeof(addr)) < 0) {
+ return -errno;
+ }
+
+ return 0;
+ }
+
+ case ENDPOINT_UNIX:
+#ifdef _WIN32
+ return -ENOTSUP;
+#else
+ {
+ struct sockaddr_un addr;
+
+ memset(&addr, 0, sizeof(addr));
+ addr.sun_family = AF_UNIX;
+
+ if (ep->path[0] == '\0') {
+ return -EINVAL;
+ }
+
+ if (strlen(ep->path) >= sizeof(addr.sun_path)) {
+ return -ENAMETOOLONG;
+ }
+
+ strcpy(addr.sun_path, ep->path);
+
+ if (bufferevent_socket_connect(
+ bev,
+ (struct sockaddr *)&addr,
+ sizeof(addr)) < 0) {
+ return -errno;
+ }
+
+ return 0;
+ }
+#endif
+
+ default:
+ return -ENOTSUP;
+ }
+}
+
+static void pipe_read_cb(struct bufferevent *src, void *arg)
+{
+ conn_t *conn = arg;
+ struct bufferevent *dst;
+
+ if (src == conn->client) {
+ dst = conn->upstream;
+ } else if (src == conn->upstream) {
+ dst = conn->client;
+ } else {
+ return;
+ }
+
+ struct evbuffer *input = bufferevent_get_input(src);
+ struct evbuffer *output = bufferevent_get_output(dst);
+
+ evbuffer_add_buffer(output, input);
+
+ if (evbuffer_get_length(output) >= BEV_READ_HIGH_WATER) {
+ bufferevent_disable(src, EV_READ);
+ }
+}
+
+static void pipe_write_cb(struct bufferevent *dst, void *arg)
+{
+ conn_t *conn = arg;
+ struct bufferevent *src;
+
+ if (dst == conn->client) {
+ src = conn->upstream;
+ } else if (dst == conn->upstream) {
+ src = conn->client;
+ } else {
+ return;
+ }
+
+ struct evbuffer *output = bufferevent_get_output(dst);
+
+ if (evbuffer_get_length(output) < BEV_WRITE_RESUME_WATER) {
+ bufferevent_enable(src, EV_READ);
+ }
+}
+
+
+static void worker_adopt_client_fd(struct worker *w, struct accepted_client *ac) {
+ conn_t *conn = calloc(1, sizeof(*conn));
+ if (conn == NULL) {
+ evutil_closesocket(ac->fd);
+ return;
+ }
+
+ const struct route *r = ac->route;
+
+ conn->owner = w;
+ conn->route = r;
+
+ conn->peer_addr = ac->peer_addr;
+ conn->peer_addr_len = ac->peer_addr_len;
+
+ conn->client = bufferevent_socket_new(w->base, ac->fd, BEV_OPT_CLOSE_ON_FREE);
+ if (conn->client == NULL) {
+ free(conn);
+ evutil_closesocket(ac->fd);
+ return;
+ }
+
+ if (ac->route->upstream.kind == ENDPOINT_BUILTIN) {
+ start_stream_builtin(conn);
+ return;
+ }
+
+ if (ac->route->upstream.kind == ENDPOINT_FILE) {
+ start_stream_file(conn);
+ return;
+ }
+
+ conn->upstream = bufferevent_socket_new(w->base, -1, BEV_OPT_CLOSE_ON_FREE);
+ if (conn->upstream == NULL) {
+ free_conn(conn);
+ return;
+ }
+
+ int rc = set_socket_keepalive(ac->fd, r);
+ if (rc < 0) {
+ LOG_WARN("failed to enable client TCP keepalive",
+ "err", _LOGV(strerror(-rc))
+ );
+ }
+
+ bufferevent_setwatermark(conn->client, EV_READ, 0, BEV_READ_HIGH_WATER);
+ bufferevent_setwatermark(conn->upstream, EV_READ, 0, BEV_READ_HIGH_WATER);
+
+ bufferevent_setcb(conn->client, pipe_read_cb, pipe_write_cb, event_cb, conn);
+ bufferevent_setcb(conn->upstream, pipe_read_cb, pipe_write_cb, event_cb, conn);
+
+ /*
+ * Do not read from the client yet.
+ *
+ * Otherwise client bytes may be copied into the upstream output buffer
+ * before the PROXY v2 header is queued.
+ */
+ bufferevent_disable(conn->client, EV_READ);
+
+ set_connect_timeout(conn, r);
+
+ rc = connect_upstream(conn->upstream, &r->upstream);
+ if (rc < 0) {
+ LOG_ERROR("upstream connect failed",
+ "upstream", _LOGV_ENDPOINT(&r->upstream),
+ "err", _LOGV(strerror(-rc))
+ );
+ free_conn(conn);
+ return;
+ }
+
+ if (r->opts.keep_alive && r->upstream.kind == ENDPOINT_INET) {
+ evutil_socket_t upstream_fd = bufferevent_getfd(conn->upstream);
+
+ if (upstream_fd >= 0) {
+ int rc = set_socket_keepalive(upstream_fd, r);
+ if (rc < 0) {
+ LOG_WARN("failed to enable upstream TCP keepalive",
+ "upstream", _LOGV_ENDPOINT(&r->upstream),
+ "err", _LOGV(strerror(-rc))
+ );
+ }
+ }
+ }
+
+ if (r->opts.proxy_v2) {
+ struct sockaddr_in local_addr;
+ socklen_t local_len = sizeof(local_addr);
+
+ memset(&local_addr, 0, sizeof(local_addr));
+
+ if (getsockname(ac->fd, (struct sockaddr *)&local_addr, &local_len) < 0) {
+ perror("getsockname");
+ free_conn(conn);
+ return;
+ }
+
+ if (ac->peer_addr_len <= 0) {
+ LOG_ERROR("invalid client address");
+ free_conn(conn);
+ return;
+ }
+
+ if (ac->peer_addr.ss_family != AF_INET ||
+ local_addr.sin_family != AF_INET) {
+ LOG_ERROR("PROXY v2 currently only supports IPv4 TCP");
+ free_conn(conn);
+ return;
+ }
+
+ unsigned char hdr[256];
+ size_t hdr_len = 0;
+
+ int rc = proxy_v2_build(
+ hdr,
+ sizeof(hdr),
+ (const struct sockaddr *)&ac->peer_addr,
+ ac->peer_addr_len,
+ (const struct sockaddr *)&local_addr,
+ local_len,
+ SOCK_STREAM,
+ &hdr_len
+ );
+
+ if (rc == 0) {
+ rc = bufferevent_write(conn->upstream, hdr, hdr_len);
+ if (rc < 0) {
+ rc = -EIO;
+ }
+ }
+
+ if (rc < 0) {
+ LOG_ERROR("failed to write PROXY v2 header", "err", _LOGV(strerror(-rc)));
+ free_conn(conn);
+ return;
+ }
+ }
+
+ bufferevent_enable(conn->client, EV_READ | EV_WRITE);
+ bufferevent_enable(conn->upstream, EV_READ | EV_WRITE);
+}
+
+void dispatch_client_fd(struct worker *w, struct accepted_client *ac) {
+ worker_adopt_client_fd(w, ac);
+}
diff --git a/src/stream_conn.h b/src/stream_conn.h
new file mode 100644
index 0000000..827b84a
--- /dev/null
+++ b/src/stream_conn.h
@@ -0,0 +1,12 @@
+#ifndef STREAM_CONN_H
+#define STREAM_CONN_H
+
+#include "stream_route.h"
+
+void dispatch_client_fd(struct worker *w, struct accepted_client *ac);
+void free_conn(conn_t *conn);
+
+void set_client_idle_timeout(conn_t *conn, const struct route *r);
+void event_cb(struct bufferevent *bev, short events, void *arg);
+
+#endif
diff --git a/src/stream_file.c b/src/stream_file.c
new file mode 100644
index 0000000..9c6cac5
--- /dev/null
+++ b/src/stream_file.c
@@ -0,0 +1,128 @@
+#include <event2/buffer.h>
+
+#include <stdlib.h>
+#include <string.h>
+
+#include "klog.h"
+#include "route.h"
+#include "stream_file.h"
+#include "stream_conn.h"
+
+#define FILE_CHUNK_SIZE 4096
+
+struct file_conn {
+ conn_t *conn;
+ FILE *fp;
+};
+
+static void file_done(struct file_conn *fc)
+{
+ if (fc == NULL) {
+ return;
+ }
+
+ if (fc->fp != NULL) {
+ fclose(fc->fp);
+ fc->fp = NULL;
+ }
+
+ free_conn(fc->conn);
+ free(fc);
+}
+
+static void file_event_cb(struct bufferevent *bev, short events, void *arg)
+{
+ struct file_conn *fc = arg;
+
+ (void)bev;
+ (void)events;
+
+ file_done(fc);
+}
+
+static void file_write_cb(struct bufferevent *bev, void *arg)
+{
+ struct file_conn *fc = arg;
+ struct evbuffer *out = bufferevent_get_output(bev);
+ char buf[FILE_CHUNK_SIZE];
+
+ while (evbuffer_get_length(out) < BEV_READ_HIGH_WATER) {
+ size_t n = fread(buf, 1, sizeof(buf), fc->fp);
+
+ if (n > 0) {
+ if (bufferevent_write(bev, buf, n) < 0) {
+ file_done(fc);
+ return;
+ }
+
+ continue;
+ }
+
+ if (ferror(fc->fp)) {
+ file_done(fc);
+ return;
+ }
+
+ /*
+ * EOF. If nothing is queued anymore, close now.
+ * Otherwise keep the write callback installed so we close
+ * after libevent drains the remaining output.
+ */
+ if (evbuffer_get_length(out) == 0) {
+ file_done(fc);
+ }
+
+ return;
+ }
+}
+
+int start_stream_file(conn_t *conn)
+{
+ const struct route *r;
+ struct file_conn *fc;
+ FILE *fp;
+
+ if (conn == NULL || conn->route == NULL) {
+ return -EINVAL;
+ }
+
+ r = conn->route;
+
+ fp = fopen(r->upstream.path, "rb");
+ if (fp == NULL) {
+ int err = errno;
+
+ LOG_ERROR("failed to open file upstream",
+ "path", _LOGV(r->upstream.path),
+ "err", _LOGV(strerror(err))
+ );
+
+ free_conn(conn);
+ return -err;
+ }
+
+ if (conn->upstream != NULL) {
+ bufferevent_free(conn->upstream);
+ conn->upstream = NULL;
+ }
+
+ fc = calloc(1, sizeof(*fc));
+ if (fc == NULL) {
+ fclose(fp);
+ free_conn(conn);
+ return -ENOMEM;
+ }
+
+ fc->conn = conn;
+ fc->fp = fp;
+
+ bufferevent_setwatermark(conn->client, EV_WRITE, 0, BEV_READ_HIGH_WATER);
+ bufferevent_setcb(conn->client, NULL, file_write_cb, file_event_cb, fc);
+ bufferevent_enable(conn->client, EV_WRITE);
+
+ set_client_idle_timeout(conn, r);
+
+ file_write_cb(conn->client, fc);
+
+ return 0;
+}
diff --git a/src/stream_file.h b/src/stream_file.h
new file mode 100644
index 0000000..c26a19f
--- /dev/null
+++ b/src/stream_file.h
@@ -0,0 +1,8 @@
+#ifndef STREAM_FILE_H
+#define STREAM_FILE_H
+
+#include "stream_route.h"
+
+int start_stream_file(conn_t *conn);
+
+#endif
diff --git a/src/stream_listener.c b/src/stream_listener.c
new file mode 100644
index 0000000..00969bf
--- /dev/null
+++ b/src/stream_listener.c
@@ -0,0 +1,161 @@
+#include <event2/listener.h>
+
+#include <string.h>
+
+#include "klog.h"
+#include "route.h"
+#include "compat_socket.h"
+#include "stream_listener.h"
+#include "stream_conn.h"
+
+static void accept_cb(
+ struct evconnlistener *listener,
+ evutil_socket_t client_fd,
+ struct sockaddr *addr,
+ int socklen,
+ void *arg
+) {
+ (void)listener;
+
+ struct stream_route_ctx *ctx = arg;
+ struct accepted_client ac = {
+ .fd = client_fd,
+ .route = ctx->route,
+ };
+
+ if (addr != NULL && socklen > 0 && (size_t)socklen <= sizeof(ac.peer_addr)) {
+ memcpy(&ac.peer_addr, addr, (size_t)socklen);
+ ac.peer_addr_len = (socklen_t)socklen;
+ } else {
+ LOG_ERROR("invalid accepted client address", "socklen", _LOGV(socklen));
+ evutil_closesocket(client_fd);
+ return;
+ }
+
+ dispatch_client_fd(ctx->worker, &ac);
+}
+
+void accept_error_cb(struct evconnlistener *listener, void *arg)
+{
+ struct stream_route_ctx *ctx = arg;
+ int err = EVUTIL_SOCKET_ERROR();
+
+ LOG_ERROR("accept error", "err", _LOGV(evutil_socket_error_to_string(err)));
+
+ evconnlistener_disable(listener);
+ event_base_loopexit(ctx->accept_base, NULL);
+}
+
+static int bind_tcp_stream_listener(struct stream_route_ctx *ctx)
+{
+ const struct route *r = ctx->route;
+ struct sockaddr_in listen_addr;
+
+ memset(&listen_addr, 0, sizeof(listen_addr));
+ listen_addr.sin_family = AF_INET;
+ listen_addr.sin_port = htons(r->listen.port);
+
+ if (inet_pton(AF_INET, r->listen.host, &listen_addr.sin_addr) != 1) {
+ LOG_ERROR("invalid listen address",
+ "listen", _LOGV_ENDPOINT(&r->listen));
+ return -EINVAL;
+ }
+
+ ctx->listener = evconnlistener_new_bind(
+ ctx->accept_base,
+ accept_cb,
+ ctx,
+ LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE,
+ -1,
+ (struct sockaddr *)&listen_addr,
+ sizeof(listen_addr)
+ );
+
+ if (ctx->listener == NULL) {
+ LOG_ERROR("evconnlistener_new_bind failed",
+ "listen", _LOGV_ENDPOINT(&r->listen));
+ return -EADDRINUSE;
+ }
+
+ return 0;
+}
+
+static int bind_unix_stream_listener(struct stream_route_ctx *ctx)
+{
+ const struct endpoint *ep = &ctx->route->listen;
+ const char *path = ep->path;
+ struct sockaddr_un sa;
+ struct evconnlistener *listener;
+
+ if (path[0] == '\0') {
+ LOG_ERROR("empty unix stream listen path",
+ "line", _LOGV(ctx->route->line_no),
+ "listen", _LOGV_ENDPOINT(ep));
+ return -EINVAL;
+ }
+
+ memset(&sa, 0, sizeof(sa));
+ sa.sun_family = AF_UNIX;
+
+ if (strlen(path) >= sizeof(sa.sun_path)) {
+ LOG_ERROR("unix stream listen path too long",
+ "line", _LOGV(ctx->route->line_no),
+ "path", _LOGV(path));
+ return -ENAMETOOLONG;
+ }
+
+ memcpy(sa.sun_path, path, strlen(path) + 1);
+
+ /*
+ * For listener sockets, stale socket files are common after crashes.
+ * This is safe enough for our config-driven proxy: if the path exists
+ * and is not ours, bind() would fail without this anyway.
+ */
+ if (unlink(path) < 0 && errno != ENOENT) {
+ LOG_ERROR("failed to remove existing unix stream socket",
+ "line", _LOGV(ctx->route->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(strerror(errno)));
+ return -errno;
+ }
+
+ listener = evconnlistener_new_bind(
+ ctx->accept_base,
+ accept_cb,
+ ctx,
+ LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE,
+ -1,
+ (struct sockaddr *)&sa,
+ sizeof(sa)
+ );
+ if (listener == NULL) {
+ int err = errno;
+
+ LOG_ERROR("failed to bind unix stream listener",
+ "line", _LOGV(ctx->route->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(strerror(err)));
+
+ return err ? -err : -EIO;
+ }
+
+ ctx->listener = listener;
+ return 0;
+}
+
+int bind_stream_listener(struct stream_route_ctx *ctx)
+{
+ switch (ctx->route->listen.kind) {
+ case ENDPOINT_INET:
+ return bind_tcp_stream_listener(ctx);
+
+ case ENDPOINT_UNIX:
+ return bind_unix_stream_listener(ctx);
+
+ default:
+ LOG_ERROR("stream listen protocol not implemented yet",
+ "line", _LOGV(ctx->route->line_no),
+ "listen", _LOGV_ENDPOINT(&ctx->route->listen));
+ return -ENOTSUP;
+ }
+}
diff --git a/src/stream_listener.h b/src/stream_listener.h
new file mode 100644
index 0000000..dbe4d24
--- /dev/null
+++ b/src/stream_listener.h
@@ -0,0 +1,9 @@
+#ifndef STREAM_LISTENER_H
+#define STREAM_LISTENER_H
+
+#include "stream_route.h"
+
+int bind_stream_listener(struct stream_route_ctx *ctx);
+void accept_error_cb(struct evconnlistener *listener, void *arg);
+
+#endif
diff --git a/src/stream_route.c b/src/stream_route.c
index 6c782a2..87610de 100644
--- a/src/stream_route.c
+++ b/src/stream_route.c
@@ -1,741 +1,91 @@
-#include <event2/bufferevent.h>
-#include <event2/buffer.h>
-#include <event2/event.h>
-#include <event2/listener.h>
-#include <event2/util.h>
-
-#include <errno.h>
-#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "klog.h"
-#include "file_conf.h"
-#include "compat_socket.h"
-#include "proxy_proto_v2.h"
+#include "stream_route.h"
#include "worker.h"
#include "route.h"
-#include "stream_route.h"
-#include "x_builtins.h"
-
-#define BEV_READ_HIGH_WATER (256 * 1024)
-#define BEV_WRITE_RESUME_WATER (128 * 1024)
-
-#define FILE_CHUNK_SIZE 4096
-
-typedef struct conn_s {
- struct worker *owner;
- const struct route *route;
-
- struct bufferevent *client;
- struct bufferevent *upstream;
-
- struct sockaddr_storage peer_addr;
- socklen_t peer_addr_len;
-} conn_t;
-
-struct file_conn {
- conn_t *conn;
- FILE *fp;
-};
-
-struct accepted_client {
- evutil_socket_t fd;
- struct sockaddr_storage peer_addr;
- socklen_t peer_addr_len;
- const struct route *route;
-};
-
-static void free_conn(conn_t *conn) {
- if (conn == NULL) {
- return;
- }
-
- if (conn->client != NULL) {
- bufferevent_free(conn->client);
- }
-
- if (conn->upstream != NULL) {
- bufferevent_free(conn->upstream);
- }
-
- free(conn);
-}
-
-static void pipe_read_cb(struct bufferevent *src, void *arg)
-{
- conn_t *conn = arg;
- struct bufferevent *dst;
-
- if (src == conn->client) {
- dst = conn->upstream;
- } else if (src == conn->upstream) {
- dst = conn->client;
- } else {
- return;
- }
-
- struct evbuffer *input = bufferevent_get_input(src);
- struct evbuffer *output = bufferevent_get_output(dst);
-
- evbuffer_add_buffer(output, input);
-
- if (evbuffer_get_length(output) >= BEV_READ_HIGH_WATER) {
- bufferevent_disable(src, EV_READ);
- }
-}
-
-static void pipe_write_cb(struct bufferevent *dst, void *arg)
-{
- conn_t *conn = arg;
- struct bufferevent *src;
-
- if (dst == conn->client) {
- src = conn->upstream;
- } else if (dst == conn->upstream) {
- src = conn->client;
- } else {
- return;
- }
-
- struct evbuffer *output = bufferevent_get_output(dst);
-
- if (evbuffer_get_length(output) < BEV_WRITE_RESUME_WATER) {
- bufferevent_enable(src, EV_READ);
- }
-}
-
-static void set_connect_timeout(conn_t *conn, const struct route *r)
-{
- struct timeval connect_timeout = {
- .tv_sec = r->opts.connect_timeout_sec,
- .tv_usec = 0,
- };
-
- bufferevent_set_timeouts(conn->upstream, NULL, &connect_timeout);
-}
-
-static void set_idle_timeouts(conn_t *conn, const struct route *r)
-{
- struct timeval idle_timeout = {
- .tv_sec = r->opts.idle_timeout_sec,
- .tv_usec = 0,
- };
-
- bufferevent_set_timeouts(conn->client, &idle_timeout, &idle_timeout);
- bufferevent_set_timeouts(conn->upstream, &idle_timeout, &idle_timeout);
-}
-
-static void set_client_idle_timeout(conn_t *conn, const struct route *r)
-{
- struct timeval idle_timeout = {
- .tv_sec = r->opts.idle_timeout_sec,
- .tv_usec = 0,
- };
-
- bufferevent_set_timeouts(conn->client, &idle_timeout, &idle_timeout);
-}
-
-static int set_socket_keepalive(evutil_socket_t fd, const struct route *r)
-{
- int v = r->opts.keep_alive ? 1 : 0;
-
- if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, (const char *)&v, sizeof(v)) < 0) {
- return -errno;
- }
-
- return 0;
-}
-
-static void event_cb(struct bufferevent *bev, short events, void *arg) {
- conn_t *conn = arg;
-
- if (events & BEV_EVENT_CONNECTED) {
- set_idle_timeouts(conn, conn->route);
-
- bufferevent_enable(conn->client, EV_READ | EV_WRITE);
- bufferevent_enable(conn->upstream, EV_READ | EV_WRITE);
- return;
- }
-
- if (events & BEV_EVENT_TIMEOUT) {
- LOG_WARN("connection timed out");
- free_conn(conn);
- return;
- }
-
- if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
- if (events & BEV_EVENT_ERROR) {
- int err = EVUTIL_SOCKET_ERROR();
- LOG_ERROR("connection error",
- "err", _LOGV(evutil_socket_error_to_string(err))
- );
- }
-
- free_conn(conn);
- }
-
- (void)bev;
-}
-
-static int connect_upstream(struct bufferevent *bev, const struct endpoint *ep)
-{
- if (ep == NULL) {
- return -EINVAL;
- }
-
- switch (ep->kind) {
- case ENDPOINT_INET: {
- struct sockaddr_in addr;
-
- memset(&addr, 0, sizeof(addr));
- addr.sin_family = AF_INET;
- addr.sin_port = htons(ep->port);
-
- if (inet_pton(AF_INET, ep->host, &addr.sin_addr) != 1) {
- return -EINVAL;
- }
-
- if (bufferevent_socket_connect(
- bev,
- (struct sockaddr *)&addr,
- sizeof(addr)) < 0) {
- return -errno;
- }
-
- return 0;
- }
-
- case ENDPOINT_UNIX:
-#ifdef _WIN32
- return -ENOTSUP;
-#else
- {
- struct sockaddr_un addr;
-
- memset(&addr, 0, sizeof(addr));
- addr.sun_family = AF_UNIX;
-
- if (ep->path[0] == '\0') {
- return -EINVAL;
- }
-
- if (strlen(ep->path) >= sizeof(addr.sun_path)) {
- return -ENAMETOOLONG;
- }
-
- strcpy(addr.sun_path, ep->path);
-
- if (bufferevent_socket_connect(
- bev,
- (struct sockaddr *)&addr,
- sizeof(addr)) < 0) {
- return -errno;
- }
-
- return 0;
- }
-#endif
-
- default:
- return -ENOTSUP;
- }
-}
-
-static const char *tcp_client_addr_string(conn_t *conn, char *buf, size_t buf_len)
-{
- struct sockaddr_storage ss;
- socklen_t len = sizeof(ss);
- evutil_socket_t fd;
- int port = 0;
-
- if (conn == NULL || conn->client == NULL || buf == NULL || buf_len == 0) {
- return NULL;
- }
-
- fd = bufferevent_getfd(conn->client);
- if (fd < 0) {
- return NULL;
- }
-
- if (getpeername(fd, (struct sockaddr *)&ss, &len) < 0) {
- return NULL;
- }
-
- if (ss.ss_family == AF_INET) {
- struct sockaddr_in *sin = (struct sockaddr_in *)&ss;
-
- if (inet_ntop(AF_INET, &sin->sin_addr, buf, buf_len) == NULL) {
- return NULL;
- }
-
- port = ntohs(sin->sin_port);
- } else if (ss.ss_family == AF_INET6) {
- struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)&ss;
-
- if (inet_ntop(AF_INET6, &sin6->sin6_addr, buf, buf_len) == NULL) {
- return NULL;
- }
-
- port = ntohs(sin6->sin6_port);
- } else {
- snprintf(buf, buf_len, "unknown");
- return buf;
- }
-
- snprintf(buf + strlen(buf), buf_len - strlen(buf), ":%d", port);
- return buf;
-}
-
-static void builtin_client_read_cb(struct bufferevent *bev, void *arg)
-{
- struct evbuffer *input = bufferevent_get_input(bev);
-
- (void)arg;
-
- evbuffer_drain(input, evbuffer_get_length(input));
-}
-
-static void builtin_close_after_write_cb(struct bufferevent *bev, void *arg)
-{
- conn_t *conn = arg;
-
- if (evbuffer_get_length(bufferevent_get_output(bev)) == 0) {
- free_conn(conn);
- }
-}
-
-static void file_done(struct file_conn *fc)
-{
- if (fc == NULL) {
- return;
- }
-
- if (fc->fp != NULL) {
- fclose(fc->fp);
- fc->fp = NULL;
- }
-
- free_conn(fc->conn);
- free(fc);
-}
-
-static void file_event_cb(struct bufferevent *bev, short events, void *arg)
-{
- struct file_conn *fc = arg;
-
- (void)bev;
- (void)events;
-
- file_done(fc);
-}
-
-static void file_write_cb(struct bufferevent *bev, void *arg)
-{
- struct file_conn *fc = arg;
- struct evbuffer *out = bufferevent_get_output(bev);
- char buf[FILE_CHUNK_SIZE];
-
- while (evbuffer_get_length(out) < BEV_READ_HIGH_WATER) {
- size_t n = fread(buf, 1, sizeof(buf), fc->fp);
-
- if (n > 0) {
- if (bufferevent_write(bev, buf, n) < 0) {
- file_done(fc);
- return;
- }
-
- continue;
- }
-
- if (ferror(fc->fp)) {
- file_done(fc);
- return;
- }
+#include "stream_listener.h"
+#include "compat_file.h"
- /*
- * EOF. If nothing is queued anymore, close now.
- * Otherwise keep the write callback installed so we close
- * after libevent drains the remaining output.
- */
- if (evbuffer_get_length(out) == 0) {
- file_done(fc);
- }
-
- return;
- }
-}
-
-static int start_tcp_file(conn_t *conn)
+int start_stream_route(struct worker *w, const struct route *r,
+ struct stream_route_ctx *ctx)
{
- const struct route *r;
- struct file_conn *fc;
- FILE *fp;
-
- if (conn == NULL || conn->route == NULL) {
- return -EINVAL;
- }
-
- r = conn->route;
-
- fp = fopen(r->upstream.path, "rb");
- if (fp == NULL) {
- int err = errno;
-
- LOG_ERROR("failed to open file upstream",
- "path", _LOGV(r->upstream.path),
- "err", _LOGV(strerror(err))
- );
+ int rc;
+ char opts[128];
- free_conn(conn);
- return -err;
- }
+ memset(ctx, 0, sizeof(*ctx));
- if (conn->upstream != NULL) {
- bufferevent_free(conn->upstream);
- conn->upstream = NULL;
- }
+ ctx->accept_base = w->base;
+ ctx->worker = w;
+ ctx->route = r;
- fc = calloc(1, sizeof(*fc));
- if (fc == NULL) {
- fclose(fp);
- free_conn(conn);
- return -ENOMEM;
+ rc = bind_stream_listener(ctx);
+ if (rc != 0) {
+ memset(ctx, 0, sizeof(*ctx));
+ return rc;
}
- fc->conn = conn;
- fc->fp = fp;
-
- bufferevent_setwatermark(conn->client, EV_WRITE, 0, BEV_READ_HIGH_WATER);
- bufferevent_setcb(conn->client, NULL, file_write_cb, file_event_cb, fc);
- bufferevent_enable(conn->client, EV_WRITE);
+ evconnlistener_set_error_cb(ctx->listener, accept_error_cb);
- set_client_idle_timeout(conn, r);
+ route_options_str(&r->opts, opts, sizeof(opts));
- file_write_cb(conn->client, fc);
+ LOG_INFO("route started",
+ "line", _LOGV(r->line_no),
+ "listen", _LOGV_ENDPOINT(&r->listen),
+ "upstream", _LOGV_ENDPOINT(&r->upstream),
+ "options", _LOGV(opts[0] ? opts : ""));
return 0;
}
-static int start_tcp_builtin(conn_t *conn)
+#ifndef _WIN32
+static void stop_unix_stream_route(struct stream_route_ctx *ctx)
{
- struct x_builtin_request req;
- struct x_builtin_response res;
const struct route *r;
- char client_addr[128];
- int rc;
-
- if (conn == NULL || conn->route == NULL) {
- return -EINVAL;
- }
-
- r = conn->route;
+ const char *path;
+ struct stat st;
- memset(&req, 0, sizeof(req));
-
- req.builtin = r->upstream.builtin;
- req.client_addr = tcp_client_addr_string(conn, client_addr, sizeof(client_addr));
-
- rc = x_builtin_handle(&req, &res);
- if (rc < 0) {
- free_conn(conn);
- return rc;
- }
-
- if (conn->upstream != NULL) {
- bufferevent_free(conn->upstream);
- conn->upstream = NULL;
- }
-
- set_client_idle_timeout(conn, r);
-
- switch (res.action) {
- case X_BUILTIN_ACTION_CLOSE:
- if (res.data_len > 0) {
- bufferevent_write(conn->client, res.data, res.data_len);
- bufferevent_setcb(conn->client, NULL, builtin_close_after_write_cb, event_cb, conn);
- bufferevent_enable(conn->client, EV_WRITE);
- } else {
- free_conn(conn);
- }
- return 0;
-
- case X_BUILTIN_ACTION_DISCARD:
- bufferevent_setcb(conn->client, builtin_client_read_cb, NULL, event_cb, conn);
- bufferevent_enable(conn->client, EV_READ);
- return 0;
-
- case X_BUILTIN_ACTION_HANG:
- bufferevent_setcb(conn->client, NULL, NULL, event_cb, conn);
- bufferevent_enable(conn->client, EV_READ);
- return 0;
-
- default:
- free_conn(conn);
- return -EINVAL;
- }
-}
-
-static void worker_adopt_client_fd(struct worker *w, struct accepted_client *ac) {
- conn_t *conn = calloc(1, sizeof(*conn));
- if (conn == NULL) {
- evutil_closesocket(ac->fd);
- return;
- }
-
- const struct route *r = ac->route;
-
- conn->owner = w;
- conn->route = r;
-
- conn->peer_addr = ac->peer_addr;
- conn->peer_addr_len = ac->peer_addr_len;
-
- conn->client = bufferevent_socket_new(w->base, ac->fd, BEV_OPT_CLOSE_ON_FREE);
- if (conn->client == NULL) {
- free(conn);
- evutil_closesocket(ac->fd);
+ if (ctx == NULL || ctx->route == NULL) {
return;
}
- if (ac->route->upstream.kind == ENDPOINT_BUILTIN) {
- start_tcp_builtin(conn);
- return;
- }
+ r = ctx->route;
- if (ac->route->upstream.kind == ENDPOINT_FILE) {
- start_tcp_file(conn);
+ if (r->listen.kind != ENDPOINT_UNIX) {
return;
}
- conn->upstream = bufferevent_socket_new(w->base, -1, BEV_OPT_CLOSE_ON_FREE);
- if (conn->upstream == NULL) {
- free_conn(conn);
- return;
- }
-
- int rc = set_socket_keepalive(ac->fd, r);
- if (rc < 0) {
- LOG_WARN("failed to enable client TCP keepalive",
- "err", _LOGV(strerror(-rc))
- );
- }
-
- bufferevent_setwatermark(conn->client, EV_READ, 0, BEV_READ_HIGH_WATER);
- bufferevent_setwatermark(conn->upstream, EV_READ, 0, BEV_READ_HIGH_WATER);
-
- bufferevent_setcb(conn->client, pipe_read_cb, pipe_write_cb, event_cb, conn);
- bufferevent_setcb(conn->upstream, pipe_read_cb, pipe_write_cb, event_cb, conn);
-
- /*
- * Do not read from the client yet.
- *
- * Otherwise client bytes may be copied into the upstream output buffer
- * before the PROXY v2 header is queued.
- */
- bufferevent_disable(conn->client, EV_READ);
-
- set_connect_timeout(conn, r);
-
- rc = connect_upstream(conn->upstream, &r->upstream);
- if (rc < 0) {
- LOG_ERROR("upstream connect failed",
- "upstream", _LOGV_ENDPOINT(&r->upstream),
- "err", _LOGV(strerror(-rc))
- );
- free_conn(conn);
+ path = r->listen.path;
+ if (path[0] == '\0') {
return;
}
- if (r->opts.keep_alive && r->upstream.kind == ENDPOINT_INET) {
- evutil_socket_t upstream_fd = bufferevent_getfd(conn->upstream);
-
- if (upstream_fd >= 0) {
- int rc = set_socket_keepalive(upstream_fd, r);
- if (rc < 0) {
- LOG_WARN("failed to enable upstream TCP keepalive",
- "upstream", _LOGV_ENDPOINT(&r->upstream),
- "err", _LOGV(strerror(-rc))
- );
- }
- }
- }
-
- if (r->opts.proxy_v2) {
- struct sockaddr_in local_addr;
- socklen_t local_len = sizeof(local_addr);
-
- memset(&local_addr, 0, sizeof(local_addr));
-
- if (getsockname(ac->fd, (struct sockaddr *)&local_addr, &local_len) < 0) {
- perror("getsockname");
- free_conn(conn);
- return;
- }
-
- if (ac->peer_addr_len <= 0) {
- LOG_ERROR("invalid client address");
- free_conn(conn);
- return;
- }
-
- if (ac->peer_addr.ss_family != AF_INET ||
- local_addr.sin_family != AF_INET) {
- LOG_ERROR("PROXY v2 currently only supports IPv4 TCP");
- free_conn(conn);
- return;
- }
-
- unsigned char hdr[256];
- size_t hdr_len = 0;
-
- int rc = proxy_v2_build(
- hdr,
- sizeof(hdr),
- (const struct sockaddr *)&ac->peer_addr,
- ac->peer_addr_len,
- (const struct sockaddr *)&local_addr,
- local_len,
- SOCK_STREAM,
- &hdr_len
- );
-
- if (rc == 0) {
- rc = bufferevent_write(conn->upstream, hdr, hdr_len);
- if (rc < 0) {
- rc = -EIO;
- }
+ if (lstat(path, &st) < 0) {
+ if (errno != ENOENT) {
+ LOG_WARN("failed to inspect unix stream listener socket",
+ "line", _LOGV(r->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(strerror(errno)));
}
-
- if (rc < 0) {
- LOG_ERROR("failed to write PROXY v2 header", "err", _LOGV(strerror(-rc)));
- free_conn(conn);
- return;
- }
- }
-
- bufferevent_enable(conn->client, EV_READ | EV_WRITE);
- bufferevent_enable(conn->upstream, EV_READ | EV_WRITE);
-}
-
-static void dispatch_client_fd(struct worker *w, struct accepted_client *ac) {
- worker_adopt_client_fd(w, ac);
-}
-
-static void accept_cb(
- struct evconnlistener *listener,
- evutil_socket_t client_fd,
- struct sockaddr *addr,
- int socklen,
- void *arg
-) {
- (void)listener;
-
- struct stream_route_ctx *ctx = arg;
- struct accepted_client ac = {
- .fd = client_fd,
- .route = ctx->route,
- };
-
- if (addr != NULL && socklen > 0 && (size_t)socklen <= sizeof(ac.peer_addr)) {
- memcpy(&ac.peer_addr, addr, (size_t)socklen);
- ac.peer_addr_len = (socklen_t)socklen;
- } else {
- LOG_ERROR("invalid accepted client address", "socklen", _LOGV(socklen));
- evutil_closesocket(client_fd);
return;
}
- dispatch_client_fd(ctx->worker, &ac);
-}
-
-static void accept_error_cb(struct evconnlistener *listener, void *arg)
-{
- struct stream_route_ctx *ctx = arg;
- int err = EVUTIL_SOCKET_ERROR();
-
- LOG_ERROR("accept error", "err", _LOGV(evutil_socket_error_to_string(err)));
-
- evconnlistener_disable(listener);
- event_base_loopexit(ctx->accept_base, NULL);
-}
-
-static int bind_tcp_stream_listener(struct stream_route_ctx *ctx)
-{
- const struct route *r = ctx->route;
- struct sockaddr_in listen_addr;
-
- memset(&listen_addr, 0, sizeof(listen_addr));
- listen_addr.sin_family = AF_INET;
- listen_addr.sin_port = htons(r->listen.port);
-
- if (inet_pton(AF_INET, r->listen.host, &listen_addr.sin_addr) != 1) {
- LOG_ERROR("invalid listen address",
- "listen", _LOGV_ENDPOINT(&r->listen));
- return -EINVAL;
- }
-
- ctx->listener = evconnlistener_new_bind(
- ctx->accept_base,
- accept_cb,
- ctx,
- LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE,
- 128,
- (struct sockaddr *)&listen_addr,
- sizeof(listen_addr)
- );
-
- if (ctx->listener == NULL) {
- LOG_ERROR("evconnlistener_new_bind failed",
- "listen", _LOGV_ENDPOINT(&r->listen));
- return -EADDRINUSE;
- }
-
- return 0;
-}
-
-int start_stream_route(struct worker *w, const struct route *r,
- struct stream_route_ctx *ctx)
-{
- int rc;
- char opts[128];
-
- memset(ctx, 0, sizeof(*ctx));
-
- ctx->accept_base = w->base;
- ctx->worker = w;
- ctx->route = r;
-
- switch (r->listen.proto) {
- case PROTO_TCP:
- rc = bind_tcp_stream_listener(ctx);
- break;
-
- default:
- LOG_ERROR("stream listen protocol not implemented yet",
+ if (!S_ISSOCK(st.st_mode)) {
+ LOG_WARN("not removing unix stream listener path because it is not a socket",
"line", _LOGV(r->line_no),
- "listen", _LOGV_ENDPOINT(&r->listen));
- rc = -ENOTSUP;
- break;
+ "path", _LOGV(path));
+ return;
}
- if (rc != 0) {
- memset(ctx, 0, sizeof(*ctx));
- return rc;
+ if (unlink(path) < 0) {
+ LOG_WARN("failed to remove unix stream listener socket",
+ "line", _LOGV(r->line_no),
+ "path", _LOGV(path),
+ "err", _LOGV(strerror(errno)));
}
-
- evconnlistener_set_error_cb(ctx->listener, accept_error_cb);
-
- route_options_str(&r->opts, opts, sizeof(opts));
-
- LOG_INFO("route started",
- "line", _LOGV(r->line_no),
- "listen", _LOGV_ENDPOINT(&r->listen),
- "upstream", _LOGV_ENDPOINT(&r->upstream),
- "options", _LOGV(opts[0] ? opts : ""));
-
- return 0;
}
+#endif
void stop_stream_route(struct stream_route_ctx *ctx)
{
@@ -747,11 +97,15 @@ void stop_stream_route(struct stream_route_ctx *ctx)
evconnlistener_free(ctx->listener);
}
+#ifndef _WIN32
+ stop_unix_stream_route(ctx);
+#endif
+
memset(ctx, 0, sizeof(*ctx));
}
#ifdef FUZZ
-int tcp_route_adopt_client_for_fuzz(
+int stream_route_adopt_client_for_fuzz(
struct event_base *base,
const struct route *r,
evutil_socket_t client_fd,
diff --git a/src/stream_route.h b/src/stream_route.h
index 7ab549e..44d88a6 100644
--- a/src/stream_route.h
+++ b/src/stream_route.h
@@ -1,6 +1,12 @@
#ifndef STREAM_ROUTE_H
#define STREAM_ROUTE_H
+#include <event2/listener.h>
+#include <event2/bufferevent.h>
+#include <event2/util.h>
+
+#include "compat_socket.h"
+
struct worker;
struct route;
struct event_base;
@@ -13,6 +19,24 @@ struct stream_route_ctx {
struct evconnlistener *listener;
};
+typedef struct conn_s {
+ struct worker *owner;
+ const struct route *route;
+
+ struct bufferevent *client;
+ struct bufferevent *upstream;
+
+ struct sockaddr_storage peer_addr;
+ socklen_t peer_addr_len;
+} conn_t;
+
+struct accepted_client {
+ evutil_socket_t fd;
+ struct sockaddr_storage peer_addr;
+ socklen_t peer_addr_len;
+ const struct route *route;
+};
+
int start_stream_route(struct worker *w, const struct route *r,
struct stream_route_ctx *ctx);