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

✨ feat(plugin register): remove ip_proto in parameter

Browse Source
This commit is contained in:
yangwei
2024-09-06 13:22:02 +08:00
parent 9c12523c9d
commit 442586ef52
4 changed files with 57 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
include/stellar/session.h
View File

@@ -9,6 +9,16 @@ extern "C"
#include "stellar/packet.h"
struct tcp_segment;
const char *tcp_segment_get_data(const struct tcp_segment *seg);
uint16_t tcp_segment_get_len(const struct tcp_segment *seg);
#define TOPIC_TCP_STREAM "TCP_STREAM" //topic message: tcp_segment
#define TOPIC_CONTROL_PACKET "CONTROL_PACKET" //topic message: packet
#define TOPIC_TCP "TCP" //topic message: session
#define TOPIC_UDP "UDP" //topic message: session
enum session_state
{
SESSION_STATE_INIT = 0,

28
include/stellar/stellar.h
View File

@@ -9,34 +9,32 @@ extern "C"
struct stellar;
/**********************************************
* PLUGIN SPEC API *
**********************************************/
//return plugin_env
typedef void *plugin_on_load_func(struct stellar *st);
typedef void plugin_on_unload_func(void *plugin_env);
struct tcp_segment;
const char *tcp_segment_get_data(const struct tcp_segment *seg);
uint16_t tcp_segment_get_len(const struct tcp_segment *seg);
#define TOPIC_TCP_STREAM "TCP_STREAM" //topic message: tcp_segment
#define TOPIC_CONTROL_PACKET "CONTROL_PACKET" //topic message: packet
#define TOPIC_TCP "TCP" //topic message: session
#define TOPIC_UDP "UDP" //topic message: session
/**********************************************
* PLUGIN EVENT API *
**********************************************/
struct packet;
typedef void plugin_on_packet_func(struct packet *pkt, unsigned char ip_protocol, void *plugin_env);
typedef void plugin_on_packet_func(struct packet *pkt, void *plugin_env);
//return plugin_id
int stellar_plugin_register(struct stellar *st, unsigned char ip_protocol, plugin_on_packet_func on_packet_input, plugin_on_packet_func on_packet_output, void *plugin_env);
int stellar_plugin_register(struct stellar *st, plugin_on_packet_func on_packet_input, plugin_on_packet_func on_packet_output, void *plugin_env);
//return polling work result, 0: no work, 1: work
typedef int plugin_on_polling_func(void *plugin_env);
//return polling plugin_id
int stellar_polling_plugin_register(struct stellar *st, plugin_on_polling_func on_polling, void *plugin_env);
/**********************************************
* STELLAR DEV API *
**********************************************/
void stellar_emit_datapath_telemetry(struct packet *pkt, const char * module, const char *str);
int stellar_get_worker_thread_num(struct stellar *st);

13
infra/plugin_manager/plugin_manager.c
View File

@@ -7,7 +7,6 @@
#include <stdbool.h>
#include "stellar_core.h"
#include "tuple.h"
#include "packet_private.h"
#include "session_private.h"
@@ -613,7 +612,7 @@ void session_exdata_runtime_free(struct stellar_exdata *exdata_rt)
*********************************************/
UT_icd registered_plugin_array_icd = {sizeof(struct registered_plugin_schema), NULL, NULL, NULL};
int stellar_plugin_register(struct stellar *st, unsigned char ip_proto, plugin_on_packet_func on_packet_input, plugin_on_packet_func on_packet_output, void *plugin_env)
int stellar_plugin_register(struct stellar *st, plugin_on_packet_func on_packet_input, plugin_on_packet_func on_packet_output, void *plugin_env)
{
struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
if(plug_mgr->registered_packet_plugin_array == NULL)
@@ -622,7 +621,6 @@ int stellar_plugin_register(struct stellar *st, unsigned char ip_proto, plugin_o
}
struct registered_plugin_schema packet_plugin_schema;
memset(&packet_plugin_schema, 0, sizeof(packet_plugin_schema));
packet_plugin_schema.ip_protocol = ip_proto;
packet_plugin_schema.on_packet[PACKET_STAGE_INPUT] = on_packet_input;
packet_plugin_schema.on_packet[PACKET_STAGE_OUTPUT] = on_packet_output;
packet_plugin_schema.plugin_env = plugin_env;
@@ -635,19 +633,14 @@ static void plugin_manager_on_packet(struct plugin_manager_schema *plug_mgr, str
if(plug_mgr==NULL || plug_mgr->registered_packet_plugin_array == NULL || pkt == NULL)return;
struct registered_plugin_schema *p=NULL;
//TODO: get innermost layer ip protocol by packet api
struct tuple6 t6;
packet_get_innermost_tuple6(pkt, &t6);
unsigned char ip_proto=t6.ip_proto;
int tid=stellar_get_current_thread_index();
//TODO : provide public api to reset pub_msg_cnt
plug_mgr->per_thread_data[tid].pub_packet_msg_cnt=0;//reset pub_msg_cnt
while ((p = (struct registered_plugin_schema *)utarray_next(plug_mgr->registered_packet_plugin_array, p)))
{
if(p->ip_protocol == ip_proto && p->on_packet[in_out])
if(p->on_packet[in_out])
{
p->on_packet[in_out](pkt, ip_proto, p->plugin_env);
p->on_packet[in_out](pkt, p->plugin_env);
}
}
stellar_mq_dispatch(plug_mgr->per_thread_data[tid].priority_mq, &plug_mgr->per_thread_data[tid].dealth_letter_queue);

62
infra/plugin_manager/test/plugin_manager_gtest_main.cpp
View File

@@ -171,7 +171,7 @@ TEST(plugin_manager_init, stellar_mq_subscribe) {
EXPECT_EQ(stellar_mq_subscribe(&st, topic_id, test_mock_on_packet_msg, 10),-1);//illgeal plugin_id
EXPECT_EQ(stellar_mq_subscribe(&st, 10, test_mock_on_packet_msg, 10),-1);//illgeal topic_id & plugin_id
int plugin_id=stellar_plugin_register(&st, 6, NULL, NULL,&st);
int plugin_id=stellar_plugin_register(&st, NULL, NULL,&st);
EXPECT_GE(plugin_id, 0);
EXPECT_EQ(stellar_mq_subscribe(&st, topic_id, test_mock_on_packet_msg, plugin_id),0);
@@ -243,7 +243,7 @@ TEST(plugin_manager_init, stellar_mq_subscribe_overwrite) {
EXPECT_EQ(stellar_mq_subscribe(&st, topic_id, test_mock_on_session_msg, 10),-1);//illgeal plugin_id
EXPECT_EQ(stellar_mq_subscribe(&st, 10, test_mock_on_session_msg, 10),-1);//illgeal topic_id & plugin_id
int plugin_id=stellar_plugin_register(&st, 0, NULL, NULL, NULL);
int plugin_id=stellar_plugin_register(&st, NULL, NULL, NULL);
EXPECT_GE(plugin_id, 0);
EXPECT_EQ(stellar_mq_subscribe(&st, topic_id, test_mock_on_session_msg, plugin_id),0);
@@ -319,11 +319,11 @@ struct packet_plugin_env
int msg_free_cnt;
};
static void test_basic_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
static void test_basic_on_packet(struct packet *pkt, void *plugin_env)
{
struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
EXPECT_TRUE(env!=NULL);
EXPECT_EQ(pkt->ip_proto, ip_protocol);
//EXPECT_EQ(pkt->ip_proto, ip_protocol);
EXPECT_EQ(pkt->st, env->plug_mgr->st);
EXPECT_EQ(packet_exdata_set(pkt, 2, pkt), -1);// illegal set
EXPECT_EQ(packet_exdata_get(pkt, 2), nullptr);// illegal get
@@ -341,7 +341,7 @@ TEST(plugin_manager, packet_plugin_illegal_exdata) {
struct packet_plugin_env env;
memset(&env, 0, sizeof(struct packet_plugin_env));
env.plug_mgr=plug_mgr;
int plugin_id=stellar_plugin_register(&st, ip_proto, test_basic_on_packet, NULL,&env);
int plugin_id=stellar_plugin_register(&st, test_basic_on_packet, NULL,&env);
EXPECT_GE(plugin_id, 0);
{
@@ -359,19 +359,19 @@ TEST(plugin_manager, packet_plugin_illegal_exdata) {
EXPECT_EQ(env.basic_on_packet_called, 1);
}
static void test_proto_filter_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
static void test_proto_filter_on_packet(struct packet *pkt, void *plugin_env)
{
struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
EXPECT_TRUE(env!=NULL);
EXPECT_EQ(pkt->ip_proto, ip_protocol);
//EXPECT_EQ(pkt->ip_proto, ip_protocol);
EXPECT_EQ(pkt->st, env->plug_mgr->st);
EXPECT_EQ(packet_exdata_set(pkt, 2, pkt), -1);// illegal set
EXPECT_EQ(packet_exdata_get(pkt, 2), nullptr);// illegal get
env->proto_filter_plugin_called[ip_protocol]+=1;
//env->proto_filter_plugin_called[ip_protocol]+=1;
return;
}
TEST(plugin_manager, packet_plugins_with_proto_filter) {
TEST(plugin_manager, DISABLED_packet_plugins_with_proto_filter) {
struct stellar st={0};
struct plugin_manager_schema *plug_mgr = plugin_manager_init(&st, NULL);
@@ -384,7 +384,7 @@ TEST(plugin_manager, packet_plugins_with_proto_filter) {
int proto_filter_plugin_num=(int)(sizeof(env.proto_filter_plugin_id) / sizeof(env.proto_filter_plugin_id[0]));
for (int i = 0; i < proto_filter_plugin_num; i++)
{
env.proto_filter_plugin_id[i] = stellar_plugin_register(&st, i, test_proto_filter_on_packet, NULL,&env);
env.proto_filter_plugin_id[i] = stellar_plugin_register(&st, test_proto_filter_on_packet, NULL,&env);
EXPECT_GE(env.proto_filter_plugin_id[i], 0);
@@ -423,11 +423,11 @@ struct test_exdata
long long value;
};
static void test_exdata_set_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
static void test_exdata_set_on_packet(struct packet *pkt, void *plugin_env)
{
struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
EXPECT_TRUE(env!=NULL);
EXPECT_EQ(pkt->ip_proto, ip_protocol);
//EXPECT_EQ(pkt->ip_proto, ip_protocol);
EXPECT_EQ(pkt->st, env->plug_mgr->st);
env->exdata_set_on_packet_called+=1;
@@ -443,11 +443,11 @@ static void test_exdata_set_on_packet(struct packet *pkt, unsigned char ip_proto
return;
}
static void test_exdata_get_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
static void test_exdata_get_on_packet(struct packet *pkt, void *plugin_env)
{
struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
EXPECT_TRUE(env!=NULL);
EXPECT_EQ(pkt->ip_proto, ip_protocol);
//EXPECT_EQ(pkt->ip_proto, ip_protocol);
EXPECT_EQ(pkt->st, env->plug_mgr->st);
int exdata_idx_len=(int)(sizeof(env->packet_exdata_idx) / sizeof(env->packet_exdata_idx[0]));
@@ -512,10 +512,10 @@ TEST(plugin_manager, packet_plugins_share_exdata) {
EXPECT_EQ(utarray_len(plug_mgr->stellar_exdata_schema_array), exdata_idx_len);
}
int exdata_set_plugin_id=stellar_plugin_register(&st, ip_proto, test_exdata_set_on_packet, NULL,&env);
int exdata_set_plugin_id=stellar_plugin_register(&st, test_exdata_set_on_packet, NULL,&env);
EXPECT_GE(exdata_set_plugin_id, 0);
int exdata_get_plugin_id=stellar_plugin_register(&st, ip_proto, test_exdata_get_on_packet, NULL,&env);
int exdata_get_plugin_id=stellar_plugin_register(&st, test_exdata_get_on_packet, NULL,&env);
EXPECT_GE(exdata_get_plugin_id, 0);
{
@@ -558,11 +558,11 @@ static void test_mq_on_packet_msg(int topic_id, const void *msg, void *plugin_en
return;
}
static void test_mq_pub_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
static void test_mq_pub_on_packet(struct packet *pkt, void *plugin_env)
{
struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
EXPECT_TRUE(env!=NULL);
EXPECT_EQ(pkt->ip_proto, ip_protocol);
//EXPECT_EQ(pkt->ip_proto, ip_protocol);
EXPECT_EQ(pkt->st, env->plug_mgr->st);
int topic_id_num=(int)(sizeof(env->packet_topic_id) / sizeof(env->packet_topic_id[0]));
for(int i=0; i<topic_id_num; i++)
@@ -608,14 +608,14 @@ TEST(plugin_manager, packet_plugins_mq_pub_sub) {
EXPECT_EQ(utarray_len(plug_mgr->stellar_mq_schema_array), topic_id_num+STELLAR_INTRINSIC_TOPIC_NUM);
}
int pub_plugin_id=stellar_plugin_register(&st, ip_proto, test_mq_pub_on_packet, NULL,&env);
int pub_plugin_id=stellar_plugin_register(&st, test_mq_pub_on_packet, NULL,&env);
EXPECT_GE(pub_plugin_id, 0);
int topic_sub_num=(int)(sizeof(env.packet_mq_sub_plugin_id) / sizeof(env.packet_mq_sub_plugin_id[0]));
for (int i = 0; i < topic_sub_num; i++)
{
env.packet_mq_sub_plugin_id[i] = stellar_plugin_register(&st, ip_proto, NULL, NULL,&env);// empty on_packet is ok
env.packet_mq_sub_plugin_id[i] = stellar_plugin_register(&st, NULL, NULL,&env);// empty on_packet is ok
EXPECT_GE(env.packet_mq_sub_plugin_id[i], 0);
for(int j = 0; j < topic_id_num; j++)
{
@@ -659,11 +659,11 @@ static void overlimit_sub_on_packet_msg(int topic_id, const void *msg, void *plu
return;
}
static void overlimit_pub_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
static void overlimit_pub_on_packet(struct packet *pkt, void *plugin_env)
{
struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
EXPECT_TRUE(env!=NULL);
EXPECT_EQ(pkt->ip_proto, ip_protocol);
//EXPECT_EQ(pkt->ip_proto, ip_protocol);
int topic_id_num=(int)(sizeof(env->packet_topic_id) / sizeof(env->packet_topic_id[0]));
int cnt=0;
int *msg;
@@ -725,14 +725,14 @@ TEST(plugin_manager, packet_plugins_pub_overlimit) {
EXPECT_EQ(utarray_len(plug_mgr->stellar_mq_schema_array), topic_id_num+STELLAR_INTRINSIC_TOPIC_NUM);
}
int pub_plugin_id=stellar_plugin_register(&st, ip_proto, overlimit_pub_on_packet, NULL,&env);
int pub_plugin_id=stellar_plugin_register(&st, overlimit_pub_on_packet, NULL,&env);
EXPECT_GE(pub_plugin_id, 0);
int topic_sub_num=(int)(sizeof(env.packet_mq_sub_plugin_id) / sizeof(env.packet_mq_sub_plugin_id[0]));
for (int i = 0; i < topic_sub_num; i++)
{
env.packet_mq_sub_plugin_id[i] = stellar_plugin_register(&st, ip_proto, NULL, NULL, &env);// empty on_packet is ok
env.packet_mq_sub_plugin_id[i] = stellar_plugin_register(&st, NULL, NULL, &env);// empty on_packet is ok
EXPECT_GE(env.packet_mq_sub_plugin_id[i], 0);
for(int j = 0; j < topic_id_num; j++)
{
@@ -780,11 +780,11 @@ static void test_exdata_free_pub_msg_free( void *msg, void *msg_free_arg)
return;
}
static void test_exdata_free_pub_msg_on_packet(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
static void test_exdata_free_pub_msg_on_packet(struct packet *pkt, void *plugin_env)
{
struct packet_plugin_env *env = (struct packet_plugin_env *)plugin_env;
EXPECT_TRUE(env!=NULL);
EXPECT_EQ(pkt->ip_proto, ip_protocol);
//EXPECT_EQ(pkt->ip_proto, ip_protocol);
EXPECT_EQ(pkt->st, env->plug_mgr->st);
EXPECT_EQ(packet_exdata_set(pkt, env->packet_exdata_idx[0], pkt), 0);
env->basic_on_packet_called+=1;
@@ -808,7 +808,7 @@ TEST(plugin_manager, packet_plugin_exdata_free_pub_msg) {
struct packet_plugin_env env;
memset(&env, 0, sizeof(struct packet_plugin_env));
env.plug_mgr=plug_mgr;
int plugin_id=stellar_plugin_register(&st, ip_proto, test_exdata_free_pub_msg_on_packet, NULL,&env);
int plugin_id=stellar_plugin_register(&st, test_exdata_free_pub_msg_on_packet, NULL,&env);
EXPECT_GE(plugin_id, 0);
env.packet_exdata_idx[0]=stellar_exdata_new_index(&st, "PACKET_EXDATA", test_exdata_free_pub_msg_exdata_free, &env);
@@ -878,7 +878,7 @@ static void pesudo_on_msg_dispatch(int topic_id, const void *msg, on_msg_cb_func
struct session *sess=(struct session *)msg;
session_cb(topic_id, sess, sub_plugin_env);
}
static void pesudo_on_packet_input(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
static void pesudo_on_packet_input(struct packet *pkt, void *plugin_env)
{
struct session_manager_plugin_env *env=(struct session_manager_plugin_env *)plugin_env;
for (int i = 0; i < env->N_session; i++)
@@ -887,7 +887,7 @@ static void pesudo_on_packet_input(struct packet *pkt, unsigned char ip_protocol
}
}
static void pesudo_on_packet_output(struct packet *pkt, unsigned char ip_protocol, void *plugin_env)
static void pesudo_on_packet_output(struct packet *pkt, void *plugin_env)
{
struct session_manager_plugin_env *env=(struct session_manager_plugin_env *)plugin_env;
for (int i = 0; i < env->N_session; i++)
@@ -906,7 +906,7 @@ static void pesudo_session_load(struct stellar *st, struct session_manager_plugi
}
env->intrinsc_tcp_input_topic_id=stellar_mq_create_topic(st, TOPIC_TCP, pesudo_on_msg_dispatch, NULL, env);
env->session_manager_plugin_id=stellar_plugin_register(st, 6, pesudo_on_packet_input, pesudo_on_packet_output, env);
env->session_manager_plugin_id=stellar_plugin_register(st, pesudo_on_packet_input, pesudo_on_packet_output, env);
}
void pesudo_session_unload(struct stellar *st, struct session_manager_plugin_env *env)
@@ -1005,7 +1005,7 @@ TEST(plugin_manager, session_plugin_on_tcp) {
env.N_per_session_pkt_cnt=10;
env.N_session=1;
int plugin_id=stellar_plugin_register(&st, 0, NULL, NULL, &env);
int plugin_id=stellar_plugin_register(&st, NULL, NULL, &env);
EXPECT_GE(plugin_id, 0);
EXPECT_EQ(pesudo_tcp_session_subscribe(&st, test_basic_on_tcp_session, plugin_id), 0);