1334 lines
40 KiB
C++
1334 lines
40 KiB
C++
|
|
#include <netinet/in.h>
|
|
#include <sys/socket.h>
|
|
#include <stdlib.h>
|
|
#include <stdio.h>
|
|
#include <string.h>
|
|
#include <errno.h>
|
|
#include <assert.h>
|
|
|
|
#include <event2/event.h>
|
|
#include <event2/listener.h>
|
|
#include <event2/bufferevent.h>
|
|
#include <event2/bufferevent_ssl.h>
|
|
#include <event2/buffer.h>
|
|
#include <event2/thread.h>
|
|
#include <event2/dns.h>
|
|
|
|
#include <openssl/ssl.h>
|
|
#include <openssl/err.h>
|
|
#include <openssl/rand.h>
|
|
#include <openssl/x509.h>
|
|
#include <openssl/x509v3.h>
|
|
|
|
#include <tfe_stream.h>
|
|
#include <tfe_utils.h>
|
|
#include <tfe_future.h>
|
|
#include <tfe_plugin.h>
|
|
#include <tfe_proxy.h>
|
|
|
|
#include <platform.h>
|
|
#include <ssl_stream.h>
|
|
#include <tcp_stream.h>
|
|
#include <proxy.h>
|
|
#include <netinet/tcp.h>
|
|
#include <event2/bufferevent_struct.h>
|
|
|
|
#ifndef TFE_CONFIG_OUTPUT_LIMIT_DEFAULT
|
|
#define TFE_CONFIG_OUTPUT_LIMIT_DEFAULT (1024 * 1024)
|
|
#endif
|
|
|
|
/* forward declaration of libevent callbacks */
|
|
static void __stream_bev_readcb(struct bufferevent *, void *);
|
|
static void __stream_bev_writecb(struct bufferevent *, void *);
|
|
static void __stream_bev_eventcb(struct bufferevent *, short, void *);
|
|
|
|
/* ====================================================================================================================
|
|
* HELPER FUNCTIONS
|
|
* ===================================================================================================================*/
|
|
|
|
static inline void __stream_log_event(struct tfe_stream_private * _stream,
|
|
enum tfe_stream_event_log_type type, enum tfe_conn_dir dir, unsigned int error, const char * str_error)
|
|
{
|
|
unsigned int log_offset = _stream->nr_log_event;
|
|
assert(log_offset < STREAM_EVENT_LOG_MAX);
|
|
|
|
_stream->log_event[log_offset].type = type;
|
|
_stream->log_event[log_offset].dir = dir;
|
|
_stream->log_event[log_offset].error = error;
|
|
_stream->log_event[log_offset].str_error = str_error;
|
|
_stream->nr_log_event++;
|
|
}
|
|
|
|
static const char * __str_stream_log_type(enum tfe_stream_event_log_type type)
|
|
{
|
|
static const char * map_event_log_type[] =
|
|
{
|
|
[EVENT_LOG_CLOSE_BY_FD_PEER] = "FD/PEER",
|
|
[EVENT_LOG_CLOSE_BY_FD_EOF] = "FD/EOF",
|
|
[EVENT_LOG_CLOSE_BY_FD_ERROR] = "FD/ERR",
|
|
[EVENT_LOG_CLOSE_BY_SSL_ERROR] = "SSL/ERR"
|
|
};
|
|
return map_event_log_type[type];
|
|
}
|
|
|
|
static inline struct tfe_stream_private * to_stream_private(const struct tfe_stream * stream)
|
|
{
|
|
return container_of(stream, struct tfe_stream_private, head);
|
|
}
|
|
|
|
static inline struct tfe_conn_private * __this_conn(struct tfe_stream_private * _stream, enum tfe_conn_dir dir)
|
|
{
|
|
return ((dir == CONN_DIR_DOWNSTREAM) ? (_stream->conn_downstream) : (_stream->conn_upstream));
|
|
}
|
|
|
|
static inline struct tfe_conn_private * __peer_conn(struct tfe_stream_private * _stream, enum tfe_conn_dir dir)
|
|
{
|
|
return ((dir == CONN_DIR_DOWNSTREAM) ? (_stream->conn_upstream) : (_stream->conn_downstream));
|
|
}
|
|
|
|
static inline enum tfe_conn_dir __bev_dir(struct tfe_stream_private * _stream, struct bufferevent * bev)
|
|
{
|
|
if (_stream->conn_downstream && bev == _stream->conn_downstream->bev)
|
|
{
|
|
return CONN_DIR_DOWNSTREAM;
|
|
}
|
|
|
|
if (_stream->conn_upstream && bev == _stream->conn_upstream->bev)
|
|
{
|
|
return CONN_DIR_UPSTREAM;
|
|
}
|
|
|
|
assert(0);
|
|
return CONN_DIR_DOWNSTREAM;
|
|
}
|
|
|
|
static inline const char * __str_dir(enum tfe_conn_dir dir)
|
|
{
|
|
return dir == CONN_DIR_DOWNSTREAM ? "DOWNSTREAM" : "UPSTREAM";
|
|
}
|
|
|
|
static inline bool __is_ssl(struct tfe_stream_private * _stream)
|
|
{
|
|
return (_stream->session_type == STREAM_PROTO_SSL);
|
|
}
|
|
|
|
static void call_plugin_close(struct tfe_stream_private * _stream)
|
|
{
|
|
unsigned int plugin_id_iter = 0;
|
|
unsigned int plugin_id = 0;
|
|
|
|
for (const struct tfe_plugin * p_info_iter = tfe_plugin_iterate(&plugin_id_iter);
|
|
p_info_iter != NULL; p_info_iter = tfe_plugin_iterate(&plugin_id_iter))
|
|
{
|
|
_stream->calling_idx = plugin_id;
|
|
struct plugin_ctx * plug_ctx = &_stream->plugin_ctxs[plugin_id];
|
|
|
|
/* TODO: do not use pme to determinate we call on_open or not ever. */
|
|
if (p_info_iter->on_close && plug_ctx->pme != NULL)
|
|
{
|
|
p_info_iter->on_close(&_stream->head, _stream->thread_ref->thread_id,
|
|
REASON_PASSIVE_CLOSED, &(plug_ctx->pme));
|
|
}
|
|
|
|
plugin_id++;
|
|
}
|
|
}
|
|
|
|
/* ====================================================================================================================
|
|
* INTERFACE
|
|
* ===================================================================================================================*/
|
|
|
|
void tfe_stream_detach(const struct tfe_stream * stream)
|
|
{
|
|
struct tfe_stream_private * _stream = to_stream_private(stream);
|
|
int plug_id = _stream->calling_idx;
|
|
_stream->plugin_ctxs[plug_id].state = PLUG_STATE_DETACHED;
|
|
}
|
|
|
|
int tfe_stream_preempt(const struct tfe_stream * stream)
|
|
{
|
|
struct tfe_stream_private * _stream = to_stream_private(stream);
|
|
int plug_id = _stream->calling_idx;
|
|
for (unsigned int i = 0; i < _stream->nr_plugin_ctxs; i++)
|
|
{
|
|
if (_stream->plugin_ctxs[i].state == PLUG_STATE_PREEPTION)
|
|
{
|
|
return -1;
|
|
}
|
|
}
|
|
_stream->plugin_ctxs[plug_id].state = PLUG_STATE_PREEPTION;
|
|
return 0;
|
|
}
|
|
|
|
void tfe_stream_suspend(const struct tfe_stream * stream, enum tfe_conn_dir by)
|
|
{
|
|
struct tfe_stream_private * _stream = to_stream_private(stream);
|
|
assert(_stream != NULL);
|
|
assert(_stream->conn_upstream != NULL && _stream->conn_downstream != NULL);
|
|
assert(_stream->conn_upstream->bev != NULL);
|
|
assert(_stream->conn_downstream->bev != NULL);
|
|
|
|
/* stream cannot be suspended twice or more */
|
|
assert(!_stream->is_suspended);
|
|
_stream->is_suspended = true;
|
|
_stream->suspended_by = by;
|
|
|
|
/* disable all events */
|
|
int ret = 0;
|
|
ret = bufferevent_disable(_stream->conn_upstream->bev, EV_READ | EV_WRITE);
|
|
assert(ret == 0);
|
|
|
|
ret = bufferevent_disable(_stream->conn_downstream->bev, EV_READ | EV_WRITE);
|
|
assert(ret == 0);
|
|
|
|
(void) ret;
|
|
}
|
|
|
|
void tfe_stream_resume(const struct tfe_stream * stream)
|
|
{
|
|
struct tfe_stream_private * _stream = to_stream_private(stream);
|
|
assert(_stream->is_suspended);
|
|
|
|
bufferevent_enable(_stream->conn_upstream->bev, EV_READ | EV_WRITE);
|
|
bufferevent_enable(_stream->conn_downstream->bev, EV_READ | EV_WRITE);
|
|
|
|
if (_stream->suspended_by == CONN_DIR_DOWNSTREAM)
|
|
{
|
|
bufferevent_trigger(_stream->conn_downstream->bev, EV_READ, BEV_OPT_DEFER_CALLBACKS);
|
|
}
|
|
else
|
|
{
|
|
bufferevent_trigger(_stream->conn_upstream->bev, EV_READ, BEV_OPT_DEFER_CALLBACKS);
|
|
}
|
|
|
|
_stream->is_suspended = false;
|
|
_stream->suspended_by = CONN_DIR_DOWNSTREAM;
|
|
}
|
|
|
|
struct tfe_stream_write_ctx * tfe_stream_write_frag_start(const struct tfe_stream * stream, enum tfe_conn_dir dir)
|
|
{
|
|
struct tfe_stream_private * _stream = to_stream_private(stream);
|
|
struct tfe_stream_write_ctx ** ref_write_ctx = NULL;
|
|
struct tfe_conn_private * this_conn = NULL;
|
|
struct tfe_conn_private * peer_conn = NULL;
|
|
|
|
if (dir == CONN_DIR_DOWNSTREAM)
|
|
{
|
|
this_conn = _stream->conn_downstream;
|
|
peer_conn = _stream->conn_upstream;
|
|
ref_write_ctx = &_stream->w_ctx_downstream;
|
|
}
|
|
else
|
|
{
|
|
this_conn = _stream->conn_upstream;
|
|
peer_conn = _stream->conn_downstream;
|
|
ref_write_ctx = &_stream->w_ctx_upstream;
|
|
}
|
|
|
|
if (*ref_write_ctx != NULL)
|
|
return NULL;
|
|
|
|
this_conn->on_writing = 1;
|
|
*ref_write_ctx = ALLOC(struct tfe_stream_write_ctx, 1);
|
|
(*ref_write_ctx)->_stream = _stream;
|
|
(*ref_write_ctx)->dir = dir;
|
|
|
|
bufferevent_disable(peer_conn->bev, EV_READ);
|
|
return *ref_write_ctx;
|
|
}
|
|
|
|
int tfe_stream_write_frag(struct tfe_stream_write_ctx * w_ctx, const unsigned char * data, size_t size)
|
|
{
|
|
if (w_ctx->_stream == NULL)
|
|
{
|
|
return -EPIPE;
|
|
}
|
|
|
|
struct tfe_conn_private * this_conn = __this_conn(w_ctx->_stream, w_ctx->dir);
|
|
int ret = 0;
|
|
if (this_conn != NULL)
|
|
{
|
|
ret = bufferevent_write(this_conn->bev, data, size);
|
|
}
|
|
else
|
|
{
|
|
return -EPIPE;
|
|
}
|
|
|
|
return ret;
|
|
}
|
|
|
|
void tfe_stream_write_frag_end(struct tfe_stream_write_ctx * w_ctx)
|
|
{
|
|
struct tfe_conn_private * this_conn;
|
|
struct tfe_conn_private * peer_conn;
|
|
|
|
/* The connection terminated before this function call */
|
|
if (w_ctx->_stream == NULL) goto __out;
|
|
this_conn = __this_conn(w_ctx->_stream, w_ctx->dir);
|
|
peer_conn = __peer_conn(w_ctx->_stream, w_ctx->dir);
|
|
|
|
if (this_conn != NULL)
|
|
{
|
|
this_conn->on_writing = 0;
|
|
bufferevent_enable(peer_conn->bev, EV_READ);
|
|
}
|
|
|
|
if (w_ctx->dir == CONN_DIR_DOWNSTREAM)
|
|
{
|
|
assert(w_ctx->_stream->w_ctx_downstream == w_ctx);
|
|
w_ctx->_stream->w_ctx_downstream = NULL;
|
|
}
|
|
else
|
|
{
|
|
assert(w_ctx->_stream->w_ctx_upstream == w_ctx);
|
|
w_ctx->_stream->w_ctx_upstream = NULL;
|
|
}
|
|
|
|
__out:
|
|
free(w_ctx);
|
|
}
|
|
|
|
int tfe_stream_write(const struct tfe_stream * stream, enum tfe_conn_dir dir, const unsigned char * data, size_t size)
|
|
{
|
|
int ret = 0;
|
|
struct tfe_stream_write_ctx * wctx = tfe_stream_write_frag_start(stream, dir);
|
|
ret = tfe_stream_write_frag(wctx, data, size);
|
|
tfe_stream_write_frag_end(wctx);
|
|
return ret;
|
|
}
|
|
|
|
static int conn_private_ratelimit_setup(struct tfe_conn_private * conn, struct tfe_proxy_rate_limit_options * opt)
|
|
{
|
|
conn->ratelimit_bucket = ev_token_bucket_cfg_new(opt->read_rate, opt->read_burst,
|
|
opt->write_rate, opt->write_burst, NULL);
|
|
|
|
if(unlikely(conn->ratelimit_bucket == NULL))
|
|
{
|
|
TFE_LOG_ERROR(g_default_logger, "Failed at setting ratelimit bucket, "
|
|
"read_rate = %u, read_burst = %u, write_rate = %u, write_burst = %u",
|
|
opt->read_rate, opt->read_burst, opt->write_rate, opt->write_burst);
|
|
return -1;
|
|
}
|
|
|
|
bufferevent_set_rate_limit(conn->bev, conn->ratelimit_bucket);
|
|
return 0;
|
|
}
|
|
|
|
int tfe_stream_action_set_opt(const struct tfe_stream * stream, enum tfe_stream_action_opt type,
|
|
void * value, size_t size)
|
|
{
|
|
struct tfe_stream_private * _stream = to_stream_private(stream);
|
|
|
|
#define __SAVE_PARAM(what) do { \
|
|
if(size != sizeof(__typeof__(what))) { return -EINVAL; } \
|
|
else { what = *(__typeof(what) *)value; } } while(0) \
|
|
|
|
switch (type)
|
|
{
|
|
case ACTION_OPT_FOWARD_BYTES: __SAVE_PARAM(_stream->forward_bytes);
|
|
break;
|
|
case ACTION_OPT_DEFER_BYTES: __SAVE_PARAM(_stream->defer_bytes);
|
|
break;
|
|
case ACTION_OPT_DEFER_TIME_TV: __SAVE_PARAM(_stream->defer_timeval);
|
|
break;
|
|
case ACTION_OPT_DROP_BYTES: __SAVE_PARAM(_stream->drop_bytes);
|
|
break;
|
|
}
|
|
|
|
#undef __SAVE_PARAM
|
|
return 0;
|
|
}
|
|
|
|
int tfe_stream_set_integer_opt(struct tfe_stream * stream, enum tfe_stream_opt_level level, int type, int val)
|
|
{
|
|
return 1;
|
|
}
|
|
/* ====================================================================================================================
|
|
* CONNECTION STRUCTURE AND OPERATION FUCTIONS
|
|
* ===================================================================================================================*/
|
|
|
|
evutil_socket_t __conn_private_release_fd(struct tfe_conn_private * conn)
|
|
{
|
|
evutil_socket_t __to_release_fd = conn->fd;
|
|
conn->fd = 0;
|
|
return __to_release_fd;
|
|
}
|
|
|
|
static void __conn_private_destory(struct tfe_conn_private * conn)
|
|
{
|
|
bufferevent_disable(conn->bev, EV_READ | EV_WRITE);
|
|
|
|
if(conn->ratelimit_bucket)
|
|
{
|
|
ev_token_bucket_cfg_free(conn->ratelimit_bucket);
|
|
conn->ratelimit_bucket = NULL;
|
|
}
|
|
|
|
bufferevent_free(conn->bev);
|
|
if (conn->fd > 0)
|
|
{
|
|
evutil_closesocket(conn->fd);
|
|
}
|
|
|
|
free(conn);
|
|
TFE_PROXY_STAT_INCREASE(STAT_FD_INSTANT_CLOSE, 1);
|
|
}
|
|
|
|
static void __conn_private_destory_with_ssl(struct event_base * ev_base,
|
|
struct tfe_conn_private * conn, struct ssl_stream * ssl_stream)
|
|
{
|
|
if (ssl_stream == NULL) return __conn_private_destory(conn);
|
|
evutil_socket_t __to_closed_fd = __conn_private_release_fd(conn);
|
|
ssl_stream_free_and_close_fd(ssl_stream, ev_base, __to_closed_fd);
|
|
return __conn_private_destory(conn);
|
|
}
|
|
|
|
static void __stream_bev_passthrough_readcb(struct bufferevent * bev, void * arg)
|
|
{
|
|
struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
|
|
struct tfe_conn_private * peer_conn = __peer_conn(_stream, __bev_dir(_stream, bev));
|
|
|
|
struct evbuffer * __input_buffer = bufferevent_get_input(bev);
|
|
if (peer_conn == NULL)
|
|
{
|
|
evbuffer_drain(__input_buffer, evbuffer_get_length(__input_buffer));
|
|
return;
|
|
}
|
|
|
|
struct evbuffer * __output_buffer = bufferevent_get_output(peer_conn->bev);
|
|
evbuffer_add_buffer(__output_buffer, __input_buffer);
|
|
}
|
|
|
|
static void __stream_bev_passthrough_writecb(struct bufferevent * bev, void * arg)
|
|
{
|
|
struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
|
|
struct tfe_conn_private ** ref_this_conn{};
|
|
struct tfe_conn_private ** ref_peer_conn{};
|
|
|
|
if (__bev_dir(_stream, bev) == CONN_DIR_UPSTREAM)
|
|
{
|
|
ref_this_conn = &_stream->conn_upstream;
|
|
ref_peer_conn = &_stream->conn_downstream;
|
|
}
|
|
|
|
if (__bev_dir(_stream, bev) == CONN_DIR_DOWNSTREAM)
|
|
{
|
|
ref_this_conn = &_stream->conn_downstream;
|
|
ref_peer_conn = &_stream->conn_upstream;
|
|
}
|
|
|
|
struct evbuffer * __output_buffer = bufferevent_get_output(bev);
|
|
assert(__output_buffer != NULL);
|
|
|
|
if (*ref_peer_conn == NULL && evbuffer_get_length(__output_buffer) == 0)
|
|
{
|
|
__conn_private_destory(*ref_this_conn);
|
|
*ref_this_conn = NULL;
|
|
}
|
|
|
|
if (*ref_peer_conn == NULL && *ref_this_conn == NULL)
|
|
{
|
|
call_plugin_close(_stream);
|
|
tfe_stream_destory(_stream);
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
static void __stream_bev_passthrough_eventcb(struct bufferevent * bev, short events, void * arg)
|
|
{
|
|
struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
|
|
struct tfe_conn_private ** ref_this_conn{};
|
|
struct tfe_conn_private ** ref_peer_conn{};
|
|
|
|
enum tfe_conn_dir conn_dir = __bev_dir(_stream, bev);
|
|
const char * str_conn_dir = __str_dir(conn_dir);
|
|
|
|
if (__bev_dir(_stream, bev) == CONN_DIR_UPSTREAM)
|
|
{
|
|
ref_this_conn = &_stream->conn_upstream;
|
|
ref_peer_conn = &_stream->conn_downstream;
|
|
}
|
|
|
|
if (__bev_dir(_stream, bev) == CONN_DIR_DOWNSTREAM)
|
|
{
|
|
ref_this_conn = &_stream->conn_downstream;
|
|
ref_peer_conn = &_stream->conn_upstream;
|
|
}
|
|
|
|
if (events & BEV_EVENT_ERROR || events & BEV_EVENT_EOF)
|
|
{
|
|
if (evbuffer_get_length(bufferevent_get_input(bev)))
|
|
{
|
|
__stream_bev_passthrough_readcb(bev, arg);
|
|
}
|
|
|
|
if(events & BEV_EVENT_ERROR)
|
|
{
|
|
unsigned long err;
|
|
while ((err = (bufferevent_get_openssl_error(bev))))
|
|
{
|
|
const char *msg = (const char*)ERR_reason_error_string(err);
|
|
const char *lib = (const char*)ERR_lib_error_string(err);
|
|
const char *func = (const char*)ERR_func_error_string(err);
|
|
|
|
TFE_LOG_INFO(g_default_logger, "%s %s connection error, bufferevent_get_openssl_error() = %lu: %s %s %s",
|
|
_stream->str_stream_addr, str_conn_dir, err, lib, func, msg);
|
|
}
|
|
|
|
if (errno)
|
|
{
|
|
TFE_LOG_INFO(g_default_logger, "%s %s connection error, errno = %d, %s",
|
|
_stream->str_stream_addr, str_conn_dir, errno, strerror(errno));
|
|
}
|
|
}
|
|
|
|
goto __close_connection;
|
|
}
|
|
|
|
return;
|
|
|
|
__close_connection:
|
|
if (*ref_peer_conn != NULL)
|
|
{
|
|
struct bufferevent * __peer_bev = (*ref_peer_conn)->bev;
|
|
struct evbuffer * __peer_output_buffer = bufferevent_get_output(__peer_bev);
|
|
|
|
if (evbuffer_get_length(__peer_output_buffer) == 0)
|
|
{
|
|
__conn_private_destory(*ref_peer_conn);
|
|
*ref_peer_conn = NULL;
|
|
}
|
|
}
|
|
|
|
if (*ref_this_conn != NULL)
|
|
{
|
|
__conn_private_destory(*ref_this_conn);
|
|
*ref_this_conn = NULL;
|
|
}
|
|
|
|
if (*ref_this_conn == NULL && *ref_peer_conn == NULL)
|
|
{
|
|
call_plugin_close(_stream);
|
|
tfe_stream_destory(_stream);
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
/*
|
|
* Callback for read events on the up- and downstream connection bufferevents.
|
|
* Called when there is data ready in the input evbuffer.
|
|
*/
|
|
|
|
static void __stream_bev_readcb(struct bufferevent * bev, void * arg)
|
|
{
|
|
struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
|
|
enum tfe_conn_dir dir = __bev_dir(_stream, bev);
|
|
struct tfe_conn_private * peer_conn = __peer_conn(_stream, dir);
|
|
|
|
/* Peer connection is terminated, drain all data.
|
|
* This connection will be destoryed in __event_cb */
|
|
struct evbuffer * inbuf = bufferevent_get_input(bev);
|
|
if (peer_conn == NULL)
|
|
{
|
|
evbuffer_drain(inbuf, evbuffer_get_length(inbuf));
|
|
return;
|
|
}
|
|
|
|
struct evbuffer * outbuf = bufferevent_get_output(peer_conn->bev);
|
|
assert(inbuf != NULL && outbuf != NULL);
|
|
|
|
enum tfe_stream_action action_tmp = ACTION_FORWARD_DATA;
|
|
enum tfe_stream_action action_final = ACTION_FORWARD_DATA;
|
|
|
|
size_t drain_size = 0;
|
|
size_t contigous_len = evbuffer_get_length(inbuf);
|
|
unsigned char * contiguous_data = evbuffer_pullup(inbuf, contigous_len);
|
|
|
|
_stream->defer_bytes = 0;
|
|
_stream->drop_bytes = 0;
|
|
_stream->forward_bytes = 0;
|
|
|
|
unsigned int plugin_id_iter = 0;
|
|
unsigned int plugin_id = 0;
|
|
|
|
for (const struct tfe_plugin * p_info_iter = tfe_plugin_iterate(&plugin_id_iter);
|
|
p_info_iter != NULL; p_info_iter = tfe_plugin_iterate(&plugin_id_iter), plugin_id++)
|
|
{
|
|
_stream->calling_idx = plugin_id;
|
|
struct plugin_ctx * plug_ctx = &_stream->plugin_ctxs[plugin_id];
|
|
|
|
if (p_info_iter->on_open != NULL && plug_ctx->is_plugin_opened == 0)
|
|
{
|
|
p_info_iter->on_open(&_stream->head, _stream->thread_ref->thread_id, dir, &(plug_ctx->pme));
|
|
plug_ctx->is_plugin_opened = 1;
|
|
}
|
|
|
|
if (plug_ctx->state == PLUG_STATE_DETACHED)
|
|
continue;
|
|
|
|
if (p_info_iter->on_data != NULL)
|
|
{
|
|
action_tmp = p_info_iter->on_data(&_stream->head, _stream->thread_ref->thread_id,
|
|
dir, contiguous_data, contigous_len, &(plug_ctx->pme));
|
|
}
|
|
|
|
if (plug_ctx->state == PLUG_STATE_PREEPTION)
|
|
action_final = action_tmp;
|
|
}
|
|
|
|
switch (action_final)
|
|
{
|
|
case ACTION_FORWARD_DATA:
|
|
if (_stream->forward_bytes > 0)
|
|
{
|
|
evbuffer_remove_buffer(inbuf, outbuf, _stream->forward_bytes);
|
|
}
|
|
else
|
|
{
|
|
evbuffer_add_buffer(outbuf, inbuf);
|
|
}
|
|
break;
|
|
|
|
case ACTION_DROP_DATA:
|
|
if (_stream->drop_bytes > 0)
|
|
{
|
|
drain_size = _stream->drop_bytes;
|
|
}
|
|
else
|
|
{
|
|
drain_size = evbuffer_get_length(inbuf);
|
|
}
|
|
evbuffer_drain(inbuf, drain_size);
|
|
break;
|
|
|
|
case ACTION_DEFER_DATA:
|
|
if (_stream->defer_bytes > 0)
|
|
{
|
|
bufferevent_setwatermark(bev, EV_WRITE, _stream->defer_bytes, 0);
|
|
}
|
|
break;
|
|
default: assert(0);
|
|
break;
|
|
}
|
|
|
|
if(_stream->need_to_be_kill)
|
|
{
|
|
const static struct linger sl{.l_onoff = 1, .l_linger = 0};
|
|
|
|
/* Set SO_LINGER, the fd will be closed by RST */
|
|
setsockopt(_stream->conn_upstream->fd, SOL_SOCKET, SO_LINGER, &sl, sizeof(sl));
|
|
setsockopt(_stream->conn_downstream->fd, SOL_SOCKET, SO_LINGER, &sl, sizeof(sl));
|
|
|
|
/* Destroy STREAM */
|
|
TFE_PROXY_STAT_INCREASE(STAT_STREAM_CLS_KILL, 1);
|
|
return tfe_stream_destory(_stream);
|
|
}
|
|
|
|
#if 0
|
|
if (evbuffer_get_length(outbuf) >= TFE_CONFIG_OUTPUT_LIMIT_DEFAULT)
|
|
{
|
|
bufferevent_setwatermark(peer_conn->bev, EV_WRITE, TFE_CONFIG_OUTPUT_LIMIT_DEFAULT / 2, TFE_CONFIG_OUTPUT_LIMIT_DEFAULT);
|
|
bufferevent_disable(bev, EV_READ);
|
|
}
|
|
#endif
|
|
}
|
|
|
|
/*
|
|
* Callback for write events on the up- and downstream connection bufferevents.
|
|
* Called when either all data from the output evbuffer has been written,
|
|
* or if the outbuf is only half full again after having been full.
|
|
*/
|
|
static void __stream_bev_writecb(struct bufferevent * bev, void * arg)
|
|
{
|
|
struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
|
|
struct event_base * ev_base = bufferevent_get_base(bev);
|
|
|
|
struct tfe_conn_private ** ref_this_conn{};
|
|
struct tfe_conn_private ** ref_peer_conn{};
|
|
struct ssl_stream ** ref_this_ssl_stream{};
|
|
enum tfe_conn_dir conn_dir = __bev_dir(_stream, bev);
|
|
|
|
if (conn_dir == CONN_DIR_UPSTREAM)
|
|
{
|
|
ref_this_conn = &_stream->conn_upstream;
|
|
ref_this_ssl_stream = &_stream->ssl_upstream;
|
|
ref_peer_conn = &_stream->conn_downstream;
|
|
}
|
|
|
|
if (conn_dir == CONN_DIR_DOWNSTREAM)
|
|
{
|
|
ref_this_conn = &_stream->conn_downstream;
|
|
ref_this_ssl_stream = &_stream->ssl_downstream;
|
|
ref_peer_conn = &_stream->conn_upstream;
|
|
}
|
|
|
|
struct evbuffer * __output_buffer = bufferevent_get_output(bev);
|
|
assert(__output_buffer != NULL);
|
|
|
|
if (*ref_peer_conn == NULL /* Peer connection is closed */
|
|
&& (*ref_this_conn)->on_writing == 0 /* No body is prepare to write data, eg. No body call stream_write */
|
|
&& evbuffer_get_length(__output_buffer) == 0) /* Nothing is in send queue */
|
|
{
|
|
__conn_private_destory_with_ssl(ev_base, *ref_this_conn, *ref_this_ssl_stream);
|
|
__stream_log_event(_stream, EVENT_LOG_CLOSE_BY_FD_PEER, conn_dir, 0, NULL);
|
|
*ref_this_conn = NULL;
|
|
*ref_this_ssl_stream = NULL;
|
|
}
|
|
|
|
if (*ref_peer_conn == NULL && *ref_this_conn == NULL)
|
|
{
|
|
call_plugin_close(_stream);
|
|
tfe_stream_destory(_stream);
|
|
}
|
|
}
|
|
|
|
/*
|
|
* Callback for meta events on the up- and downstream connection bufferevents.
|
|
* Called when EOF has been reached, a connection has been made, and on errors.
|
|
*/
|
|
static void __stream_bev_eventcb(struct bufferevent * bev, short events, void * arg)
|
|
{
|
|
struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
|
|
struct event_base * ev_base = bufferevent_get_base(bev);
|
|
|
|
struct tfe_conn_private ** ref_this_conn{};
|
|
struct tfe_conn_private ** ref_peer_conn{};
|
|
struct ssl_stream ** ref_this_ssl_stream{};
|
|
struct ssl_stream ** ref_peer_ssl_stream{};
|
|
struct tfe_stream_write_ctx ** ref_this_write_ctx{};
|
|
|
|
enum tfe_conn_dir conn_dir = __bev_dir(_stream, bev);
|
|
const char * str_conn_dir = __str_dir(conn_dir);
|
|
enum tfe_conn_dir peer_conn_dir{};
|
|
|
|
if (__bev_dir(_stream, bev) == CONN_DIR_UPSTREAM)
|
|
{
|
|
ref_this_conn = &_stream->conn_upstream;
|
|
ref_peer_conn = &_stream->conn_downstream;
|
|
ref_this_ssl_stream = &_stream->ssl_upstream;
|
|
ref_peer_ssl_stream = &_stream->ssl_downstream;
|
|
ref_this_write_ctx = &_stream->w_ctx_upstream;
|
|
peer_conn_dir = CONN_DIR_DOWNSTREAM;
|
|
}
|
|
|
|
if (__bev_dir(_stream, bev) == CONN_DIR_DOWNSTREAM)
|
|
{
|
|
ref_this_conn = &_stream->conn_downstream;
|
|
ref_peer_conn = &_stream->conn_upstream;
|
|
ref_this_ssl_stream = &_stream->ssl_downstream;
|
|
ref_peer_ssl_stream = &_stream->ssl_upstream;
|
|
ref_this_write_ctx = &_stream->w_ctx_downstream;
|
|
peer_conn_dir = CONN_DIR_UPSTREAM;
|
|
}
|
|
|
|
if (events & BEV_EVENT_ERROR || events & BEV_EVENT_EOF)
|
|
{
|
|
if (evbuffer_get_length(bufferevent_get_input(bev)))
|
|
{
|
|
__stream_bev_readcb(bev, arg);
|
|
}
|
|
|
|
if(events & BEV_EVENT_ERROR)
|
|
{
|
|
if(_stream->session_type==STREAM_PROTO_SSL)
|
|
{
|
|
ssl_stream_log_error(bev, __bev_dir(_stream, bev), _stream->ssl_mgr);
|
|
}
|
|
else if(errno)
|
|
{
|
|
TFE_LOG_INFO(g_default_logger, "%s %s connection error, errno = %d, %s",
|
|
_stream->str_stream_addr, str_conn_dir, errno, strerror(errno));
|
|
}
|
|
}
|
|
|
|
if(events & BEV_EVENT_ERROR) __stream_log_event(_stream, EVENT_LOG_CLOSE_BY_FD_ERROR, conn_dir, 0, NULL);
|
|
if(events & BEV_EVENT_EOF) __stream_log_event(_stream, EVENT_LOG_CLOSE_BY_FD_EOF, conn_dir, 0, NULL);
|
|
goto __close_connection;
|
|
}
|
|
|
|
return;
|
|
|
|
__close_connection:
|
|
if (*ref_peer_conn != NULL)
|
|
{
|
|
struct bufferevent * __peer_bev = (*ref_peer_conn)->bev;
|
|
struct evbuffer * __peer_output_buffer = bufferevent_get_output(__peer_bev);
|
|
|
|
if (evbuffer_get_length(__peer_output_buffer) == 0 && (*ref_peer_conn)->on_writing == 0)
|
|
{
|
|
__conn_private_destory_with_ssl(ev_base, *ref_peer_conn, *ref_peer_ssl_stream);
|
|
*ref_peer_conn = NULL;
|
|
*ref_peer_ssl_stream = NULL;
|
|
|
|
__stream_log_event(_stream, EVENT_LOG_CLOSE_BY_FD_PEER, peer_conn_dir, 0, NULL);
|
|
}
|
|
}
|
|
|
|
if (*ref_this_conn != NULL)
|
|
{
|
|
/* There is a frag writter setted, need to clear the reference of stream in the writter to indicate the connection is closed */
|
|
if (*ref_this_write_ctx != NULL)
|
|
{
|
|
(*ref_this_write_ctx)->_stream = NULL;
|
|
}
|
|
|
|
__conn_private_destory_with_ssl(ev_base, *ref_this_conn, *ref_this_ssl_stream);
|
|
*ref_this_conn = NULL;
|
|
*ref_this_ssl_stream = NULL;
|
|
}
|
|
|
|
if (*ref_this_conn == NULL && *ref_peer_conn == NULL)
|
|
{
|
|
goto __call_plugin_close;
|
|
}
|
|
|
|
return;
|
|
|
|
__call_plugin_close:
|
|
call_plugin_close(_stream);
|
|
tfe_stream_destory(_stream);
|
|
}
|
|
|
|
static tfe_conn_private * __conn_private_create_by_bev(struct tfe_stream_private * stream, struct bufferevent * bev)
|
|
{
|
|
struct tfe_conn_private * __conn_private = ALLOC(struct tfe_conn_private, 1);
|
|
__conn_private->bev = bev;
|
|
__conn_private->fd = bufferevent_getfd(bev);
|
|
|
|
if(stream->tcp_passthough)
|
|
{
|
|
bufferevent_setcb(__conn_private->bev, __stream_bev_passthrough_readcb,
|
|
__stream_bev_passthrough_writecb, __stream_bev_passthrough_eventcb, stream);
|
|
}
|
|
else
|
|
{
|
|
bufferevent_setcb(__conn_private->bev, __stream_bev_readcb, __stream_bev_writecb, __stream_bev_eventcb, stream);
|
|
}
|
|
bufferevent_disable(__conn_private->bev, EV_READ | EV_WRITE);
|
|
|
|
struct tfe_proxy * proxy_ref = stream->proxy_ref;
|
|
if(unlikely(proxy_ref->en_rate_limit))
|
|
{
|
|
conn_private_ratelimit_setup(__conn_private, &proxy_ref->rate_limit_options);
|
|
}
|
|
|
|
return __conn_private;
|
|
}
|
|
|
|
static tfe_conn_private * __conn_private_create_by_fd(struct tfe_stream_private * stream, evutil_socket_t fd)
|
|
{
|
|
struct tfe_conn_private * __conn_private = ALLOC(struct tfe_conn_private, 1);
|
|
struct event_base * __ev_base = stream->thread_ref->evbase;
|
|
struct tfe_proxy * proxy_ref = stream->proxy_ref;
|
|
|
|
__conn_private->_stream_ref = stream;
|
|
__conn_private->bev = bufferevent_socket_new(__ev_base, fd, BEV_OPT_DEFER_CALLBACKS | BEV_OPT_THREADSAFE );
|
|
__conn_private->fd = fd;
|
|
|
|
bufferevent_disable(__conn_private->bev, EV_READ | EV_WRITE);
|
|
|
|
if (!__conn_private->bev)
|
|
{
|
|
TFE_LOG_ERROR(__STREAM_LOGGER(stream), "Failed at creating bufferevent for fd %d", fd);
|
|
goto __errout;
|
|
}
|
|
|
|
if (stream->tcp_passthough)
|
|
{
|
|
bufferevent_setcb(__conn_private->bev, __stream_bev_passthrough_readcb,
|
|
__stream_bev_passthrough_writecb, __stream_bev_passthrough_eventcb, stream);
|
|
}
|
|
else
|
|
{
|
|
bufferevent_setcb(__conn_private->bev, __stream_bev_readcb,
|
|
__stream_bev_writecb, __stream_bev_eventcb, stream);
|
|
}
|
|
|
|
if(proxy_ref->en_rate_limit)
|
|
{
|
|
conn_private_ratelimit_setup(__conn_private, &proxy_ref->rate_limit_options);
|
|
}
|
|
|
|
return __conn_private;
|
|
|
|
__errout:
|
|
if (__conn_private != NULL) free(__conn_private);
|
|
return NULL;
|
|
}
|
|
|
|
void __conn_private_enable(struct tfe_conn_private * conn_private)
|
|
{
|
|
assert(conn_private != NULL && conn_private->bev != NULL);
|
|
bufferevent_enable(conn_private->bev, EV_READ | EV_WRITE);
|
|
}
|
|
|
|
void ssl_downstream_create_on_success(future_result_t * result, void * user)
|
|
{
|
|
struct tfe_stream_private * _stream = (struct tfe_stream_private *) user;
|
|
struct ssl_stream * downstream = ssl_downstream_create_result_release_stream(result);
|
|
struct bufferevent * bev = ssl_downstream_create_result_release_bev(result);
|
|
|
|
_stream->conn_downstream = __conn_private_create_by_bev(_stream, bev);
|
|
_stream->ssl_downstream = downstream;
|
|
|
|
future_destroy(_stream->future_downstream_create);
|
|
_stream->future_downstream_create = NULL;
|
|
_stream->defer_fd_downstream = 0;
|
|
|
|
assert(_stream->conn_downstream != NULL && _stream->conn_upstream != NULL);
|
|
|
|
ssl_stream_dump_info(_stream->ssl_downstream, _stream->ssl_downstream_info_dump,
|
|
sizeof(_stream->ssl_downstream_info_dump));
|
|
|
|
ssl_stream_dump_info(_stream->ssl_upstream, _stream->ssl_upstream_info_dump,
|
|
sizeof(_stream->ssl_upstream_info_dump));
|
|
|
|
__conn_private_enable(_stream->conn_downstream);
|
|
__conn_private_enable(_stream->conn_upstream);
|
|
|
|
return;
|
|
}
|
|
|
|
void ssl_downstream_create_on_fail(enum e_future_error err, const char * what, void * user)
|
|
{
|
|
struct tfe_stream_private * _stream = (struct tfe_stream_private *) user;
|
|
assert(_stream != NULL && _stream->session_type == STREAM_PROTO_SSL);
|
|
|
|
TFE_LOG_INFO(g_default_logger, "%s Failed to create SSL downstream, close the connection : %s. ",
|
|
_stream->str_stream_addr, what);
|
|
|
|
__stream_log_event(_stream, EVENT_LOG_CLOSE_BY_SSL_ERROR, CONN_DIR_DOWNSTREAM, 0, NULL);
|
|
tfe_stream_destory(_stream);
|
|
}
|
|
|
|
void ssl_upstream_create_on_success(future_result_t * result, void * user)
|
|
{
|
|
struct tfe_stream_private * _stream = (struct tfe_stream_private *) user;
|
|
evutil_socket_t fd=-1;
|
|
enum ssl_stream_action ssl_action=ssl_upstream_create_result_release_action(result);
|
|
if(SSL_ACTION_PASSTHROUGH==ssl_action)
|
|
{
|
|
_stream->tcp_passthough=1;
|
|
fd=ssl_upstream_create_result_release_fd(result);
|
|
_stream->conn_upstream=__conn_private_create_by_fd(_stream, fd);
|
|
_stream->conn_downstream=__conn_private_create_by_fd(_stream, _stream->defer_fd_downstream);
|
|
__conn_private_enable(_stream->conn_downstream);
|
|
__conn_private_enable(_stream->conn_upstream);
|
|
|
|
}
|
|
else if(SSL_ACTION_SHUTDOWN==ssl_action)
|
|
{
|
|
tfe_stream_destory(_stream);
|
|
}
|
|
else
|
|
{
|
|
struct ssl_stream * upstream = ssl_upstream_create_result_release_stream(result);
|
|
struct bufferevent * bev = ssl_upstream_create_result_release_bev(result);
|
|
assert(upstream != NULL && bev != NULL);
|
|
|
|
/* Create connection ctx by bev */
|
|
_stream->conn_upstream = __conn_private_create_by_bev(_stream, bev);
|
|
_stream->ssl_upstream = upstream;
|
|
|
|
assert(_stream->conn_upstream != NULL);
|
|
assert(_stream->ssl_upstream != NULL);
|
|
|
|
/* Next, create downstream */
|
|
_stream->future_downstream_create = future_create("ssl_down", ssl_downstream_create_on_success,
|
|
ssl_downstream_create_on_fail, _stream);
|
|
|
|
ssl_async_downstream_create(_stream->future_downstream_create, _stream->ssl_mgr,
|
|
_stream->ssl_upstream, _stream->defer_fd_downstream, _stream->thread_ref->thread_id);
|
|
}
|
|
future_destroy(_stream->future_upstream_create);
|
|
_stream->future_upstream_create = NULL;
|
|
_stream->defer_fd_upstream = 0;
|
|
|
|
}
|
|
|
|
void ssl_upstream_create_on_fail(enum e_future_error err, const char * what, void * user)
|
|
{
|
|
struct tfe_stream_private * _stream = (struct tfe_stream_private *) user;
|
|
assert(_stream != NULL && _stream->session_type == STREAM_PROTO_SSL);
|
|
|
|
TFE_LOG_INFO(g_default_logger, "%s Failed to create SSL upstream, close the connection : %s. ",
|
|
_stream->str_stream_addr, what);
|
|
|
|
__stream_log_event(_stream, EVENT_LOG_CLOSE_BY_SSL_ERROR, CONN_DIR_UPSTREAM, 0, NULL);
|
|
tfe_stream_destory(_stream);
|
|
}
|
|
|
|
struct tfe_stream * tfe_stream_create(struct tfe_proxy * pxy, struct tfe_thread_ctx * thread_ctx)
|
|
{
|
|
struct tfe_stream_private * _stream = ALLOC(struct tfe_stream_private, 1);
|
|
TFE_PROXY_STAT_INCREASE(STAT_STREAM_OPEN, 1);
|
|
|
|
_stream->thread_ref = thread_ctx;
|
|
_stream->proxy_ref = pxy;
|
|
_stream->stream_logger = pxy->logger;
|
|
|
|
unsigned int total_plugin_count = tfe_plugin_total_counts();
|
|
_stream->plugin_ctxs = ALLOC(struct plugin_ctx, total_plugin_count);
|
|
return &_stream->head;
|
|
}
|
|
|
|
void __stream_access_log_write(struct tfe_stream_private * stream)
|
|
{
|
|
const char * str_passthrough = stream->tcp_passthough ? "PASSTHROUGH" : "-";
|
|
const char * str_kill = stream->need_to_be_kill ? "KILL" : "-";
|
|
|
|
char str_log_event[TFE_STRING_MAX] = "";
|
|
unsigned int offset = 0;
|
|
|
|
/* Write event abstract log. It is used to determine which connection is broken */
|
|
for(unsigned int i = 0; i < stream->nr_log_event; i++)
|
|
{
|
|
struct tfe_stream_event_log * ev_log = &stream->log_event[i];
|
|
const char * str_dir = ev_log->dir == CONN_DIR_DOWNSTREAM ? "DOWN" : "UP";
|
|
offset += snprintf(str_log_event + offset, sizeof(str_log_event) - offset,
|
|
"%s/%s ", __str_stream_log_type(ev_log->type), str_dir);
|
|
}
|
|
|
|
MESA_handle_runtime_log(stream->stream_logger, RLOG_LV_INFO, "access",
|
|
"%d %d %d %s %s %s %s %s %s", stream->log_fd_downstream, stream->log_fd_upstream, stream->keyring_id,
|
|
stream->str_stream_addr, str_passthrough, str_kill, str_log_event,
|
|
stream->ssl_downstream_info_dump, stream->ssl_upstream_info_dump);
|
|
}
|
|
|
|
static int ev_log_to_stat_map[__EVENT_LOG_CLOSE_MAX][__CONN_DIR_MAX]{{-1}};
|
|
void __ev_log_to_stat_map_init() __attribute__((constructor, used));
|
|
void __ev_log_to_stat_map_init()
|
|
{
|
|
ev_log_to_stat_map[EVENT_LOG_CLOSE_BY_FD_PEER][CONN_DIR_DOWNSTREAM] = -1;
|
|
ev_log_to_stat_map[EVENT_LOG_CLOSE_BY_FD_EOF][CONN_DIR_DOWNSTREAM] = STAT_STREAM_CLS_DOWN_EOF;
|
|
ev_log_to_stat_map[EVENT_LOG_CLOSE_BY_FD_ERROR][CONN_DIR_DOWNSTREAM] = STAT_STREAM_CLS_DOWN_ERR;
|
|
ev_log_to_stat_map[EVENT_LOG_CLOSE_BY_SSL_ERROR][CONN_DIR_DOWNSTREAM] = STAT_STREAM_CLS_DOWN_ERR;
|
|
ev_log_to_stat_map[EVENT_LOG_CLOSE_BY_FD_PEER][CONN_DIR_UPSTREAM] = -1;
|
|
ev_log_to_stat_map[EVENT_LOG_CLOSE_BY_FD_EOF][CONN_DIR_UPSTREAM] = STAT_STREAM_CLS_UP_EOF;
|
|
ev_log_to_stat_map[EVENT_LOG_CLOSE_BY_FD_ERROR][CONN_DIR_UPSTREAM] = STAT_STREAM_CLS_UP_ERR;
|
|
ev_log_to_stat_map[EVENT_LOG_CLOSE_BY_SSL_ERROR][CONN_DIR_UPSTREAM] = STAT_STREAM_CLS_UP_ERR;
|
|
}
|
|
|
|
void __stream_close_stat(struct tfe_stream_private * stream)
|
|
{
|
|
TFE_PROXY_STAT_INCREASE(STAT_STREAM_CLS, 1);
|
|
if(stream->nr_log_event > 0)
|
|
{
|
|
struct tfe_stream_event_log * ev_log = &stream->log_event[0];
|
|
assert(ev_log_to_stat_map[ev_log->type][ev_log->dir] >= 0);
|
|
TFE_PROXY_STAT_INCREASE(ev_log_to_stat_map[ev_log->type][ev_log->dir], 1);
|
|
}
|
|
}
|
|
|
|
void tfe_stream_destory(struct tfe_stream_private * stream)
|
|
{
|
|
struct tfe_thread_ctx * thread = stream->thread_ref;
|
|
struct event_base * ev_base = thread->evbase;
|
|
|
|
__stream_access_log_write(stream);
|
|
__stream_close_stat(stream);
|
|
|
|
if (stream->head.addr)
|
|
{
|
|
FREE(&(stream->head.addr));
|
|
}
|
|
|
|
if (stream->str_stream_addr)
|
|
{
|
|
FREE(&(stream->str_stream_addr));
|
|
}
|
|
|
|
if (__is_ssl(stream) && stream->ssl_upstream)
|
|
{
|
|
evutil_socket_t __to_closed_fd = __conn_private_release_fd(stream->conn_upstream);
|
|
ssl_stream_free_and_close_fd(stream->ssl_upstream, ev_base, __to_closed_fd);
|
|
}
|
|
|
|
if (__is_ssl(stream) && stream->ssl_downstream)
|
|
{
|
|
evutil_socket_t __to_closed_fd = __conn_private_release_fd(stream->conn_downstream);
|
|
ssl_stream_free_and_close_fd(stream->ssl_downstream, ev_base, __to_closed_fd);
|
|
}
|
|
|
|
if (stream->conn_upstream)
|
|
{
|
|
__conn_private_destory(stream->conn_upstream);
|
|
}
|
|
|
|
if (stream->conn_downstream)
|
|
{
|
|
__conn_private_destory(stream->conn_downstream);
|
|
}
|
|
|
|
if (stream->defer_fd_downstream)
|
|
{
|
|
evutil_closesocket(stream->defer_fd_downstream);
|
|
}
|
|
|
|
if (stream->defer_fd_upstream)
|
|
{
|
|
evutil_closesocket(stream->defer_fd_upstream);
|
|
}
|
|
|
|
if (stream->future_downstream_create)
|
|
{
|
|
future_destroy(stream->future_downstream_create);
|
|
}
|
|
|
|
if (stream->future_upstream_create)
|
|
{
|
|
future_destroy(stream->future_upstream_create);
|
|
}
|
|
FREE(&(stream->plugin_ctxs));
|
|
|
|
tfe_proxy_thread_ctx_release(stream->thread_ref);
|
|
stream->proxy_ref = NULL;
|
|
FREE(&(stream));
|
|
}
|
|
|
|
int __fd_ttl_option_setup(struct tfe_stream_private * _stream, evutil_socket_t fd, int ttl)
|
|
{
|
|
struct sockaddr_storage sk_storage;
|
|
socklen_t sk_storage_len = sizeof(sk_storage);
|
|
|
|
if (getsockname(fd, (struct sockaddr *) &sk_storage, &sk_storage_len) < 0)
|
|
{
|
|
TFE_STREAM_LOG_ERROR(_stream, "getsockname(fd = %d) failed: %s", fd, strerror(errno));
|
|
return -1;
|
|
}
|
|
|
|
/* For IPv4, set TTL */
|
|
unsigned int __ttl = (unsigned int) ttl;
|
|
const char * __str_family = NULL;
|
|
int ret = 0;
|
|
|
|
if (sk_storage.ss_family == AF_INET)
|
|
{
|
|
ret = setsockopt(fd, IPPROTO_IP, IP_TTL, &__ttl, sizeof(__ttl));
|
|
__str_family = "AF_INET";
|
|
}
|
|
else if (sk_storage.ss_family == AF_INET6)
|
|
{
|
|
ret = setsockopt(fd, IPPROTO_IPV6, IPV6_UNICAST_HOPS, &__ttl, sizeof(__ttl));
|
|
__str_family = "AF_INET6";
|
|
}
|
|
else { assert(0); }
|
|
|
|
if (ret < 0)
|
|
{
|
|
TFE_STREAM_LOG_ERROR(_stream, "setsockopt(ttl = %u, fd = %d, family = %s) failed: %s",
|
|
__ttl, fd, __str_family, strerror(errno));
|
|
return -2;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
void __stream_fd_option_setup(struct tfe_stream_private * _stream, evutil_socket_t fd, tfe_conn_dir dir)
|
|
{
|
|
struct tfe_stream * stream = &_stream->head;
|
|
struct tfe_proxy_tcp_options * tcp_options = &_stream->proxy_ref->tcp_options;
|
|
|
|
/* Make it non-blocking */
|
|
evutil_make_socket_nonblocking(fd);
|
|
|
|
/* Recv Buffer */
|
|
if (tcp_options->sz_rcv_buffer >= 0)
|
|
{
|
|
if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, (const void *) &tcp_options->sz_rcv_buffer, sizeof(int)) == -1)
|
|
{
|
|
TFE_LOG_ERROR(g_default_logger, "%s: setsockopt(SO_RCVBUF, %d) failed, ignored: %s",
|
|
stream->str_stream_info, tcp_options->sz_rcv_buffer, strerror(errno));
|
|
}
|
|
}
|
|
|
|
/* Send Buffer */
|
|
if (tcp_options->sz_snd_buffer >= 0)
|
|
{
|
|
if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, (const void *) &tcp_options->sz_snd_buffer, sizeof(int)) == -1)
|
|
{
|
|
TFE_LOG_ERROR(g_default_logger, "%s: setsockopt(SO_SNDBUF, %d) failed, ignored: %s",
|
|
stream->str_stream_info, tcp_options->sz_snd_buffer, strerror(errno));
|
|
}
|
|
}
|
|
|
|
/* Keep-alive */
|
|
if (tcp_options->so_keepalive > 0)
|
|
{
|
|
if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, (const void *) &tcp_options->so_keepalive, sizeof(int)) == -1)
|
|
{
|
|
TFE_LOG_ERROR(g_default_logger, "%s: setsockopt(SO_KEEPALIVE, %d) failed, ignored: %s",
|
|
stream->str_stream_info, tcp_options->so_keepalive, strerror(errno));
|
|
}
|
|
}
|
|
|
|
if (tcp_options->tcp_keepcnt > 0)
|
|
{
|
|
if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, (const void *) &tcp_options->tcp_keepcnt, sizeof(int)) == -1)
|
|
{
|
|
TFE_LOG_ERROR(g_default_logger, "%s: setsockopt(TCP_KEEPCNT, %d) failed, ignored: %s",
|
|
stream->str_stream_info, tcp_options->tcp_keepcnt, strerror(errno));
|
|
}
|
|
}
|
|
|
|
if (tcp_options->tcp_keepintvl > 0)
|
|
{
|
|
if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, (const void *) &tcp_options->tcp_keepintvl, sizeof(int)) == -1)
|
|
{
|
|
TFE_LOG_ERROR(g_default_logger, "%s: setsockopt(TCP_KEEPINTVL, %d) failed, ignored: %s",
|
|
stream->str_stream_info, tcp_options->tcp_keepintvl, strerror(errno));
|
|
}
|
|
}
|
|
|
|
if (tcp_options->tcp_keepidle > 0)
|
|
{
|
|
if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE, (const void *) &tcp_options->tcp_keepidle, sizeof(int)) == -1)
|
|
{
|
|
TFE_LOG_ERROR(g_default_logger, "%s: setsockopt(TCP_KEEPIDLE, %d) failed, ignored: %s",
|
|
stream->str_stream_info, tcp_options->tcp_keepidle, strerror(errno));
|
|
}
|
|
}
|
|
|
|
if (tcp_options->tcp_user_timeout > 0)
|
|
{
|
|
if (setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, (const void *) &tcp_options->tcp_user_timeout, sizeof(int))
|
|
== -1)
|
|
{
|
|
TFE_LOG_ERROR(g_default_logger, "%s: setsockopt(TCP_USER_TIMEOUT, %d) failed, ignored: %s",
|
|
stream->str_stream_info, tcp_options->tcp_user_timeout, strerror(errno));
|
|
}
|
|
}
|
|
|
|
int __ttl = (dir == CONN_DIR_UPSTREAM) ? tcp_options->tcp_ttl_upstream : tcp_options->tcp_ttl_downstream;
|
|
if (__ttl > 0 && __fd_ttl_option_setup(_stream, fd, __ttl) < 0)
|
|
{
|
|
TFE_LOG_ERROR(g_default_logger, "%s: Failed at setup FD's ttl option, ttl = %d, fd = %d",
|
|
stream->str_stream_info, __ttl, fd);
|
|
}
|
|
}
|
|
|
|
int tfe_stream_init_by_fds(struct tfe_stream * stream, evutil_socket_t fd_downstream, evutil_socket_t fd_upstream)
|
|
{
|
|
struct tfe_stream_private * _stream = container_of(stream, struct tfe_stream_private, head);
|
|
|
|
_stream->defer_fd_downstream = fd_downstream;
|
|
_stream->defer_fd_upstream = fd_upstream;
|
|
_stream->log_fd_downstream = fd_downstream;
|
|
_stream->log_fd_upstream = fd_upstream;
|
|
|
|
__stream_fd_option_setup(_stream, fd_downstream, CONN_DIR_DOWNSTREAM);
|
|
__stream_fd_option_setup(_stream, fd_upstream, CONN_DIR_UPSTREAM);
|
|
|
|
_stream->head.addr = tfe_stream_addr_create_by_fd(fd_downstream, CONN_DIR_DOWNSTREAM);
|
|
if (unlikely(_stream->head.addr == NULL))
|
|
{
|
|
TFE_LOG_ERROR(_stream->stream_logger, "Failed to create address from fd %d, %d, terminate fds.",
|
|
fd_downstream, fd_upstream);
|
|
goto __errout;
|
|
}
|
|
|
|
_stream->str_stream_addr = tfe_stream_addr_to_str(_stream->head.addr);
|
|
stream->str_stream_info = _stream->str_stream_addr;
|
|
|
|
if (_stream->session_type == STREAM_PROTO_PLAIN)
|
|
{
|
|
_stream->conn_downstream = __conn_private_create_by_fd(_stream, fd_downstream);
|
|
_stream->conn_upstream = __conn_private_create_by_fd(_stream, fd_upstream);
|
|
|
|
/* Defer FD has been transfer to conn_downstream/conn_upstream */
|
|
_stream->defer_fd_downstream = 0;
|
|
_stream->defer_fd_upstream = 0;
|
|
|
|
if (unlikely(_stream->conn_downstream == NULL || _stream->conn_upstream == NULL))
|
|
{
|
|
goto __errout;
|
|
}
|
|
|
|
assert(_stream->conn_downstream != NULL);
|
|
assert(_stream->conn_upstream != NULL);
|
|
|
|
__conn_private_enable(_stream->conn_downstream);
|
|
__conn_private_enable(_stream->conn_upstream);
|
|
|
|
TFE_PROXY_STAT_INCREASE(STAT_STREAM_TCP_PLAIN, 1);
|
|
}
|
|
|
|
if (_stream->session_type == STREAM_PROTO_SSL)
|
|
{
|
|
_stream->ssl_mgr = _stream->proxy_ref->ssl_mgr_handler;
|
|
|
|
_stream->future_upstream_create = future_create("ssl_up",
|
|
ssl_upstream_create_on_success, ssl_upstream_create_on_fail, (void *) _stream);
|
|
|
|
/* Defer setup conn_downstream & conn_upstream in async callbacks. */
|
|
ssl_async_upstream_create(_stream->future_upstream_create,
|
|
_stream->ssl_mgr, fd_upstream, fd_downstream, _stream->thread_ref->thread_id);
|
|
|
|
TFE_PROXY_STAT_INCREASE(STAT_STREAM_TCP_SSL, 1);
|
|
}
|
|
|
|
return 0;
|
|
|
|
__errout:
|
|
return -1;
|
|
}
|
|
|
|
int tfe_stream_option_set(struct tfe_stream * stream, enum tfe_stream_option opt, const void * arg, size_t sz_arg)
|
|
{
|
|
struct tfe_stream_private * _stream = container_of(stream, struct tfe_stream_private, head);
|
|
|
|
if (opt == TFE_STREAM_OPT_SESSION_TYPE)
|
|
{
|
|
assert(sz_arg == sizeof(enum tfe_stream_proto));
|
|
_stream->session_type = *(enum tfe_stream_proto *) arg;
|
|
}
|
|
else if (opt == TFE_STREAM_OPT_PASSTHROUGH)
|
|
{
|
|
assert(sz_arg == sizeof(bool));
|
|
_stream->tcp_passthough = *(bool *) arg;
|
|
}
|
|
else if (opt == TFE_STREAM_OPT_KEYRING_ID)
|
|
{
|
|
assert(sz_arg == sizeof(unsigned int));
|
|
_stream->keyring_id = *(unsigned int *) arg;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
void tfe_stream_write_access_log(const struct tfe_stream * stream, int level, const char * fmt, ...)
|
|
{
|
|
va_list arg_ptr;
|
|
va_start(arg_ptr, fmt);
|
|
struct tfe_stream_private * _stream = container_of(stream, struct tfe_stream_private, head);
|
|
|
|
/* Format input content */
|
|
char * __tmp_buffer;
|
|
vasprintf(&__tmp_buffer, fmt, arg_ptr);
|
|
|
|
/* Log content with stream tag */
|
|
MESA_handle_runtime_log(_stream->stream_logger, level, "access", "%s %s", _stream->str_stream_addr, __tmp_buffer);
|
|
free(__tmp_buffer);
|
|
}
|
|
|
|
|
|
int tfe_stream_shutdown(const struct tfe_stream * stream)
|
|
{
|
|
return 0;
|
|
}
|
|
|
|
int tfe_stream_shutdown_dir(const struct tfe_stream * stream, enum tfe_conn_dir dir)
|
|
{
|
|
return 0;
|
|
}
|
|
|
|
void tfe_stream_kill(const struct tfe_stream * stream)
|
|
{
|
|
struct tfe_stream_private * _stream = to_stream_private(stream);
|
|
_stream->need_to_be_kill = true;
|
|
}
|