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

TSG-17862 Proxy支持Internal IP address和Exteral IP Address的扫描, 支持以Topic方式上传HTTP请求体/应答体

Browse Source
This commit is contained in:
fengweihao
2023-12-14 15:08:19 +08:00
parent b801ca9d3b
commit b3700966fd
15 changed files with 577 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
plugin/business/tsg-http/src/tsg_logger.cpp
View File

@@ -5,6 +5,7 @@
#include <tfe_utils.h>
#include <tfe_resource.h>
#include "mpack.h"
#include "tsg_proxy_logger.h"
struct json_spec
@@ -55,7 +56,57 @@ void get_http_body_uuid(char *uuid)
return;
}
struct proxy_logger* proxy_log_handle_create(const char* profile, const char* section, void* local_logger)
size_t file_bucket_upload_once(struct proxy_logger* handle, char *uuid, struct evbuffer *http_body)
{
int kafka_status=0;
mpack_writer_t writer;
char *mpack_data=NULL, *data=NULL;
size_t mpack_size=0, datalen=0;
mpack_writer_init_growable(&writer, &mpack_data, &mpack_size);
mpack_build_map(&writer);
mpack_write_cstr(&writer, "uuid");
mpack_write_cstr(&writer, uuid);
mpack_write_cstr(&writer, "fileType");
mpack_write_cstr(&writer, "txt");
mpack_write_cstr(&writer, "combineMode");
mpack_write_cstr(&writer, "seek");
mpack_write_cstr(&writer, "offset");
mpack_write_u64(&writer, 0);
mpack_write_cstr(&writer, "lastChunkFlag");
mpack_write_u32(&writer, 1);
datalen = evbuffer_get_length(http_body);
if(datalen > 0)
{
data = (char *)evbuffer_pullup(http_body, datalen);
mpack_write_cstr(&writer, "chunk");
mpack_start_bin(&writer, datalen);
mpack_write_bytes(&writer, (const char *)data, datalen);
mpack_finish_bin(&writer);
}
mpack_write_cstr(&writer, "length");
mpack_write_u64(&writer, datalen);
mpack_complete_map(&writer); // mpack_init_map
mpack_error_t errorno=mpack_writer_destroy(&writer);
if(errorno!=mpack_ok)
{
TFE_LOG_ERROR(handle->local_logger, "Mpack writer destroy is error(%s), uuid: %s", mpack_error_to_string(errorno), uuid);
}
kafka_status = tfe_kafka_logger_send(handle->kafka_logger, TOPIC_BUCKET, mpack_data, mpack_size);
if(kafka_status<0)
{
TFE_LOG_ERROR(handle->local_logger, "Kafka produce failed: %s", rd_kafka_err2name(rd_kafka_last_error()));
}
free(mpack_data);
mpack_data = NULL;
mpack_size = 0;
return datalen;
}
struct proxy_logger* proxy_log_handle_create(const char* profile, const char* section, void* local_logger)
{
struct tango_cache_parameter *log_file_upload_para=NULL;
struct proxy_logger* instance=ALLOC(struct proxy_logger,1);
@@ -157,12 +208,8 @@ int proxy_send_log(struct proxy_logger* handle, const struct proxy_log* log_msg)
common_obj=cJSON_CreateObject();
gettimeofday(&cur_time, NULL);
cJSON_AddNumberToObject(common_obj, "start_timestamp_ms", get_time_ms(http->start_time));
cJSON_AddNumberToObject(common_obj, "end_timestamp_ms", get_time_ms(cur_time));
cJSON_AddStringToObject(common_obj, "http_version", app_proto[http->major_version]);
cJSON_AddStringToObject(common_obj, "decoded_as", "HTTP");
unsigned int category_id_val[64]={0};
char opt_val[24]={0}; uint16_t opt_out_size;
@@ -236,6 +283,8 @@ int proxy_send_log(struct proxy_logger* handle, const struct proxy_log* log_msg)
s2c_byte_num = log_msg->s2c_byte_num;
}
cJSON_AddStringToObject(common_obj, "http_version", app_proto[http->major_version]);
cJSON_AddStringToObject(common_obj, "decoded_as", "HTTP");
cJSON_AddNumberToObject(common_obj, "out_link_id", 0);
cJSON_AddNumberToObject(common_obj, "in_link_id", 0);
cJSON_AddStringToObject(common_obj, "sled_ip", handle->kafka_logger->local_ip_str);
@@ -271,18 +320,12 @@ int proxy_send_log(struct proxy_logger* handle, const struct proxy_log* log_msg)
#define FILE_CHUNK_UUID_LEN 40
char uuid[FILE_CHUNK_UUID_LEN]={0};
size_t datalen=0;
for(size_t i=0; i<log_msg->result_num; i++)
{
if(log_msg->result[i].do_log!=1)
{
continue;
}
if(handle->en_hoslog!=1)
{
continue;
}
if(log_msg->result[i].do_log!=1) continue;
if(handle->en_hoslog!=1) continue;
if(log_msg->req_body!=NULL)
{
@@ -293,7 +336,15 @@ int proxy_send_log(struct proxy_logger* handle, const struct proxy_log* log_msg)
else
{
get_http_body_uuid(uuid);
cJSON_AddStringToObject(common_obj, "http_request_body", uuid);
datalen=file_bucket_upload_once(handle, uuid, log_msg->req_body);
if(datalen>0)
{
cJSON_AddStringToObject(common_obj, "http_request_body", uuid);
}
else
{
TFE_LOG_ERROR(handle->local_logger, "Upload req_body failed.");
}
}
}
if(log_msg->resp_body!=NULL)
@@ -305,7 +356,15 @@ int proxy_send_log(struct proxy_logger* handle, const struct proxy_log* log_msg)
else
{
get_http_body_uuid(uuid);
cJSON_AddStringToObject(common_obj, "http_response_body", uuid);
datalen=file_bucket_upload_once(handle, uuid, log_msg->resp_body);
if(datalen>0)
{
cJSON_AddStringToObject(common_obj, "http_response_body", uuid);
}
else
{
TFE_LOG_ERROR(handle->local_logger, "Upload resp_body failed.");
}
}
}
}
@@ -361,7 +420,7 @@ int proxy_send_log(struct proxy_logger* handle, const struct proxy_log* log_msg)
TFE_LOG_DEBUG(handle->local_logger, "%s", log_payload);
kafka_status = tfe_kafka_logger_send(handle->kafka_logger, log_payload, strlen(log_payload));
kafka_status = tfe_kafka_logger_send(handle->kafka_logger, TOPIC_LOGGER, log_payload, strlen(log_payload));
free(log_payload);
cJSON_Delete(per_hit_obj);
if(kafka_status<0)