gfwleak/tango-tfe
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'feature-tcp-forward' into 'develop-tfe3a'

Browse Source 
Merge of feature-tcp-forward to develop-tfe3a

See merge request tango/tfe!3
This commit is contained in:
陆秋文
2018-08-23 13:57:09 +08:00
parent 43aa9529a4 405f046b22
commit 7995af9c77
13 changed files with 460 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
common/CMakeLists.txt
View File

@@ -1,3 +1,4 @@
add_library(common src/tfe_stat.cpp src/tfe_utils.cpp)
target_include_directories(common PUBLIC ${CMAKE_CURRENT_LIST_DIR}/include)
target_link_libraries(common MESA_handle_logger)

1
common/include/tfe_future.h
View File

@@ -1,3 +1,4 @@
#pragma once
enum e_future_error
{

5
common/include/tfe_stream.h
View File

@@ -3,9 +3,7 @@
#include <stddef.h>
#include <stdint.h>
#include <tfe_types.h>
#define TFE_STRING_MAX 2048
#define TFE_SYMBOL_MAX 64
#include <tfe_utils.h>
enum tfe_session_proto
{
@@ -46,7 +44,6 @@ struct tfe_conn
struct tfe_stream
{
struct tfe_conn upstream;
struct tfe_conn downstream;
};

33
common/include/tfe_utils.h
View File

@@ -1,17 +1,32 @@
//#define ALLOC(t,n) (t *)calloc(sizeof(t),(n))
/*
* @file tfe_utils.h
* This file provides common usage marcos and helper functions.
*/
/* Allocates an array of objects using malloc() */
#define ALLOC(type, number) \
((type *)calloc(sizeof(type), number))
#pragma once
#include <MESA/MESA_handle_logger.h>
#define TFE_STRING_MAX 2048
#define TFE_SYMBOL_MAX 64
#ifndef TFE_CONFIG_BACKLOG_DEFAULT
#define TFE_CONFIG_BACKLOG_DEFAULT 20
#endif
#define ALLOC(type, number) ((type *)calloc(sizeof(type), number))
#define likely(expr) __builtin_expect((expr), 1)
#define unlikely(expr) __builtin_expect((expr), 0)
#define TFE_LOG_ERROR(handler, fmt, ...) \
do { MESA_handle_runtime_log(handler, RLOG_LV_FATAL, NULL, fmt, ##__VA_ARGS__); } while(0) \
#define TFE_LOG_INFO(handler, fmt, ...) \
do { MESA_handle_runtime_log(handler, RLOG_LV_INFO, NULL, fmt, ##__VA_ARGS__); } while(0) \
#define TFE_LOG_DEBUG(handler, fmt, ...) \
do { MESA_handle_runtime_log(handler, RLOG_LV_DEBUG, NULL, fmt, ##__VA_ARGS__); } while(0) \
int addr_sock_to_layer(struct sockaddr * sock_addr, int sockaddrlen, struct layer_addr * layer_addr);
int addr_layer_to_sock(struct layer_addr * layer_addr, struct sockaddr * sock_addr);
char* tfe_strdup(const char* s);
#define TFE_LOG_ERROR
#define TFE_LOG_INFO
#define TFE_LOG_DEBUG

1
platform/CMakeLists.txt
View File

@@ -1,4 +1,3 @@
add_executable(tfe src/cert.cpp src/future.cpp src/kni.cpp src/tfe_stream.cpp src/main.cpp src/proxy.cpp)
target_include_directories(tfe PUBLIC ${CMAKE_CURRENT_LIST_DIR}/include/external)
target_include_directories(tfe PRIVATE ${CMAKE_CURRENT_LIST_DIR}/include/internal)

7
platform/include/internal/kni.h
View File

@@ -1,2 +1,7 @@
void* io_kni_init(const char* unix_domain_path, struct event_base * attach);
#pragma once
struct tfe_proxy;
struct kni_acceptor_ctx;
struct kni_acceptor_ctx * kni_acceptor_init(struct tfe_proxy *proxy, const char *profile, void *logger);
void kni_acceptor_deinit(struct kni_acceptor_ctx *ctx);

19
platform/include/internal/proxy.h Normal file
View File

@@ -0,0 +1,19 @@
#pragma once
#include <tfe_stream.h>
#include <event2/event.h>
struct tfe_proxy;
struct tfe_proxy_accept_para
{
/* Both upstream and downstream FDs */
evutil_socket_t upstream_fd;
evutil_socket_t downstream_fd;
/* Session Type */
enum tfe_session_proto session_type;
};
struct tfe_proxy * tfe_proxy_new(const char * profile);
int tfe_proxy_fds_accept(struct tfe_proxy * ctx, const struct tfe_proxy_accept_para * para);
void tfe_proxy_run(struct tfe_proxy * proxy);

9
platform/include/internal/ssl_stream.h
View File

@@ -1,5 +1,7 @@
#pragma once
#include "tfe_future.h"
#include <event2/event.h>
#include <tfe_future.h>
struct ssl_client_hello
{
@@ -7,11 +9,14 @@ struct ssl_client_hello
char* sni;
char* cipher_suites;
};
struct ssl_client_hello* ssl_get_peek_result(future_result_t* result);
void ssl_free_peek_result(struct ssl_client_hello* client_hello);
void ssl_async_peek_client_hello(struct future* future, evutil_socket_t fd, struct event_base *evbase);
void ssl_async_connect_origin(struct future* future, const struct ssl_client_hello* client_hello, evutil_socket_t fd, const char* sni, struct event_base *evbase);
void ssl_async_connect_origin(struct future* future, const struct ssl_client_hello* client_hello,
evutil_socket_t fd, const char* sni, struct event_base *evbase);
struct ssl_downstream * ssl_downstream_create();
void ssl_upstream_free(struct ssl_upstream * p);

5
platform/include/internal/stream.h
View File

@@ -78,6 +78,8 @@ struct tfe_conn_private
struct tfe_stream_private
{
struct tfe_stream head;
struct tfe_proxy *proxy;
enum tfe_session_proto session_type;
struct tfe_conn_private conn_upstream;
struct tfe_conn_private conn_downstream;
@@ -96,11 +98,14 @@ struct tfe_stream_private
uint8_t is_plugin_opened;
int calling_idx;
size_t forward_bytes;
size_t defere_bytes;
size_t drop_bytes;
enum tfe_app_proto app_proto;
int plugin_num;
struct plugin_ctx * plug_ctx;
unsigned char passthrough; /* 1 if SSL passthrough is active */

295
platform/src/kni.cpp
View File

@@ -1,5 +1,298 @@
#include <stdio.h>
void* io_kni_init(const char* unix_domain_path, struct event_base * attach)
#include <stdlib.h>
#include <sys/types.h>
#include <sys/un.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/errno.h>
#include <MESA/MESA_prof_load.h>
#include <tfe_stream.h>
#include <evutil.h>
#include <event2/listener.h>
#include <pthread.h>
#include <assert.h>
#include <proxy.h>
#include <kni.h>
#ifndef TFE_CONFIG_KNI_UXDOMAIN_PATH_DEFAULT
#define TFE_CONFIG_KNI_UXDOMAIN_PATH_DEFAULT "/var/run/.tfe_kni_acceptor_handler"
#endif
enum KNI_TLV_TYPE
{
KNI_TLV_TYPE_PRO = 0x01,
};
enum KNI_TLV_VALUE
{
KNI_TLV_VALUE_HTTP = 0x01,
KNI_TLV_VALUE_SSL = 0x02,
};
struct kni_tlv_info
{
char type;
short len;
char value;
};
struct kni_acceptor_ctx
{
/* INPUT */
struct tfe_proxy * proxy;
const char * profile;
void * logger;
/* CONFIG */
char str_unixdomain_file[TFE_STRING_MAX];
/* PERSIST RUNTIME RESOURCE */
int fd_unixdomain;
struct event_base * ev_base;
struct evconnlistener * ev_listener;
pthread_t thread;
/* PEER CONNECTION RESOUCE
* should close by __kni_conn_close() */
struct event * ev_kni_conn;
int fd_kni_conn;
pid_t pid_kni_conn;
};
void __kni_conn_close(struct kni_acceptor_ctx * ctx)
{
if (ctx->fd_kni_conn != 0) close(ctx->fd_kni_conn);
if (ctx->ev_kni_conn != NULL) event_free(ctx->ev_kni_conn);
if (ctx->pid_kni_conn != 0) ctx->pid_kni_conn = 0;
}
void __kni_event_cb(evutil_socket_t fd, short what, void * user)
{
struct kni_acceptor_ctx * __ctx = (struct kni_acceptor_ctx *)user;
struct cmsghdr * __cmsghdr;
struct tfe_proxy_accept_para __accept_para;
int * __fds = NULL;
assert(__ctx != NULL && __ctx->thread == pthread_self());
assert(what & EV_READ);
/* We use IOVEC to recieve the fds make by KNI.
* This is a kind of magic skill to share socket fds between two(or more) process.
* http://man7.org/tlpi/code/online/dist/sockets/scm_rights_send.c.html
*/
constexpr static int __TRANS_FDS_MAX = 2;
constexpr static int __CONTROLLEN = CMSG_SPACE(__TRANS_FDS_MAX * sizeof(int));
char __buffer[512] = {0};
struct kni_tlv_info * __tlv_info = (struct kni_tlv_info *)(__buffer);
struct iovec __iovec[1];
struct msghdr __msghdr;
char __cmptr[__CONTROLLEN];
__iovec[0].iov_base = __buffer;
__iovec[0].iov_len = sizeof(__buffer);
__msghdr.msg_iov = __iovec;
__msghdr.msg_iovlen = 1;
__msghdr.msg_name = NULL;
__msghdr.msg_namelen = 0;
__msghdr.msg_control = (void *)(__cmptr);
__msghdr.msg_controllen = (size_t)__CONTROLLEN;
ssize_t rd = recvmsg(fd, &__msghdr, 0);
if (rd == -1 && (errno == EINTR || errno == EAGAIN))
{
return;
}
else if (rd < 0)
{
TFE_LOG_ERROR(__ctx->logger, "Failed at recving fds from KNI connection: %s", strerror(errno));
goto __close_kni_connection;
}
else if (rd == 0)
{
TFE_LOG_INFO(__ctx->logger, "KNI connected from process %u", __ctx->pid_kni_conn);
goto __close_kni_connection;
}
__cmsghdr = CMSG_FIRSTHDR(&__msghdr);
__fds = (int *) (CMSG_DATA(__cmsghdr));
assert(__tlv_info->type == KNI_TLV_TYPE_PRO);
enum tfe_session_proto __session_proto;
if (__tlv_info->value == KNI_TLV_VALUE_HTTP)
{
__session_proto = SESSION_PROTO_PLAIN;
}
else if (__tlv_info->value == KNI_TLV_VALUE_SSL)
{
__session_proto = SESSION_PROTO_SSL;
}
else
{
assert(0);
goto __close_kni_connection;
}
__accept_para.session_type = __session_proto;
__accept_para.downstream_fd = __fds[0];
__accept_para.upstream_fd = __fds[1];
if (tfe_proxy_fds_accept(__ctx->proxy, &__accept_para) < 0)
{
goto __drop_recieved_fds;
}
return;
__close_kni_connection:
__kni_conn_close(__ctx);
__drop_recieved_fds:
evutil_closesocket(__fds[0]);
evutil_closesocket(__fds[1]);
}
void __kni_listener_accept_cb(struct evconnlistener * listener, evutil_socket_t fd,
struct sockaddr * sk_addr, int sk_len, void * user)
{
struct kni_acceptor_ctx * __ctx = (struct kni_acceptor_ctx *)user;
struct event * __event = NULL;
struct ucred __cr{};
socklen_t __cr_len = sizeof(struct ucred);
/* There is only one KNI process can connect to TFE.
* If ev_kni_conn is not NULL, there's already a KNI connected to TFE.
* We need to refuse this connection
*/
if (unlikely(__ctx->ev_kni_conn != NULL))
{
TFE_LOG_ERROR(__ctx->logger, "One KNI(PID = %d) has been connected to our program, "
"close this connection", __ctx->pid_kni_conn);
goto __close_this_connection;
}
/* Get Peer's PID */
if (getsockopt(fd, SOL_SOCKET, SO_PEERCRED, (void *)&__cr, &__cr_len) < 0)
{
TFE_LOG_ERROR(__ctx->logger, "Failed at getsockopt(SO_PEERCRED) for fd %d, close this connection");
goto __close_this_connection;
}
__event = event_new(__ctx->ev_base, fd, EV_READ | EV_PERSIST, __kni_event_cb, __ctx);
if (unlikely(__event == NULL))
{
TFE_LOG_ERROR(__ctx->logger, "Failed at creating event, close this connection.");
goto __close_this_connection;
}
__ctx->fd_kni_conn = fd;
__ctx->ev_kni_conn = __event;
__ctx->pid_kni_conn = __cr.pid;
TFE_LOG_INFO(__ctx->logger, "KNI connected from process %u", __ctx->pid_kni_conn);
return;
__close_this_connection:
__kni_conn_close(__ctx);
}
void * __kni_listener_thread_entry(void * args)
{
struct kni_acceptor_ctx * __ctx = (struct kni_acceptor_ctx *)args;
assert(__ctx != NULL && __ctx->thread == pthread_self());
TFE_LOG_DEBUG(__ctx->logger, "Starting KNI listener thread...");
event_base_dispatch(__ctx->ev_base);
TFE_LOG_DEBUG(__ctx->logger, "Stoping KNI listener thread...");
return (void *)NULL;
}
void kni_acceptor_deinit(struct kni_acceptor_ctx *ctx)
{
if (ctx != NULL && ctx->ev_listener != NULL)
{
evconnlistener_free(ctx->ev_listener);
}
if (ctx != NULL && ctx->ev_base != NULL)
{
event_base_free(ctx->ev_base);
}
if (ctx != NULL && ctx->fd_unixdomain != 0)
{
close(ctx->fd_unixdomain);
}
if (ctx != NULL)
{
free(ctx);
}
return;
}
struct kni_acceptor_ctx * kni_acceptor_init(struct tfe_proxy *proxy, const char *profile, void *logger)
{
struct kni_acceptor_ctx * __ctx = ALLOC(struct kni_acceptor_ctx, 1);
struct sockaddr_un __sockaddr_un;
int ret = 0;
__ctx->profile = profile;
__ctx->logger = logger;
/* Read the unix domain socket file, this file is used to recieve fds from KNI */
MESA_load_profile_string_def(profile, "kni", "uxdomain", __ctx->str_unixdomain_file,
sizeof(__ctx->str_unixdomain_file), TFE_CONFIG_KNI_UXDOMAIN_PATH_DEFAULT);
if (unlikely(unlink(__ctx->str_unixdomain_file) < 0))
{
TFE_LOG_ERROR(__ctx->logger, "Failed at unlink undomain file %s: %s",
__ctx->str_unixdomain_file, strerror(errno)); goto __errout;
}
__sockaddr_un.sun_family = AF_UNIX;
strncpy(__sockaddr_un.sun_path, __ctx->str_unixdomain_file, sizeof(__sockaddr_un.sun_path));
/* Create new event base, this event base will be dispatched at separated thread */
__ctx->ev_base = event_base_new();
if (unlikely(__ctx->ev_base == NULL))
{
TFE_LOG_ERROR(__ctx->logger, "Failed at creating event_base. ");
goto __errout;
}
/* Create a listener */
__ctx->ev_listener = evconnlistener_new_bind(__ctx->ev_base, __kni_listener_accept_cb, __ctx, 0,
TFE_CONFIG_BACKLOG_DEFAULT, (struct sockaddr *)(&__sockaddr_un), sizeof(__sockaddr_un));
if (unlikely(__ctx->ev_listener == NULL))
{
TFE_LOG_ERROR(__ctx->logger, "Failed at creating evconnlistener.");
goto __errout;
}
/* Create a thread to dispatch ctx->evbase */
ret = pthread_create(&__ctx->thread, NULL, __kni_listener_thread_entry, (void *)__ctx);
if (ret < 0)
{
TFE_LOG_ERROR(__ctx->logger, "Failed at creating listener thread: %s", strerror(errno));
goto __errout;
}
TFE_LOG_INFO(__ctx->logger, "KNI UNIXDOMAIN FILE: %s", __ctx->str_unixdomain_file);
TFE_LOG_INFO(__ctx->logger, "KNI LISTENER FD: %d", __ctx->fd_unixdomain);
return __ctx;
__errout:
kni_acceptor_deinit(__ctx);
return NULL;
}

9
platform/src/main.cpp
View File

@@ -13,12 +13,3 @@
#include <MESA/MESA_handle_logger.h>
#include <MESA/MESA_prof_load.h>
#include <MESA/wired_cfg.h>
extern struct tfe_instance __g_tfe_instance;
extern struct tfe_config __g_tfe_config;
struct tfe_instance* g_tfe_instance = &__g_tfe_instance;
struct tfe_config * g_tfe_cfg = &__g_tfe_config;
const char* module_name="TFE";

117
platform/src/proxy.cpp
View File

@@ -1,3 +1,6 @@
/*
* Proxy engine, built around libevent 2.x.
*/
#include <sys/types.h>
#include <sys/socket.h>
@@ -22,48 +25,9 @@
#include <tfe_utils.h>
#include <tfe_stream.h>
#include <stream.h>
#include <kni.h>
#include <proxy.h>
#include <sescache.h>
/*
* Proxy engine, built around libevent 2.x.
*/
#define TFE_BACKLOG_DEFAULT 20
const char * module_name_pxy = "TFE_PXY";
extern struct tfe_instance * g_tfe_instance;
__thread int __currect_thread_id;
static void __dummy_event_handler(evutil_socket_t fd, short what, void * arg)
{
return;
}
/*
* Thread entry point; runs the event loop of the event base.
* Does not exit until the libevent loop is broken explicitly.
*/
static void * __tfe_thrmgr_thread_entry(void * arg)
{
struct tfe_thread_ctx * ctx = (struct tfe_thread_ctx *) arg;
struct timeval timer_delay = {60, 0};
struct event * ev;
ev = event_new(ctx->evbase, -1, EV_PERSIST, __dummy_event_handler, NULL);
if (!ev) return (void *)NULL;
evtimer_add(ev, &timer_delay);
ctx->running = 1;
__currect_thread_id = ctx->thread_id;
event_base_dispatch(ctx->evbase);
event_free(ev);
return (void *)NULL;
}
#include <kni.h>
static int signals[] = {SIGTERM, SIGQUIT, SIGHUP, SIGINT, SIGPIPE, SIGUSR1};
@@ -89,9 +53,41 @@ struct tfe_proxy
void * io_mod;
};
const char * module_name_pxy = "TFE_PXY";
extern struct tfe_instance * g_tfe_instance;
__thread int __currect_thread_id;
static void __dummy_event_handler(evutil_socket_t fd, short what, void * arg)
{
return;
}
/*
* Signal handler for SIGTERM, SIGQUIT, SIGINT, SIGHUP, SIGPIPE and SIGUSR1.
* Thread entry point; runs the event loop of the event base.
* Does not exit until the libevent loop is broken explicitly.
*/
static void * __tfe_thrmgr_thread_entry(void * arg)
{
struct tfe_thread_ctx * ctx = (struct tfe_thread_ctx *) arg;
struct timeval timer_delay = {60, 0};
struct event * ev;
ev = event_new(ctx->evbase, -1, EV_PERSIST, __dummy_event_handler, NULL);
if (!ev) return (void *)NULL;
evtimer_add(ev, &timer_delay);
ctx->running = 1;
__currect_thread_id = ctx->thread_id;
event_base_dispatch(ctx->evbase);
event_free(ev);
return (void *)NULL;
}
static void proxy_signal_cb(evutil_socket_t fd, short what, void * arg)
{
tfe_proxy * ctx = (tfe_proxy *) arg;
@@ -105,10 +101,10 @@ static void proxy_signal_cb(evutil_socket_t fd, short what, void * arg)
case SIGUSR1:
break;
case SIGPIPE:
TFE_LOG_ERROR("Warning: Received SIGPIPE; ignoring.\n");
TFE_LOG_ERROR(ctx->main_logger, "Warning: Received SIGPIPE; ignoring.\n");
break;
default:
TFE_LOG_ERROR("Warning: Received unexpected signal %i\n", fd);
TFE_LOG_ERROR(ctx->main_logger, "Warning: Received unexpected signal %i\n", fd);
break;
}
}
@@ -140,19 +136,20 @@ unsigned int select_work_thread(struct tfe_proxy * pxy)
/*
* Callback for accept events on the socket listener bufferevent.
*/
static void io_mod_accept_cb(evutil_socket_t upstream_fd, evutil_socket_t downstream_fd,
enum tfe_session_proto session_type, void * arg)
int tfe_proxy_fds_accept(struct tfe_proxy * ctx, const struct tfe_proxy_accept_para * para)
{
struct tfe_proxy * pxy = (struct tfe_proxy *) arg;
unsigned int worker_tid = select_work_thread(ctx);
tfe_thread_ctx * worker_thread_ctx = &ctx->work_threads[worker_tid];
unsigned int worker_tid = select_work_thread(pxy);
tfe_thread_ctx * worker_thread_ctx = &pxy->work_threads[worker_tid];
struct tfe_stream_private * stream = tfe_stream_create(para->upstream_fd,
para->downstream_fd, para->session_type, worker_thread_ctx);
struct tfe_stream_private * stream = tfe_stream_create(upstream_fd,
downstream_fd, session_type, worker_thread_ctx);
if (stream == NULL) goto __errout;
tfe_stream_setup(stream);
assert(stream != NULL);
return tfe_stream_setup(stream);
__errout:
return -1;
}
/*
@@ -160,7 +157,7 @@ static void io_mod_accept_cb(evutil_socket_t upstream_fd, evutil_socket_t downst
* Socket clisock is the privsep client socket used for binding to ports.
* Returns ctx on success, or NULL on error.
*/
struct tfe_proxy * tfe_proxy_new(tfe_config * cfg)
struct tfe_proxy * tfe_proxy_new(const char * profile)
{
struct tfe_proxy * proxy = ALLOC(struct tfe_proxy, 1);
assert(proxy != NULL);
@@ -182,6 +179,7 @@ struct tfe_proxy * tfe_proxy_new(tfe_config * cfg)
proxy->modules[1].proto = APP_PROTO_HTTP2;
proxy->work_threads = ALLOC(struct tfe_thread_ctx, proxy->nr_work_threads);
proxy->io_mod = kni_acceptor_init(proxy, profile, NULL);
for (unsigned int i = 0; i < proxy->nr_work_threads; i++)
{
@@ -202,8 +200,7 @@ struct tfe_proxy * tfe_proxy_new(tfe_config * cfg)
}
proxy->gcev = event_new(proxy->evbase, -1, EV_PERSIST, proxy_gc_cb, proxy);
if (!proxy->gcev)
goto error_out;
if (!proxy->gcev) goto error_out;
evtimer_add(proxy->gcev, &gc_delay);
return proxy;
@@ -271,8 +268,8 @@ void proxy_loopbreak(tfe_proxy * ctx)
*/
void proxy_free(tfe_proxy * ctx)
{
}
}
int main(int argc, char *argv[])
{
@@ -284,11 +281,7 @@ int main(int argc, char *argv[])
//TODO: Initiate Local Cert Cache, Decryption Mirror, Field Stat, Logger and etc.
//NOTICE: Maat, Cert Store,Tango Cache are initiated in bussiness plugin.
#if 0
g_tfe_instance=ALLOC(struct tfe_instance,1);
proxy=tfe_proxy_new(g_tfe_cfg);
#endif
proxy=tfe_proxy_new(main_profile);
tfe_proxy_run(proxy);
proxy_free(proxy);
}

62
platform/src/tfe_stream.cpp
View File

@@ -23,6 +23,8 @@
#include <tfe_stream.h>
#include <tfe_utils.h>
#include <tfe_future.h>
#include <ssl_stream.h>
#include <stream.h>
#include <cert.h>
@@ -481,29 +483,59 @@ void peek_client_hello_on_fail(enum e_future_error err, const char * what, void
assert(0);
}
/*
* Callback for accept events on the socket listener bufferevent.
* Called when a new incoming connection has been accepted.
* Initiates the connection to the server. The incoming connection
* from the client is not being activated until we have a successful
* connection to the server, because we need the server's certificate
* in order to set up the SSL session to the client.
* For consistency, plain TCP works the same way, even if we could
* start reading from the client while waiting on the connection to
* the server to connect.
*/
int ssl_stream_setup(struct tfe_stream_private * _stream)
{
return 0;
}
int __plain_stream_conn_private_init(struct tfe_stream_private * _stream,
struct tfe_conn_private * _conn, evutil_socket_t fd)
{
struct tfe_proxy * proxy = _stream->proxy;
struct event_base * ev_base = _stream->thrmgr_ref->evbase;
_conn->bev = bufferevent_socket_new(ev_base, fd, BEV_OPT_DEFER_CALLBACKS)
_conn->fd = fd;
_conn->closed = 0;
_conn->need_shutdown = 0;
_conn->on_writing = 0;
}
int plain_stream_setup(struct tfe_stream_private * _stream)
{
return 0;
}
void tfe_stream_setup(struct tfe_stream_private * _stream)
{
struct future * f_sni = NULL;
tfe_thread_ctx * thread = _stream->thrmgr_ref;
if (_stream->session_type == SESSION_PROTO_SSL)
{
_stream->ssl_downstream = ssl_downstream_create();
_stream->async_future = future_create(peek_client_hello_on_succ, peek_client_hello_on_fail, _stream);
ssl_async_peek_client_hello(_stream->ssl_downstream->future_sni_peek, _stream->fd_downstream,
_stream->thrmgr_ref->evbase);
}
else if (_stream->session_type == SESSION_PROTO_PLAIN)
{
bufferevent_setcb(_stream->head.upstream.bev, tfe_stream_readcb, tfe_stream_writecb, tfe_stream_eventcb, _stream);
bufferevent_setcb(_stream->head.downstream.bev, tfe_stream_readcb, tfe_stream_writecb, tfe_stream_eventcb, _stream);
bufferevent_enable(_stream->head.upstream.bev, EV_READ | EV_WRITE);
bufferevent_enable(_stream->head.downstream.bev, EV_READ | EV_WRITE);
}
else
{
}
switch (_stream->session_type)
{
case SESSION_PROTO_SSL:
// for SSL, defer dst connection setup to initial_readcb
_stream->ssl_downstream = ssl_downstream_create();
_stream->async_future = future_create(peek_client_hello_on_succ, peek_client_hello_on_fail, _stream);
ssl_async_peek_client_hello(_stream->ssl_downstream->future_sni_peek, _stream->fd_downstream,
_stream->thrmgr_ref->evbase);
thread->stat.value[SSL_NUM]++;
break;
default: