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

TSG-22699 适配控制报文修改

Browse Source
This commit is contained in:
wangmenglan
2024-09-23 18:35:47 +08:00
committed by luwenpeng
parent a571c85b47
commit 707b418250
21 changed files with 325 additions and 265 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
common/src/tfe_packet_io.cpp
View File

@@ -156,7 +156,7 @@ struct packet_identify
extern int tcp_policy_enforce(struct tcp_policy_enforcer *tcp_enforcer, struct tfe_cmsg *cmsg);
extern int tfe_proxy_fds_accept(struct tfe_proxy * ctx, int fd_downstream, int fd_upstream, int fd_fake_c, int fd_fake_s, struct tfe_cmsg * cmsg);
extern void chaining_policy_enforce(struct chaining_policy_enforcer *enforcer, struct tfe_cmsg *cmsg, uint64_t rule_id);
extern void chaining_policy_enforce(struct chaining_policy_enforcer *enforcer, struct tfe_cmsg *cmsg, uuid_t rule_id);
/******************************************************************************
* dup packet filter
@@ -821,7 +821,7 @@ static void send_event_log(struct session_ctx *s_ctx, int thread_seq, void *ctx)
int ret = 0;
int do_log = 0;
uint8_t hit_no_intercept = 0;
uint64_t rule_id = 0;
uuid_t rule_id;
uint16_t length = 0;
uint8_t ssl_intercept_status = 0;
uint64_t ssl_upstream_latency = 0;
@@ -844,7 +844,7 @@ static void send_event_log(struct session_ctx *s_ctx, int thread_seq, void *ctx)
ret = tfe_cmsg_get_value(s_ctx->cmsg, TFE_CMSG_POLICY_DO_LOG, (unsigned char *)&do_log, sizeof(do_log), &length);
if (ret < 0 || do_log == 0)
return;
ret = tfe_cmsg_get_value(s_ctx->cmsg, TFE_CMSG_POLICY_ID, (unsigned char *)&rule_id, sizeof(rule_id), &length);
ret = tfe_cmsg_get_value(s_ctx->cmsg, TFE_CMSG_POLICY_ID, (unsigned char *)rule_id, UUID_LEN, &length);
if (ret < 0)
return;
ret = tfe_cmsg_get_value(s_ctx->cmsg, TFE_CMSG_HIT_NO_INTERCEPT, (unsigned char *)&hit_no_intercept, sizeof(hit_no_intercept), &length);
@@ -881,7 +881,7 @@ static void send_event_log(struct session_ctx *s_ctx, int thread_seq, void *ctx)
// proxy rule list
mpack_build_array(&writer);
mpack_write_u64(&writer, rule_id);
mpack_write_bin(&writer, (const char*)rule_id, UUID_LEN);
mpack_complete_array(&writer);
tfe_cmsg_get_value(s_ctx->cmsg, TFE_CMSG_SSL_INTERCEPT_STATE, (unsigned char *)&ssl_intercept_status, sizeof(ssl_intercept_status), &length);
@@ -1127,22 +1127,24 @@ static int handle_session_opening(struct metadata *meta, marsio_buff_t *rx_buff,
packet_get_innermost_tuple4(&pkt, &inner_tuple4);
tfe_cmsg_get_value(parser->cmsg, TFE_CMSG_TCP_RESTORE_PROTOCOL, (unsigned char *)&stream_protocol_in_char, sizeof(stream_protocol_in_char), &size);
uint64_t rule_id = 0;
ret = intercept_policy_select(thread->ref_proxy->int_ply_enforcer, parser->tfe_policy_ids, parser->tfe_policy_id_num, &rule_id);
uuid_t rule_id = {0};
char str_rule_id[UUID_STR_LEN] = {0};
ret = intercept_policy_select(thread->ref_proxy->int_ply_enforcer, parser->tfe_policy_ids, parser->tfe_policy_id_num, rule_id);
uuid_unparse(rule_id, str_rule_id);
if (ret != 0)
{
is_passthrough = 1;
set_passthrough_reason(parser->cmsg, reason_invalid_intercept_param);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, 0, meta->session_id, "active", "passthrough", "invalid intercept param");
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_rule_id, meta->session_id, "active", "passthrough", "invalid intercept param");
goto passthrough;
}
tfe_cmsg_set(parser->cmsg, TFE_CMSG_POLICY_ID, (const unsigned char *)&rule_id, sizeof(uint64_t));
tfe_cmsg_set(parser->cmsg, TFE_CMSG_POLICY_ID, (const unsigned char *)rule_id, UUID_LEN);
ret = intercept_policy_enforce(thread->ref_proxy->int_ply_enforcer, parser->cmsg);
if (ret != 0) {
is_passthrough = 1;
set_passthrough_reason(parser->cmsg, reason_invalid_intercept_param);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, rule_id, meta->session_id, "active", "passthrough", "invalid intercept param");
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_rule_id, meta->session_id, "active", "passthrough", "invalid intercept param");
goto passthrough;
}
@@ -1151,7 +1153,7 @@ static int handle_session_opening(struct metadata *meta, marsio_buff_t *rx_buff,
is_passthrough = 1;
__atomic_fetch_add(&packet_io_fs->hit_no_intercept_num, 1, __ATOMIC_RELAXED);
set_passthrough_reason(parser->cmsg, reason_no_intercept_param);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, rule_id, meta->session_id, "active", "passthrough", "hit no intercept");
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_rule_id, meta->session_id, "active", "passthrough", "hit no intercept");
goto passthrough;
}
__atomic_fetch_add(&packet_io_fs->hit_intercept_num, 1, __ATOMIC_RELAXED);
@@ -1162,7 +1164,7 @@ static int handle_session_opening(struct metadata *meta, marsio_buff_t *rx_buff,
is_passthrough = 1;
__atomic_fetch_add(&packet_io_fs->tcp_pcy_inval_num, 1, __ATOMIC_RELAXED);
set_passthrough_reason(parser->cmsg, reason_invalid_tcp_policy_param);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
goto passthrough;
}
@@ -1177,7 +1179,7 @@ static int handle_session_opening(struct metadata *meta, marsio_buff_t *rx_buff,
is_passthrough = 1;
__atomic_fetch_add(&packet_io_fs->tcp_pcy_inval_num, 1, __ATOMIC_RELAXED);
set_passthrough_reason(parser->cmsg, reason_invalid_tcp_policy_param);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
goto passthrough;
}
tcp_restore_info_dump(&restore_info, meta->session_id, logger);
@@ -1189,7 +1191,7 @@ static int handle_session_opening(struct metadata *meta, marsio_buff_t *rx_buff,
is_passthrough = 1;
__atomic_fetch_add(&packet_io_fs->tcp_pcy_inval_num, 1, __ATOMIC_RELAXED);
set_passthrough_reason(parser->cmsg, reason_invalid_tcp_policy_param);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
goto passthrough;
}
@@ -1201,7 +1203,7 @@ static int handle_session_opening(struct metadata *meta, marsio_buff_t *rx_buff,
is_passthrough = 1;
__atomic_fetch_add(&packet_io_fs->tcp_pcy_inval_num, 1, __ATOMIC_RELAXED);
set_passthrough_reason(parser->cmsg, reason_invalid_tcp_policy_param);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
goto passthrough;
}
@@ -1219,7 +1221,7 @@ static int handle_session_opening(struct metadata *meta, marsio_buff_t *rx_buff,
is_passthrough = 1;
__atomic_fetch_add(&packet_io_fs->tcp_pcy_inval_num, 1, __ATOMIC_RELAXED);
set_passthrough_reason(parser->cmsg, reason_invalid_tcp_policy_param);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
goto passthrough;
}
@@ -1230,7 +1232,7 @@ static int handle_session_opening(struct metadata *meta, marsio_buff_t *rx_buff,
is_passthrough = 1;
__atomic_fetch_add(&packet_io_fs->tcp_pcy_inval_num, 1, __ATOMIC_RELAXED);
set_passthrough_reason(parser->cmsg, reason_invalid_tcp_policy_param);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
goto passthrough;
}
}
@@ -1248,7 +1250,7 @@ static int handle_session_opening(struct metadata *meta, marsio_buff_t *rx_buff,
is_passthrough = 1;
__atomic_fetch_add(&packet_io_fs->tcp_pcy_inval_num, 1, __ATOMIC_RELAXED);
set_passthrough_reason(parser->cmsg, reason_invalid_tcp_policy_param);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_rule_id, meta->session_id, "active", "passthrough", "invalid tcp policy param");
goto passthrough;
}
__atomic_fetch_add(&packet_io_fs->can_intercept_num, 1, __ATOMIC_RELAXED);
@@ -1258,11 +1260,11 @@ static int handle_session_opening(struct metadata *meta, marsio_buff_t *rx_buff,
set_passthrough_reason(parser->cmsg, reason_underlying_stream_error);
if (parser->intercpet_data & IS_SINGLE) {
__atomic_fetch_add(&packet_io_fs->asymmetric_num, 1, __ATOMIC_RELAXED);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, rule_id, meta->session_id, "active", "passthrough", "asymmetric traffic");
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_rule_id, meta->session_id, "active", "passthrough", "asymmetric traffic");
}
else if (parser->intercpet_data & IS_TUNNEL) {
__atomic_fetch_add(&packet_io_fs->tunnel_num, 1, __ATOMIC_RELAXED);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, rule_id, meta->session_id, "active", "passthrough", "tunnel traffic");
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_rule_id, meta->session_id, "active", "passthrough", "tunnel traffic");
}
}
@@ -1275,7 +1277,7 @@ passthrough:
s_ctx->session_id = meta->session_id;
tuple4_tostring(&inner_tuple4, s_ctx->session_addr, sizeof(s_ctx->session_addr));
s_ctx->cmsg = parser->cmsg;
s_ctx->policy_ids = rule_id;
memcpy(s_ctx->policy_ids, rule_id, UUID_LEN);
s_ctx->is_passthrough = is_passthrough;
metadata_deep_copy(s_ctx->ctrl_meta, meta);
sids_copy(&s_ctx->ctrl_meta->sids, &meta->sids);
@@ -1306,7 +1308,7 @@ passthrough:
sids_copy(&s_ctx->s2c_info.sids, &parser->ack_sids);
route_ctx_copy(&s_ctx->s2c_info.route_ctx, &parser->ack_route_ctx);
TFE_LOG_INFO(logger, "%s: session %lu %s active first, hit rule %lu", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->session_addr, rule_id);
TFE_LOG_INFO(logger, "%s: session %lu %s active first, hit rule %s", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->session_addr, str_rule_id);
session_table_insert(thread->session_table, s_ctx->session_id, &(s_ctx->c2s_info.tuple4), s_ctx, session_value_free_cb);
ATOMIC_INC(&(packet_io_fs->session_num));
if (parser->seq_header)
@@ -1316,7 +1318,7 @@ passthrough:
if (is_passthrough == 0)
{
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, rule_id, meta->session_id, "active", "intercept", NULL);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_rule_id, meta->session_id, "active", "intercept", NULL);
}
return 0;
}
@@ -1344,6 +1346,7 @@ static int handle_session_closing(struct metadata *meta, marsio_buff_t *rx_buff,
struct packet_io *packet_io = thread->ref_io;
struct packet_io_fs *packet_io_fs = thread->ret_fs_state;
void * logger = thread->logger;
char str_policy_id[UUID_STR_LEN] = {0};
struct session_node *node = session_table_search_by_id(thread->session_table, meta->session_id);
if (node)
@@ -1351,7 +1354,8 @@ static int handle_session_closing(struct metadata *meta, marsio_buff_t *rx_buff,
struct session_ctx *s_ctx = (struct session_ctx *)node->val_data;
tfe_fieldstat_intercept_incrby(thread->ref_acceptor_ctx->metrics, s_ctx, thread->thread_index);
TFE_LOG_INFO(logger, "%s: session %lu closing", LOG_TAG_PKTIO, s_ctx->session_id);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, s_ctx->policy_ids, meta->session_id, "closing", NULL, NULL);
uuid_unparse(s_ctx->policy_ids, str_policy_id);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_policy_id, meta->session_id, "closing", NULL, NULL);
session_table_delete_by_id(thread->session_table, meta->session_id);
ATOMIC_DEC(&(packet_io_fs->session_num));
return 0;
@@ -1369,9 +1373,10 @@ static int handle_session_resetall(struct metadata *meta, marsio_buff_t *rx_buff
struct packet_io *packet_io = thread->ref_io;
struct packet_io_fs *packet_io_fs = thread->ret_fs_state;
void * logger = thread->logger;
char str_policy_id[UUID_STR_LEN] = {0};
TFE_LOG_ERROR(logger, "%s: session %lu resetall: notification clears all session tables !!!", LOG_TAG_PKTIO, meta->session_id);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, 0, meta->session_id, "resetall", NULL, NULL);
tfe_dp_telemetry_on_ctrl_pkt(packet_io->instance, rx_buff, str_policy_id, meta->session_id, "resetall", NULL, NULL);
ATOMIC_ZERO(&(packet_io_fs->session_num));
for (int i = 0; i < acceptor_ctx->nr_worker_threads; i++)
{
@@ -1452,6 +1457,7 @@ static int handle_raw_packet_from_nf(struct packet_io *handle, marsio_buff_t *rx
char *header = NULL;
int header_len = 0;
void * logger = thread->logger;
char str_policy_id[UUID_STR_LEN] = {0};
int raw_len = marsio_buff_datalen(rx_buff);
char *raw_data = marsio_buff_mtod(rx_buff);
@@ -1466,7 +1472,7 @@ static int handle_raw_packet_from_nf(struct packet_io *handle, marsio_buff_t *rx
LOG_TAG_PKTIO, meta.session_id, meta.raw_len, meta.is_e2i_dir, meta.is_ctrl_pkt, meta.l7offset, meta.is_decrypted, meta.sids.num);
throughput_metrics_inc(&packet_io_fs->raw_pkt_rx, 1, raw_len);
throughput_metrics_inc(&packet_io_fs->raw_bypass, 1, raw_len);
tfe_dp_on_raw_pkt(packet_io->instance, rx_buff, 0, "raw", "passthrough", "miss metadata");
tfe_dp_on_raw_pkt(packet_io->instance, rx_buff, str_policy_id, "raw", "passthrough", "miss metadata");
marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1);
return -1;
}
@@ -1478,7 +1484,7 @@ static int handle_raw_packet_from_nf(struct packet_io *handle, marsio_buff_t *rx
throughput_metrics_inc(&packet_io_fs->raw_pkt_rx, 1, raw_len);
throughput_metrics_inc(&packet_io_fs->raw_bypass, 1, raw_len);
throughput_metrics_inc(&packet_io_fs->dup_bypass, 1, raw_len);
tfe_dp_on_raw_pkt(packet_io->instance, rx_buff, 0, "duplicated", "passthrough", NULL);
tfe_dp_on_raw_pkt(packet_io->instance, rx_buff, str_policy_id, "duplicated", "passthrough", NULL);
marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1);
return -1;
}
@@ -1501,12 +1507,13 @@ static int handle_raw_packet_from_nf(struct packet_io *handle, marsio_buff_t *rx
tuple4_tostring(&inner_addr, buffer, sizeof(buffer));
TFE_LOG_ERROR(logger, "packet from nf %lu: %s (ipid: %u) miss session table", meta.session_id, buffer, ipid);
}
tfe_dp_on_raw_pkt(packet_io->instance, rx_buff, 0, "decrypted", "passthrough", "miss session");
tfe_dp_on_raw_pkt(packet_io->instance, rx_buff, str_policy_id, "decrypted", "passthrough", "miss session");
marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1);
return -1;
}
struct session_ctx *s_ctx = (struct session_ctx *)node->val_data;
uuid_unparse(s_ctx->policy_ids, str_policy_id);
if (s_ctx->is_passthrough > 0) {
throughput_metrics_inc(&packet_io_fs->raw_pkt_rx, 1, raw_len);
@@ -1525,14 +1532,14 @@ static int handle_raw_packet_from_nf(struct packet_io *handle, marsio_buff_t *rx
send_event_log(s_ctx, thread_seq, ctx);
tfe_cmsg_set_flag(s_ctx->cmsg, TFE_CMSG_FLAG_INIT);
}
tfe_dp_on_raw_pkt(packet_io->instance, rx_buff, s_ctx->policy_ids, "decrypted", "passthrough", NULL);
tfe_dp_on_raw_pkt(packet_io->instance, rx_buff, str_policy_id, "decrypted", "passthrough", NULL);
marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1);
return 0;
}
if (meta.is_decrypted)
{
tfe_dp_on_raw_pkt(packet_io->instance, rx_buff, s_ctx->policy_ids, "decrypted", "intercept", NULL);
tfe_dp_on_raw_pkt(packet_io->instance, rx_buff, str_policy_id, "decrypted", "intercept", NULL);
throughput_metrics_inc(&packet_io_fs->decrypt_rx, 1, raw_len);
if (memcmp(&inner_addr, &s_ctx->c2s_info.tuple4, sizeof(struct tuple4)) == 0) {
add_ether_header(raw_data, packet_io->config.tap_c_mac, packet_io->config.tap_s_mac);
@@ -1557,7 +1564,7 @@ static int handle_raw_packet_from_nf(struct packet_io *handle, marsio_buff_t *rx
}
else
{
tfe_dp_on_raw_pkt(packet_io->instance, rx_buff, s_ctx->policy_ids, "raw", "intercept", NULL);
tfe_dp_on_raw_pkt(packet_io->instance, rx_buff, str_policy_id, "raw", "intercept", NULL);
throughput_metrics_inc(&packet_io_fs->raw_pkt_rx, 1, raw_len);
if (memcmp(&inner_addr, &s_ctx->c2s_info.tuple4, sizeof(struct tuple4)) == 0) {
s_ctx->c2s_info.sids = meta.sids;