Initial commitHEADv1.0.1main

11 files changed, 1419 insertions, 0 deletions

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
new file mode 100644
index 0000000..075e7fd
--- /dev/null
+++ b/src/CMakeLists.txt
@@ -0,0 +1,15 @@
+cmake_minimum_required (VERSION 2.8...3.10)
+
+file(GLOB SRC "*.cpp")
+set(DEPEND_DYN_LIB MESA_handle_logger MESA_prof_load MESA_htable fieldstat4)
+
+include_directories(/opt/tsg/framework/include/)
+
+# Shared Library Output
+add_library(${lib_name} SHARED ${SRC})
+set_target_properties(${lib_name} PROPERTIES LINK_FLAGS "-Wl,--version-script=${PROJECT_SOURCE_DIR}/src/version.map")
+set_target_properties(${lib_name}  PROPERTIES PREFIX "")
+target_link_libraries(${lib_name} ${DEPEND_DYN_LIB} )
+set_target_properties(${lib_name} PROPERTIES OUTPUT_NAME ${lib_name})
+
+install(TARGETS ${lib_name} LIBRARY DESTINATION ${CMAKE_INSTALL_PREFIX}/sapp/stellar_plugin/ftp_decoder COMPONENT LIBRARIES)
diff --git a/src/ftp_decoder_entry.cpp b/src/ftp_decoder_entry.cpp
new file mode 100644
index 0000000..3dc0df0
--- /dev/null
+++ b/src/ftp_decoder_entry.cpp
@@ -0,0 +1,227 @@
+#include "ftp_decoder_inner.h"
+#include "ftp_decoder.h"
+#include "ftp_decoder_util.h"
+#include <stellar/session.h>
+#include <sys/types.h>
+#include <MESA/MESA_prof_load.h>
+#include <MESA/MESA_handle_logger.h>
+
+void *__ftp_decoder_logger_handle;
+
+static void ftp_msg_free_cb(struct session *sess, void *msg, void *msg_free_arg)
+{
+    free(msg);
+}
+
+static void *ftp_context_ctx_new_cb(struct session *sess, void *plugin_env)
+{
+    struct ftp_decoder_env *fenv = (struct ftp_decoder_env *)plugin_env;
+    size_t payload_len;
+    const char *payload = session_get0_current_payload(sess, &payload_len);
+    int curdir = ftp_session_get_pkt_dir(sess);
+    int thread_idx = session_get_current_thread_id(sess);
+    int is_ctrl_link = ftp_ctrl_identify(sess, payload, payload_len, curdir);
+    if (is_ctrl_link)
+    {
+        struct ftp_context *fctx = (struct ftp_context *)calloc(1, sizeof(struct ftp_context));
+        fctx->link_type = FTP_LINK_CTRL;
+        ftp_decoder_stat_incrby(thread_idx, fenv, FTPD_STAT_CTRL_LINK_OPEN, 1);
+        ftp_runtime_log(RLOG_LV_DEBUG, "ftp ctrl link new: %s", session_get0_readable_addr(sess));
+        return fctx;
+    }
+    ftp_runtime_log(RLOG_LV_DEBUG, "ftp_data_identify: is not ctrl link: '%s' ", session_get0_readable_addr(sess));
+
+    struct ftp_context *data_ctx = ftp_data_identify(sess, fenv);
+    if (data_ctx)
+    {
+        data_ctx->link_type = FTP_LINK_DATA;
+        ftp_decoder_stat_incrby(thread_idx, fenv, FTPD_STAT_DATA_LINK_OPEN, 1);
+        ftp_runtime_log(RLOG_LV_DEBUG, "ftp data link new: '%s' ", session_get0_readable_addr(sess));
+        return data_ctx;
+    }
+    ftp_runtime_log(RLOG_LV_DEBUG, "ftp_data_identify: is not data link: '%s' ", session_get0_readable_addr(sess));
+    stellar_session_plugin_dettach_current_session(sess);
+    return NULL;
+}
+
+void ftp_context_ctx_free_cb(struct session *sess, void *session_ctx, void *plugin_env)
+{
+    struct ftp_context *fctx = (struct ftp_context *)session_ctx;
+    if (NULL == fctx)
+    {
+        return;
+    }
+    int thread_idx = session_get_current_thread_id(sess);
+    if (fctx->link_type == FTP_LINK_CTRL)
+    {
+        ftp_decoder_stat_incrby(thread_idx, (struct ftp_decoder_env *)plugin_env, FTPD_STAT_CTRL_LINK_CLOSE, 1);
+    }
+    else
+    {
+        ftp_decoder_stat_incrby(thread_idx, (struct ftp_decoder_env *)plugin_env, FTPD_STAT_DATA_LINK_CLOSE, 1);
+    }
+    ftp_runtime_log(RLOG_LV_DEBUG, "ftp_context_ctx_free_cb: %s", session_get0_readable_addr(sess));
+    ftp_parse_result_free(&fctx->parse_result);
+    free(fctx);
+}
+
+static void ftpd_logger_init(struct ftp_decoder_env *fenv)
+{
+    int log_level;
+    char log_path[256];
+    MESA_load_profile_int_def(FTP_DECODER_CFG_FILE, "FTP", "log_level", &log_level, 30);
+    MESA_load_profile_string_def(FTP_DECODER_CFG_FILE, "FTP", "log_path", log_path, sizeof(log_path), "./log/ftp_decoder.log");
+    MESA_handle_runtime_log_creation("./etc/sapp_log.conf");
+    fenv->logger_handle = MESA_create_runtime_log_handle(log_path, log_level);
+    __ftp_decoder_logger_handle = fenv->logger_handle;
+    return;
+}
+
+void ftp_message_reset_iterator(const struct ftp_message *msg)
+{
+    msg->fctx->parse_result.cursor = 0;
+}
+
+static void ftp_ctrl_entry(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    size_t payload_len;
+    const char *payload = session_get0_current_payload(sess, &payload_len);
+    int curdir = ftp_session_get_pkt_dir(sess);
+    if (NULL == payload || payload_len == 0)
+    {
+        return;
+    }
+    int thread_idx = session_get_current_thread_id(sess);
+    ftp_cmd_readline(&fctx->cmd_result, payload, payload_len);
+    ftp_cmd_process(sess, fctx, fenv, curdir);
+    if (curdir == PACKET_DIRECTION_C2S)
+    {
+        ftp_decoder_stat_incrby(thread_idx, fenv, FTPD_STAT_CTRL_LINK_BYTES_C2S, (long long)payload_len);
+    }
+    else
+    {
+        ftp_decoder_stat_incrby(thread_idx, fenv, FTPD_STAT_CTRL_LINK_BYTES_S2C, (long long)payload_len);
+    }
+}
+
+static void ftp_data_entry(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    enum session_state sstate = session_get_current_state(sess);
+    if (SESSION_STATE_OPENING == sstate)
+    {
+        memset(fctx->parse_result.push_result_flags, 0, sizeof(fctx->parse_result.push_result_flags));
+        ftp_runtime_log(RLOG_LV_DEBUG, "ftp data link opening: %s", session_get0_readable_addr(sess));
+    }
+
+    size_t payload_len;
+    const char *payload = session_get0_current_payload(sess, &payload_len);
+    if (NULL == payload || payload_len == 0)
+    {
+        return;
+    }
+    enum ftp_msg_type dlink_presentation = fctx->data_link_presentation;
+    if (FTP_INVENTORY != dlink_presentation && FTP_FILE_CONTENT != dlink_presentation)
+    {
+        dlink_presentation = FTP_FILE_CONTENT; // unknown data link type, default as FTP_FILE_CONTENT
+    }
+    fctx->parse_result.result_array[dlink_presentation].iov_base = (void *)payload;
+    fctx->parse_result.result_array[dlink_presentation].iov_len = payload_len;
+    fctx->parse_result.push_result_flags[dlink_presentation] = 0;
+
+    ftp_decoder_push_msg(sess, fctx, fenv);
+    int thread_idx = session_get_current_thread_id(sess);
+    int curdir = ftp_session_get_pkt_dir(sess);
+    if (curdir == PACKET_DIRECTION_C2S)
+    {
+        ftp_decoder_stat_incrby(thread_idx, fenv, FTPD_STAT_DATA_LINK_BYTES_C2S, (long long)payload_len);
+    }
+    else
+    {
+        ftp_decoder_stat_incrby(thread_idx, fenv, FTPD_STAT_DATA_LINK_BYTES_S2C, (long long)payload_len);
+    }
+}
+
+static void ftp_decoder_entry(struct session *sess, int topic_id, const void *msg, void *per_session_ctx, void *plugin_env)
+{
+    struct ftp_context *fctx = (struct ftp_context *)per_session_ctx;
+    if (FTP_LINK_CTRL == fctx->link_type)
+    {
+        ftp_ctrl_entry(sess, fctx, (struct ftp_decoder_env *)plugin_env);
+    }
+    else
+    {
+        ftp_data_entry(sess, fctx, (struct ftp_decoder_env *)plugin_env);
+    }
+}
+
+int ftp_message_iterate(const struct ftp_message *msg, enum ftp_msg_type *type, struct iovec *value)
+{
+    if (msg->fctx->parse_result.cursor >= FTP_MSG_MAX)
+    {
+        return 0;
+    }
+    int index = msg->fctx->parse_result.cursor;
+    for (int i = index; i < (int)FTP_MSG_MAX; i++)
+    {
+        if (msg->fctx->parse_result.push_result_flags[i] == 1)
+        {
+            continue;
+        }
+        if (msg->fctx->parse_result.result_array[i].iov_base == NULL || msg->fctx->parse_result.result_array[i].iov_len == 0)
+        {
+            continue;
+        }
+        *type = (enum ftp_msg_type)i;
+        value->iov_base = msg->fctx->parse_result.result_array[i].iov_base;
+        value->iov_len = msg->fctx->parse_result.result_array[i].iov_len;
+        msg->fctx->parse_result.push_result_flags[i] = 1;
+        msg->fctx->parse_result.cursor = index + 1;
+        return 1;
+    }
+    return 0;
+}
+
+int ftp_decoder_push_msg(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    struct ftp_message *fmsg = (struct ftp_message *)calloc(1, sizeof(struct ftp_message));
+    fmsg->fctx = fctx;
+    fctx->parse_result.cursor = 0;
+    if (FTP_LINK_CTRL == fctx->link_type)
+    {
+        session_mq_publish_message(sess, fenv->ftp_pub_ctrl_topic_id, fmsg);
+    }
+    else
+    {
+        session_mq_publish_message(sess, fenv->ftp_pub_data_topic_id, fmsg);
+    }
+    return 0;
+}
+
+extern "C" void *FTP_ONLOAD(struct stellar *st)
+{
+    struct ftp_decoder_env *fenv = (struct ftp_decoder_env *)calloc(1, sizeof(struct ftp_decoder_env));
+
+    fenv->thread_count = stellar_get_worker_thread_num(st);
+    assert(fenv->thread_count >= 1);
+    fenv->plugin_id = stellar_session_plugin_register(st, ftp_context_ctx_new_cb, ftp_context_ctx_free_cb, fenv);
+    fenv->tcp_sub_topic_id = stellar_session_mq_get_topic_id(st, TOPIC_TCP_STREAM);
+    stellar_session_mq_subscribe(st, fenv->tcp_sub_topic_id, ftp_decoder_entry, fenv->plugin_id);
+    fenv->ftp_pub_ctrl_topic_id = stellar_session_mq_create_topic(st, FTP_DECODER_CTRL_TOPIC, ftp_msg_free_cb, fenv);
+    fenv->ftp_pub_data_topic_id = stellar_session_mq_create_topic(st, FTP_DECODER_DATA_TOPIC, ftp_msg_free_cb, fenv);
+    ftp_hash_table_create(fenv);
+    ftp_decoder_stat_init(fenv);
+    ftpd_logger_init(fenv);
+    ftp_runtime_log(RLOG_LV_DEBUG, "FTP_ONLOAD success.");
+    return fenv;
+}
+
+extern "C" void FTP_UNLOAD(void *plugin_env)
+{
+    struct ftp_decoder_env *fenv = (struct ftp_decoder_env *)plugin_env;
+    ftp_runtime_log(RLOG_LV_DEBUG, "FTP_UNLOAD ...");
+    ftp_hash_table_destroy(fenv);
+    ftp_decoder_stat_free(fenv);
+    __ftp_decoder_logger_handle = NULL;
+    MESA_destroy_runtime_log_handle(fenv->logger_handle);
+    free(fenv);
+}
\ No newline at end of file
diff --git a/src/ftp_decoder_hash.cpp b/src/ftp_decoder_hash.cpp
new file mode 100644
index 0000000..752505c
--- /dev/null
+++ b/src/ftp_decoder_hash.cpp
@@ -0,0 +1,169 @@
+#include "ftp_decoder_inner.h"
+#include "ftp_decoder_hash.h"
+#include "ftp_decoder_util.h"
+#include <MESA_htable.h>
+#include <cstdint>
+#include <arpa/inet.h>
+
+static thread_local char __ftp_hash_string_buf[1024];
+
+void ftp_declare_datalink_free(void *data)
+{
+    struct ftp_context *fctx = (struct ftp_context *)data;
+    ftp_parse_result_free(&fctx->parse_result);
+    free(fctx);
+}
+
+static void ftp_declare_datalink_free_only_key(void *data)
+{
+    return; // do nothing
+}
+
+int ftp_hash_add(void *handle, const void *key, u_int32_t key_len, void *value)
+{
+    if (MESA_htable_add(handle, (const uint8_t *)key, key_len, value) < 0)
+    {
+        ftp_declare_datalink_free(value);
+        return -1;
+    }
+    return 0;
+}
+
+void ftp_hash_del(void *handle, const struct ftp_link_key *hkey)
+{
+    const void *key;
+    u_int32_t key_len;
+    if (SESSION_ADDR_TYPE_IPV4_TCP == hkey->addr_type || SESSION_ADDR_TYPE_IPV4_UDP == hkey->addr_type)
+    {
+        key = &hkey->tuplev4;
+        key_len = sizeof(struct session_addr_ipv4);
+    }
+    else
+    {
+        key = &hkey->tuplev6;
+        key_len = sizeof(struct session_addr_ipv6);
+    }
+    MESA_htable_del(handle, (const uint8_t *)key, key_len, ftp_declare_datalink_free_only_key);
+}
+
+void ftp_make_hkey_v4(struct session_addr_ipv4 *keyv4, uint32_t sip_net, uint32_t dip_net, uint16_t dip_port_net)
+{
+    memset(keyv4, 0, sizeof(struct session_addr_ipv4));
+    keyv4->saddr = sip_net;
+    keyv4->daddr = dip_net;
+    keyv4->sport = 0;
+    keyv4->dport = dip_port_net;
+}
+
+void ftp_make_hkey_v6(struct session_addr_ipv6 *keyv6, const struct in6_addr *sip, const struct in6_addr *dip, uint16_t dip_port_net)
+{
+    memset(keyv6, 0, sizeof(struct session_addr_ipv6));
+    memcpy(keyv6->saddr, sip, IPV6_ADDR_LEN);
+    memcpy(keyv6->daddr, dip, IPV6_ADDR_LEN);
+    keyv6->sport = 0;
+    keyv6->dport = dip_port_net;
+}
+
+int ftp_build_hashkey_from_session(struct session *sess, struct ftp_link_key *hkey)
+{
+    enum session_addr_type saddr_type;
+    struct session_addr *saddr = session_get0_addr(sess, &saddr_type);
+    hkey->addr_type = saddr_type;
+    if (SESSION_ADDR_TYPE_IPV4_TCP == saddr_type)
+    {
+        ftp_make_hkey_v4(&hkey->tuplev4, saddr->ipv4.saddr, saddr->ipv4.daddr, saddr->ipv4.dport);
+    }
+    else if (SESSION_ADDR_TYPE_IPV6_TCP == saddr_type)
+    {
+        ftp_make_hkey_v6(&hkey->tuplev6, (struct in6_addr *)saddr->ipv6.saddr, (struct in6_addr *)saddr->ipv6.daddr, saddr->ipv6.dport);
+    }
+    else
+    {
+        return -1;
+    }
+    return 0;
+}
+
+void *ftp_hash_search(void *handle, const struct ftp_link_key *hkey)
+{
+    const void *key;
+    u_int32_t key_len;
+    if (SESSION_ADDR_TYPE_IPV4_TCP == hkey->addr_type || SESSION_ADDR_TYPE_IPV4_UDP == hkey->addr_type)
+    {
+        key = &hkey->tuplev4;
+        key_len = sizeof(struct session_addr_ipv4);
+    }
+    else
+    {
+        key = &hkey->tuplev6;
+        key_len = sizeof(struct session_addr_ipv6);
+    }
+    return MESA_htable_search(handle, (const uint8_t *)key, key_len);
+}
+
+const char *ftp_hash_key_to_str(const struct ftp_link_key *hkey)
+{
+    char sip_str[INET6_ADDRSTRLEN], dip_str[INET6_ADDRSTRLEN];
+    unsigned short sport_host, dport_host;
+    if (SESSION_ADDR_TYPE_IPV4_TCP == hkey->addr_type || SESSION_ADDR_TYPE_IPV4_UDP == hkey->addr_type)
+    {
+        inet_ntop(AF_INET, &hkey->tuplev4.saddr, sip_str, INET_ADDRSTRLEN);
+        inet_ntop(AF_INET, &hkey->tuplev4.daddr, dip_str, INET_ADDRSTRLEN);
+        sport_host = ntohs(hkey->tuplev4.sport);
+        dport_host = ntohs(hkey->tuplev4.dport);
+    }
+    else
+    {
+        inet_ntop(AF_INET, hkey->tuplev6.saddr, sip_str, INET6_ADDRSTRLEN);
+        inet_ntop(AF_INET, hkey->tuplev6.daddr, dip_str, INET6_ADDRSTRLEN);
+        sport_host = ntohs(hkey->tuplev6.sport);
+        dport_host = ntohs(hkey->tuplev6.dport);
+    }
+    snprintf(__ftp_hash_string_buf, sizeof(__ftp_hash_string_buf), "%s:%u -> %s:%u", sip_str, sport_host, dip_str, dport_host);
+    return __ftp_hash_string_buf;
+}
+
+int ftp_hash_table_create(struct ftp_decoder_env *fenv)
+{
+    int opt, opt_len;
+    fenv->data_link_table = (void **)calloc(fenv->thread_count, sizeof(void *));
+    for (int i = 0; i < fenv->thread_count; i++)
+    {
+        fenv->data_link_table[i] = MESA_htable_born();
+        assert(fenv->data_link_table[i]);
+
+        opt = 4096;
+        opt_len = sizeof(int);
+        MESA_htable_set_opt(fenv->data_link_table[i], MHO_HASH_SLOT_SIZE, &opt, opt_len);
+
+        opt = 0;
+        opt_len = sizeof(int);
+        MESA_htable_set_opt(fenv->data_link_table[i], MHO_THREAD_SAFE, &opt, opt_len);
+
+        opt = 0;
+        opt_len = sizeof(int);
+        MESA_htable_set_opt(fenv->data_link_table[i], MHO_SCREEN_PRINT_CTRL, &opt, opt_len);
+
+        opt = 10;
+        opt_len = sizeof(int);
+        MESA_htable_set_opt(fenv->data_link_table[i], MHO_EXPIRE_TIME, &opt, opt_len);
+        MESA_htable_set_opt(fenv->data_link_table[i], MHO_CBFUN_DATA_FREE, (void *)&ftp_declare_datalink_free, sizeof(void *));
+        MESA_htable_mature(fenv->data_link_table[i]);
+    }
+    return 0;
+}
+
+void ftp_hash_table_destroy(struct ftp_decoder_env *fenv)
+{
+    unsigned int remain_items;
+    for (int i = 0; i < fenv->thread_count; i++)
+    {
+        remain_items = MESA_htable_get_elem_num(fenv->data_link_table[0]);
+        if (remain_items > 0)
+        {
+            ftp_runtime_log(RLOG_LV_DEBUG, "ftp_hash_table_destroy(): thread:%d declare datalink table remain items:%u\n", i, remain_items);
+        }
+        MESA_htable_destroy(fenv->data_link_table[i], ftp_declare_datalink_free);
+    }
+    free(fenv->data_link_table);
+}
\ No newline at end of file
diff --git a/src/ftp_decoder_hash.h b/src/ftp_decoder_hash.h
new file mode 100644
index 0000000..21a702f
--- /dev/null
+++ b/src/ftp_decoder_hash.h
@@ -0,0 +1,38 @@
+#pragma once
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+#include <stellar/session.h>
+#include <stellar/session_exdata.h>
+#include <stellar/session_mq.h>
+#include <stellar/stellar.h>
+#ifdef __cplusplus
+}
+#endif
+#include <stddef.h>
+#include <MESA/MESA_htable.h>
+
+/*
+    for ftp passive mode, source port is random, so we can't use tuple4 to identify data link.
+    so always set source port to 0!
+*/
+struct ftp_link_key
+{
+    enum session_addr_type addr_type;
+    int __pad__;
+    union
+    {
+        struct session_addr_ipv4 tuplev4;
+        struct session_addr_ipv6 tuplev6;
+    };
+} __attribute__((packed));
+
+int ftp_hash_add(void *handle, const void *key, u_int32_t key_len, void *value);
+void *ftp_hash_search(void *handle, const struct ftp_link_key *hkey);
+void ftp_hash_del(void *handle, const struct ftp_link_key *hkey);
+void ftp_declare_datalink_free(void *data);
+void ftp_make_hkey_v4(struct session_addr_ipv4 *keyv4, uint32_t sip_net, uint32_t dip_net, uint16_t dip_port_net);
+void ftp_make_hkey_v6(struct session_addr_ipv6 *keyv6, const struct in6_addr *sip, const struct in6_addr *dip, uint16_t dip_port_net);
+const char *ftp_hash_key_to_str(const struct ftp_link_key *hkey);
+int ftp_build_hashkey_from_session(struct session *sess, struct ftp_link_key *hkey);
\ No newline at end of file
diff --git a/src/ftp_decoder_inner.h b/src/ftp_decoder_inner.h
new file mode 100644
index 0000000..9f632a1
--- /dev/null
+++ b/src/ftp_decoder_inner.h
@@ -0,0 +1,125 @@
+#pragma once
+#include <stdio.h>
+#include <stdlib.h>
+#include <assert.h>
+#include <string.h>
+#include "ftp_decoder.h"
+#include <MESA/MESA_handle_logger.h>
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+#include <stellar/session.h>
+#include <stellar/session_exdata.h>
+#include <stellar/session_mq.h>
+#include <stellar/stellar.h>
+#ifdef __cplusplus
+}
+#endif
+#include "ftp_decoder_stat.h"
+#include "ftp_decoder_hash.h"
+
+#define FTPD_IDENTIRY_MIN_LEN 4
+#define FTPD_IDENTIRY_MAX_LEN 32
+#define FTPD_CMD_MAX_LENGTH 128
+#define FTP_DECODER_CFG_FILE "./conf/ftp_decoder/ftp.conf"
+#define FTP_DECODER_FIELDSTAT_NAME "ftp_decoder_statistics"
+#define FTP_DECODER_FIELDSTAT_OUTPUT_FILE "./metrics/ftp_decoder_fs4.json"
+#define FTP_DECODER_FIELDSTAT_OUTPUT_INTERVAL 3
+
+extern void *__ftp_decoder_logger_handle;
+#define ftp_runtime_log(log_level, format, ...)                                                                    \
+    do                                                                                                             \
+    {                                                                                                              \
+        if (MESA_handle_runtime_log_level_enabled(__ftp_decoder_logger_handle, log_level))                         \
+        {                                                                                                          \
+            MESA_handle_runtime_log(__ftp_decoder_logger_handle, log_level, "ftp_decoder", format, ##__VA_ARGS__); \
+        }                                                                                                          \
+    } while (0)
+
+#define IOVEC_PRINT(iov) (int)(iov).iov_len, (char *)(iov).iov_base
+#define IOVEC_PRINT_PTR(iov_p) (int)(iov_p->iov_len), (char *)(iov_p->iov_base)
+#ifndef fstring
+typedef struct iovec fstring;
+#endif
+enum ftp_link_type
+{
+    FTP_LINK_CTRL,
+    FTP_LINK_DATA,
+};
+
+enum ftp_data_link_type
+{
+    FTP_DATA_LINK_FILE,
+    FTP_DATA_LINK_INVENTORY,
+};
+
+struct ftp_interact_line
+{
+    fstring cmd_refer; // pointer to packet payload
+    fstring arg_refer; // pointer to packet payload
+};
+
+struct ftp_parse_result
+{
+    /* all data is persistent, sync with data link, need be free when session close */
+    int cursor;
+    struct iovec result_array[FTP_MSG_MAX];
+    uint8_t push_result_flags[FTP_MSG_MAX];
+};
+
+struct ftp_context
+{
+    enum ftp_link_type link_type;
+    enum ftp_msg_type data_link_presentation; // FTP_INVENTORY or FTP_FILE_CONTENT
+    struct ftp_interact_line cmd_result;
+    struct ftp_parse_result parse_result;
+    struct ftp_interact_line last_cmd;
+    struct ftp_link_key last_data_link_key;
+};
+
+struct ftp_interact_parser
+{
+    enum ftp_msg_type cmd_msg_type;
+    const char *cmd_name;
+    int cmd_len;
+    int (*cmd_handler)(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv);
+};
+
+struct ftp_stat
+{
+    struct fieldstat_easy *fs4_instance;
+    int fs4_counter_id[FTPD_STAT_MAX];
+};
+
+struct ftp_decoder_env
+{
+    void *logger_handle;
+    int plugin_id;
+    int thread_count;
+    int tcp_sub_topic_id;
+    int ftp_pub_ctrl_topic_id;
+    int ftp_pub_data_topic_id;
+    void **data_link_table; // key is tuple3 for PASV, tuple4 for PORT
+    struct ftp_stat fstat;
+};
+
+struct ftp_message
+{
+    struct ftp_context *fctx;
+};
+
+extern "C" void *FTP_ONLOAD(struct stellar *st);
+extern "C" void FTP_UNLOAD(void *plugin_env);
+int ftp_hash_table_create(struct ftp_decoder_env *fenv);
+void ftp_hash_table_destroy(struct ftp_decoder_env *fenv);
+int ftp_ctrl_identify_by_payload(const char *payload, size_t len, u_int8_t curdir);
+int ftp_ctrl_identify_by_addr(struct session *sess);
+int ftp_ctrl_identify(struct session *sess, const char *payload, size_t len, u_int8_t curdir);
+struct ftp_context *ftp_data_identify(struct session *sess, struct ftp_decoder_env *fenv);
+int ftp_cmd_readline(struct ftp_interact_line *line, const char *payload, size_t len);
+int ftp_cmd_process(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv, int curdir);
+int ftp_decoder_push_msg(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv);
+int ftp_parse_ipv4_port_style(const fstring *cmd_str, unsigned int *ipv4_net, unsigned short *port_net);
+int ftp_parse_ipv6_port_style(const fstring *cmd_str, unsigned short *port_net);
+int ftp_parse_eprt_ipport_style(const fstring *arg_str, struct in6_addr *ipd_addr, unsigned short *port_net);
\ No newline at end of file
diff --git a/src/ftp_decoder_proto.cpp b/src/ftp_decoder_proto.cpp
new file mode 100644
index 0000000..dd5f14a
--- /dev/null
+++ b/src/ftp_decoder_proto.cpp
@@ -0,0 +1,566 @@
+#include "ftp_decoder_inner.h"
+#include "ftp_decoder.h"
+#include "ftp_decoder_util.h"
+#include "ftp_decoder_hash.h"
+#include <sys/types.h>
+#include <arpa/inet.h>
+
+/*
+                                            -------------
+                                            |/---------\|
+                                            ||   User  ||    --------
+                                            ||Interface|<--->| User |
+                                            |\----^----/|    --------
+                  ----------                |     |     |
+                  |/------\|  FTP Commands  |/----V----\|
+                  ||Server|<---------------->|   User  ||
+                  ||  PI  ||   FTP Replies  ||    PI   ||
+                  |\--^---/|                |\----^----/|
+                  |   |    |                |     |     |
+      --------    |/--V---\|      Data      |/----V----\|    --------
+      | File |<--->|Server|<---------------->|  User   |<--->| File |
+      |System|    || DTP  ||   Connection   ||   DTP   ||    |System|
+      --------    |\------/|                |\---------/|    --------
+                  ----------                -------------
+
+                  Server-FTP                   USER-FTP
+*/
+
+/*
+    ctrl link command               data link tcp/ip stack event
+---------------------------------------------------------------------------------------------------
+    PORT 192,168,38,2,202,95
+                                    listen tcp port: 51807
+    200 PORT command successful.
+    LIST
+                                    server use src tcp port 20 connecting to client tcp port 51807
+                                    data connection is established
+                                    send 'ls -l' result to cliet
+    226 Directory send OK.
+                                    close the tcp data connection
+    waiting for next command...
+*/
+static const struct iovec G_FTP_EMPTY_IOV = {NULL, 0};
+
+static struct ftp_context *ftp_get_declare_datalink_ctx(struct session *sess, struct ftp_context *ctrl_fctx, struct ftp_decoder_env *fenv)
+{
+    int thread_id = session_get_current_thread_id(sess);
+    struct ftp_context *declare_data_link_ftx = (struct ftp_context *)ftp_hash_search(fenv->data_link_table[thread_id], &ctrl_fctx->last_data_link_key);
+    if (NULL == declare_data_link_ftx)
+    {
+        ftp_runtime_log(RLOG_LV_FATAL, "declare data link not found, key=%s", ftp_hash_key_to_str(&ctrl_fctx->last_data_link_key));
+        return NULL;
+    }
+    return declare_data_link_ftx;
+}
+
+static void ftp_assemble_uri(struct session *sess, const struct iovec *raw_arg, struct iovec *uri)
+{
+    char dip_str[INET6_ADDRSTRLEN] = {};
+    char tmp_url[2048] = {};
+    enum session_addr_type saddr_type;
+    struct session_addr *saddr = session_get0_addr(sess, &saddr_type);
+    if (saddr_type == SESSION_ADDR_TYPE_IPV4_TCP)
+    {
+        inet_ntop(AF_INET, &saddr->ipv4.daddr, dip_str, INET_ADDRSTRLEN);
+    }
+    else
+    {
+        inet_ntop(AF_INET6, saddr->ipv6.daddr, dip_str, INET6_ADDRSTRLEN);
+    }
+    int ret = snprintf(tmp_url, sizeof(tmp_url), "ftp://%s/%.*s", dip_str, IOVEC_PRINT(*raw_arg));
+    if (uri->iov_base)
+    {
+        free(uri->iov_base);
+    }
+    uri->iov_base = calloc(1, ret);
+    memcpy(uri->iov_base, tmp_url, ret);
+    uri->iov_len = ret;
+}
+
+int ftp_cmd_handler_do_user(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    fstring_safe_dup(&fctx->cmd_result.arg_refer, &fctx->parse_result.result_array[FTP_ACCOUNT]);
+    ftp_runtime_log(RLOG_LV_DEBUG, "'%s': USER: %.*s", session_get0_readable_addr(sess), IOVEC_PRINT(fctx->parse_result.result_array[FTP_ACCOUNT]));
+    return 0;
+}
+
+int ftp_cmd_handler_do_pass(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    fstring_safe_dup(&fctx->cmd_result.arg_refer, &fctx->parse_result.result_array[FTP_PASSWORD]);
+    ftp_runtime_log(RLOG_LV_DEBUG, "'%s': PASS: %.*s", session_get0_readable_addr(sess), IOVEC_PRINT(fctx->parse_result.result_array[FTP_PASSWORD]));
+    return 0;
+}
+
+int ftp_parse_ipv4_port_style(const fstring *cmd_str, unsigned int *ipv4_net, unsigned short *port_net)
+{
+    unsigned int fields[6];
+    char raw_cmd_str_tmp[cmd_str->iov_len + 1] = {};
+    memcpy(raw_cmd_str_tmp, cmd_str->iov_base, cmd_str->iov_len);
+    char only_integer_str_tmp[cmd_str->iov_len + 1] = {};
+    if (sscanf(raw_cmd_str_tmp, "%*[a-zA-Z (]%[0-9,]", only_integer_str_tmp) <= 0)
+    {
+        if (sscanf(raw_cmd_str_tmp, "%[1234567890,]", only_integer_str_tmp) <= 0)
+        {
+            ftp_runtime_log(RLOG_LV_FATAL, "ftp_parse_ipv4_port_style parse error: %.*s", IOVEC_PRINT_PTR(cmd_str));
+            return -1;
+        }
+    }
+    int ret = sscanf(only_integer_str_tmp, "%u,%u,%u,%u,%u,%u", &fields[0], &fields[1], &fields[2], &fields[3], &fields[4], &fields[5]);
+    if (ret != 6)
+    {
+        ftp_runtime_log(RLOG_LV_FATAL, "ftp_parse_ipv4_port_style parse error: %.*s", cmd_str->iov_len, only_integer_str_tmp);
+        return -1;
+    }
+    unsigned int dst_ip_host = (fields[0] << 24) | (fields[1] << 16) | (fields[2] << 8) | fields[3];
+    unsigned short dst_port_host = (fields[4] << 8) | fields[5];
+    *ipv4_net = htonl(dst_ip_host);
+    *port_net = htons(dst_port_host);
+    return 0;
+}
+/* PORT command only support IPv4, arg pattern: h1,h2,h3,h4,p1,p2 */
+int ftp_cmd_handler_do_port(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    fstring_safe_dup(&fctx->cmd_result.cmd_refer, &fctx->parse_result.result_array[FTP_TRANS_MODE]);
+    fctx->parse_result.push_result_flags[FTP_TRANS_MODE] = 0;
+
+    unsigned int dst_ip_net;
+    unsigned short dst_port_net;
+    if (ftp_parse_ipv4_port_style(&fctx->cmd_result.arg_refer, &dst_ip_net, &dst_port_net) < 0)
+    {
+        return -1;
+    }
+    enum session_addr_type saddr_type;
+    struct session_addr *saddr = session_get0_addr(sess, &saddr_type);
+    fctx->last_data_link_key.addr_type = SESSION_ADDR_TYPE_IPV4_TCP;
+    // in active mode, new data link src ip address is server ip address
+    ftp_make_hkey_v4(&fctx->last_data_link_key.tuplev4, saddr->ipv4.daddr, dst_ip_net, dst_port_net);
+
+    struct ftp_context *new_data_link_ctx = ftp_decoder_context_deep_dup(fctx);
+    int thread_id = session_get_current_thread_id(sess);
+    ftp_hash_add(fenv->data_link_table[thread_id], &fctx->last_data_link_key.tuplev4, sizeof(struct session_addr_ipv4), new_data_link_ctx);
+
+    ftp_decoder_stat_incrby(thread_id, fenv, FTPD_STAT_NEGOTIATE_DATA_LINK, 1);
+
+    ftp_runtime_log(RLOG_LV_DEBUG, "do_port_cb: %.*s, parsed dip:%x, dport:%u", IOVEC_PRINT(fctx->parse_result.result_array[FTP_TRANS_MODE]), dst_ip_net, dst_port_net);
+    return 0;
+}
+
+/* PASV command only support IPv4, the response like: '227 Entering Passive Mode (218,13,32,6,78,40).' */
+int ftp_cmd_handler_do_pasv(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    fstring_safe_dup(&fctx->cmd_result.cmd_refer, &fctx->parse_result.result_array[FTP_TRANS_MODE]);
+    fctx->parse_result.push_result_flags[FTP_TRANS_MODE] = 0;
+    ftp_runtime_log(RLOG_LV_DEBUG, "'%s' PASV command", session_get0_readable_addr(sess));
+    return 0;
+}
+
+int ftp_cmd_handler_do_list(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    fstring_safe_dup(&fctx->cmd_result.cmd_refer, &fctx->parse_result.result_array[FTP_TRANS_DIR]);
+    fstring_safe_dup(&G_FTP_EMPTY_IOV, &fctx->parse_result.result_array[FTP_URI]);
+    struct ftp_context *declare_data_link_ftx = ftp_get_declare_datalink_ctx(sess, fctx, fenv);
+    if (declare_data_link_ftx)
+    {
+        fstring_safe_dup(&fctx->cmd_result.cmd_refer, &declare_data_link_ftx->parse_result.result_array[FTP_TRANS_DIR]);
+        fstring_safe_dup(&G_FTP_EMPTY_IOV, &declare_data_link_ftx->parse_result.result_array[FTP_URI]);
+        declare_data_link_ftx->data_link_presentation = FTP_INVENTORY;
+    }
+    ftp_runtime_log(RLOG_LV_DEBUG, "'%s': LIST command", session_get0_readable_addr(sess));
+    return 0;
+}
+
+static void ftp_cmd_handler_do_stor_retr_common(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    fstring_safe_dup(&fctx->cmd_result.cmd_refer, &fctx->parse_result.result_array[FTP_TRANS_DIR]);
+    ftp_assemble_uri(sess, &fctx->cmd_result.arg_refer, &fctx->parse_result.result_array[FTP_URI]);
+    fctx->parse_result.push_result_flags[FTP_TRANS_DIR] = 0;
+    fctx->parse_result.push_result_flags[FTP_URI] = 0;
+
+    struct ftp_context *declare_data_link_ftx = ftp_get_declare_datalink_ctx(sess, fctx, fenv);
+    if (declare_data_link_ftx)
+    {
+        declare_data_link_ftx->data_link_presentation = FTP_FILE_CONTENT;
+        fstring_safe_dup(&fctx->parse_result.result_array[FTP_TRANS_DIR], &declare_data_link_ftx->parse_result.result_array[FTP_TRANS_DIR]);
+        fstring_safe_dup(&fctx->parse_result.result_array[FTP_URI], &declare_data_link_ftx->parse_result.result_array[FTP_URI]);
+        // ftp_update_parse_result(declare_data_link_ftx, &fctx->parse_result.result_array[FTP_URI], FTP_URI);
+    }
+}
+
+int ftp_cmd_handler_do_stor(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    ftp_cmd_handler_do_stor_retr_common(sess, fctx, fenv);
+    ftp_runtime_log(RLOG_LV_DEBUG, "'%s', STOR uri:%.*s", session_get0_readable_addr(sess), IOVEC_PRINT(fctx->parse_result.result_array[FTP_URI]));
+    return 0;
+}
+
+int ftp_cmd_handler_do_retr(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    ftp_cmd_handler_do_stor_retr_common(sess, fctx, fenv);
+    ftp_runtime_log(RLOG_LV_DEBUG, "'%s', RETR uri:%.*s", session_get0_readable_addr(sess), IOVEC_PRINT(fctx->parse_result.result_array[FTP_URI]));
+    return 0;
+}
+
+int ftp_parse_eprt_ipport_style(const fstring *arg_str, struct in6_addr *ipd_addr, unsigned short *port_net)
+{
+    unsigned int port_host;
+    int inet_proto;
+    char ip6_addr_str[INET6_ADDRSTRLEN] = {};
+    char raw_cmd_str_tmp[arg_str->iov_len + 1] = {};
+    memcpy(raw_cmd_str_tmp, arg_str->iov_base, arg_str->iov_len);
+
+    char *save_ptr, *ptr;
+    const char *delim = " |\t";
+    ptr = strtok_r(raw_cmd_str_tmp, delim, &save_ptr);
+    if (NULL == ptr)
+    {
+        ftp_runtime_log(RLOG_LV_FATAL, "ftp_parse_eprt_ipport_style parse error: %.*s", IOVEC_PRINT_PTR(arg_str));
+        return -1;
+    }
+    inet_proto = atoi(ptr);
+    if (2 != inet_proto)
+    {
+        ftp_runtime_log(RLOG_LV_FATAL, "ftp_parse_eprt_ipport_style parse error: %.*s, not support inet: %d", IOVEC_PRINT_PTR(arg_str), inet_proto);
+        return -1;
+    }
+    ptr = strtok_r(NULL, delim, &save_ptr);
+    if (NULL == ptr)
+    {
+        ftp_runtime_log(RLOG_LV_FATAL, "ftp_parse_eprt_ipport_style parse error: %.*s", IOVEC_PRINT_PTR(arg_str));
+        return -1;
+    }
+    strncpy(ip6_addr_str, ptr, INET6_ADDRSTRLEN - 1);
+    ptr = strtok_r(NULL, delim, &save_ptr);
+    if (NULL == ptr)
+    {
+        ftp_runtime_log(RLOG_LV_FATAL, "ftp_parse_eprt_ipport_style parse error: %.*s", IOVEC_PRINT_PTR(arg_str));
+        return -1;
+    }
+    port_host = (unsigned int)atoi(ptr);
+
+    while (strtok_r(NULL, "|", &save_ptr))
+        ;
+    inet_pton(AF_INET6, ip6_addr_str, ipd_addr);
+    *port_net = htons((unsigned short)port_host);
+    return 0;
+}
+
+/* EPRT support IPv4 and IPv6, pattern: EPRT<space><d><net-prt><d><net-addr><d><tcp-port><d>
+   example:
+       EPRT |1|132.235.1.2|6275|
+       EPRT |2|1080::8:800:200C:417A|5282|
+   refer: https://datatracker.ietf.org/doc/html/rfc2428#section-2
+*/
+int ftp_cmd_handler_do_eprt(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    struct in6_addr ipd_addr;
+    unsigned short port_net;
+    fstring_safe_dup(&fctx->cmd_result.cmd_refer, &fctx->parse_result.result_array[FTP_TRANS_MODE]);
+    if (ftp_parse_eprt_ipport_style(&fctx->cmd_result.arg_refer, &ipd_addr, &port_net) < 0)
+    {
+        return -1;
+    }
+
+    enum session_addr_type saddr_type;
+    struct session_addr *saddr = session_get0_addr(sess, &saddr_type);
+    fctx->last_data_link_key.addr_type = SESSION_ADDR_TYPE_IPV6_TCP;
+    // in active mode, new data link src ip address is server ip address
+    ftp_make_hkey_v6(&fctx->last_data_link_key.tuplev6, (struct in6_addr *)&saddr->ipv6.daddr, &ipd_addr, port_net);
+
+    struct ftp_context *new_data_link_ctx = ftp_decoder_context_deep_dup(fctx);
+    int thread_id = session_get_current_thread_id(sess);
+    ftp_hash_add(fenv->data_link_table[thread_id], &fctx->last_data_link_key.tuplev6, sizeof(struct session_addr_ipv6), new_data_link_ctx);
+
+    ftp_decoder_stat_incrby(thread_id, fenv, FTPD_STAT_NEGOTIATE_DATA_LINK, 1);
+
+    ftp_runtime_log(RLOG_LV_DEBUG, "'%s': EPRT command, port:%u", session_get0_readable_addr(sess), ntohs(port_net));
+    return 0;
+}
+
+/* LPRT support IPv4 and IPv6, pattern:  LPRT af,hal,h1,h2,h3,h4...,pal,p1,p2...
+   example:
+       LPRT 6,16,32,2,81,131,67,131,0,0,0,0,0,0,81,131,67,131,2,4,7
+   refer: https://www.rfc-editor.org/rfc/rfc1639.html
+*/
+int ftp_cmd_handler_do_lprt(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    // todo
+    ftp_runtime_log(RLOG_LV_INFO, "LPRT command not support yet!");
+    return -1;
+}
+
+/* EPSV support IPv4 and IPv6,
+   refer: https://datatracker.ietf.org/doc/html/rfc2428#autoid-3
+*/
+int ftp_cmd_handler_do_epsv(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    // todo
+    ftp_runtime_log(RLOG_LV_INFO, "EPSV command not support yet!");
+    return -1;
+}
+
+/* LPSV support IPv4 and IPv6, response is 228 xxxx...
+   refer: https://www.rfc-editor.org/rfc/rfc1639.html
+*/
+int ftp_cmd_handler_do_lpsv(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    // todo
+    ftp_runtime_log(RLOG_LV_INFO, "LPSV command not support yet!");
+    return -1;
+}
+
+static const struct ftp_interact_parser g_ftp_client_cmd_tuple[] =
+    {
+        {FTP_ACCOUNT, "USER", 4, ftp_cmd_handler_do_user},
+        {FTP_PASSWORD, "PASS", 4, ftp_cmd_handler_do_pass},
+        {FTP_TRANS_MODE, "PORT", 4, ftp_cmd_handler_do_port},
+        {FTP_TRANS_MODE, "PASV", 4, ftp_cmd_handler_do_pasv},
+        {FTP_TRANS_MODE, "EPRT", 4, ftp_cmd_handler_do_eprt},
+        {FTP_TRANS_MODE, "EPSV", 4, ftp_cmd_handler_do_epsv},
+        {FTP_TRANS_MODE, "LPRT", 4, ftp_cmd_handler_do_lprt},
+        {FTP_TRANS_MODE, "LPSV", 4, ftp_cmd_handler_do_lpsv},
+        {FTP_MSG_MAX, "LIST", 4, ftp_cmd_handler_do_list},
+        {FTP_TRANS_DIR, "STOR", 4, ftp_cmd_handler_do_stor},
+        {FTP_TRANS_DIR, "RETR", 4, ftp_cmd_handler_do_retr},
+        {FTP_MSG_MAX, NULL, 0, NULL}};
+
+int ftp_res_handler_do_220(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    fstring_safe_dup(&fctx->cmd_result.arg_refer, &fctx->parse_result.result_array[FTP_BANNER]);
+    return 0;
+}
+
+int ftp_res_handler_do_200(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    // todo
+    ftp_runtime_log(RLOG_LV_INFO, "'%s': response 200 command not support yet!", session_get0_readable_addr(sess));
+    return -1;
+}
+
+/*
+  example: 227 Entering Passive Mode (218,13,32,6,78,40).
+ */
+int ftp_res_handler_do_227(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    unsigned int dst_ip_net;
+    unsigned short dst_port_net;
+    if (ftp_parse_ipv4_port_style(&fctx->cmd_result.arg_refer, &dst_ip_net, &dst_port_net) < 0)
+    {
+        return -1;
+    }
+    enum session_addr_type saddr_type;
+    struct session_addr *saddr = session_get0_addr(sess, &saddr_type);
+    fctx->last_data_link_key.addr_type = SESSION_ADDR_TYPE_IPV4_TCP;
+    // in passive mode, new data link src ip address is client ip address
+    ftp_make_hkey_v4(&fctx->last_data_link_key.tuplev4, saddr->ipv4.saddr, dst_ip_net, dst_port_net);
+
+    struct ftp_context *new_data_link_ctx = ftp_decoder_context_deep_dup(fctx);
+    int thread_id = session_get_current_thread_id(sess);
+    ftp_hash_add(fenv->data_link_table[thread_id], &fctx->last_data_link_key.tuplev4, sizeof(struct session_addr_ipv4), new_data_link_ctx);
+    ftp_decoder_stat_incrby(thread_id, fenv, FTPD_STAT_NEGOTIATE_DATA_LINK, 1);
+    ftp_runtime_log(RLOG_LV_DEBUG, "'%s': response 227, %.*s parsed dip:0x%x, dport:%u", session_get0_readable_addr(sess), IOVEC_PRINT(fctx->cmd_result.arg_refer), ntohl(dst_ip_net), ntohs(dst_port_net));
+    return 0;
+}
+
+/*
+  example: 228 Entering Long Passive Mode (af, hal, h1, h2, h3,..., pal, p1, p2...)
+  refer: https://www.rfc-editor.org/rfc/rfc1639.html
+ */
+int ftp_res_handler_do_228(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    // todo
+    ftp_runtime_log(RLOG_LV_INFO, "'%s': response 228 command not support yet!", session_get0_readable_addr(sess));
+    return -1;
+}
+
+int ftp_parse_ipv6_port_style(const fstring *cmd_str, unsigned short *port_net)
+{
+    unsigned int port_host;
+    char raw_cmd_str_tmp[cmd_str->iov_len + 1] = {};
+    memcpy(raw_cmd_str_tmp, cmd_str->iov_base, cmd_str->iov_len);
+    char only_integer_str_tmp[cmd_str->iov_len + 1] = {};
+    if (sscanf(raw_cmd_str_tmp, "%*[a-zA-Z (|]%[0-9]", only_integer_str_tmp) <= 0)
+    {
+        if (sscanf(raw_cmd_str_tmp, "|||%s|", only_integer_str_tmp) <= 0)
+        {
+            ftp_runtime_log(RLOG_LV_FATAL, "ftp_parse_ipv4_port_style parse error: %.*s", IOVEC_PRINT_PTR(cmd_str));
+            return -1;
+        }
+    }
+    int ret = sscanf(only_integer_str_tmp, "%u", &port_host);
+    if (ret != 1)
+    {
+        ftp_runtime_log(RLOG_LV_FATAL, "ftp_parse_ipv6_port_style parse error: %.*s", cmd_str->iov_len, only_integer_str_tmp);
+        return -1;
+    }
+    *port_net = htons((unsigned short)port_host);
+    return 0;
+}
+/*
+  example: 229 229 Entering Extended Passive Mode (|||20987|)
+  refer: https://datatracker.ietf.org/doc/html/rfc2428#autoid-3
+ */
+int ftp_res_handler_do_229(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv)
+{
+    unsigned short port_net;
+    if (ftp_parse_ipv6_port_style(&fctx->cmd_result.arg_refer, &port_net) < 0)
+    {
+        ftp_runtime_log(RLOG_LV_FATAL, "'%s': response 229 parse error, %.*s", session_get0_readable_addr(sess), IOVEC_PRINT(fctx->cmd_result.arg_refer));
+        return -1;
+    }
+
+    enum session_addr_type saddr_type;
+    struct session_addr *saddr = session_get0_addr(sess, &saddr_type);
+    if (saddr_type != SESSION_ADDR_TYPE_IPV6_TCP)
+    {
+        ftp_runtime_log(RLOG_LV_FATAL, "'%s': response 229, but addr type is not ipv6 %.*s", session_get0_readable_addr(sess), IOVEC_PRINT(fctx->cmd_result.arg_refer));
+        return -1;
+    }
+    fctx->last_data_link_key.addr_type = SESSION_ADDR_TYPE_IPV6_TCP;
+    // in passive mode, new data link src ip address is client ip address
+    ftp_make_hkey_v6(&fctx->last_data_link_key.tuplev6, (struct in6_addr *)saddr->ipv6.saddr, (struct in6_addr *)saddr->ipv6.daddr, port_net);
+
+    struct ftp_context *new_data_link_ctx = ftp_decoder_context_deep_dup(fctx);
+    int thread_id = session_get_current_thread_id(sess);
+    ftp_hash_add(fenv->data_link_table[thread_id], &fctx->last_data_link_key.tuplev6, sizeof(struct session_addr_ipv6), new_data_link_ctx);
+    ftp_decoder_stat_incrby(thread_id, fenv, FTPD_STAT_NEGOTIATE_DATA_LINK, 1);
+    ftp_runtime_log(RLOG_LV_DEBUG, "'%s': EPSV response 229, port is:%u", session_get0_readable_addr(sess), ntohs(port_net));
+    return -1;
+}
+
+static const struct ftp_interact_parser g_ftp_server_response_tuple[] =
+    {
+        {FTP_BANNER, "220", 3, ftp_res_handler_do_220},
+        {FTP_MSG_MAX, "200", 3, ftp_res_handler_do_200},
+        {FTP_MSG_MAX, "227", 3, ftp_res_handler_do_227},
+        {FTP_MSG_MAX, "228", 3, ftp_res_handler_do_228},
+        {FTP_MSG_MAX, "229", 3, ftp_res_handler_do_229},
+        {FTP_MSG_MAX, NULL, 0, NULL}};
+
+int ftp_ctrl_identify_by_payload(const char *payload, size_t len, u_int8_t curdir)
+{
+    if (NULL == payload || len < FTPD_IDENTIRY_MIN_LEN)
+    {
+        return 0;
+    }
+    if (curdir == PACKET_DIRECTION_C2S)
+    {
+        if (memcmp(payload, "USER", 4) == 0)
+        {
+            return 1;
+        }
+    }
+    else
+    {
+        if (memcmp(payload, "220", 3) == 0)
+        {
+            int tmplen = MIN(len, FTPD_IDENTIRY_MAX_LEN);
+            char tmp_buffer[tmplen];
+            memcpy(tmp_buffer, payload, tmplen);
+            ftp_strtolower(tmp_buffer, tmplen);
+            if ((memmem(tmp_buffer, tmplen, "ftp", 3) != NULL))
+            {
+                return 1;
+            }
+        }
+    }
+    return 0;
+}
+
+int ftp_ctrl_identify_by_addr(struct session *sess)
+{
+    enum session_addr_type saddr_type = SESSION_ADDR_TYPE_UNKNOWN;
+    struct session_addr *saddr = session_get0_addr(sess, &saddr_type);
+    unsigned short sport_host = 0, dport_host = 0;
+    if (saddr_type == SESSION_ADDR_TYPE_IPV4_TCP)
+    {
+        sport_host = ntohs(saddr->ipv4.sport);
+        dport_host = ntohs(saddr->ipv4.dport);
+    }
+    else if (saddr_type == SESSION_ADDR_TYPE_IPV6_TCP)
+    {
+        sport_host = ntohs(saddr->ipv6.sport);
+        dport_host = ntohs(saddr->ipv6.dport);
+    }
+    else
+    {
+        return 0;
+    }
+    if (sport_host == 21 || dport_host == 21)
+    {
+        return 1;
+    }
+    if (sport_host == 25 || dport_host == 25)
+    {
+        return 0;
+    }
+    if (sport_host == 110 || dport_host == 110)
+    {
+        return 0;
+    }
+    return 0;
+}
+
+int ftp_ctrl_identify(struct session *sess, const char *payload, size_t len, u_int8_t curdir)
+{
+    if (ftp_ctrl_identify_by_addr(sess))
+    {
+        return 1;
+    }
+    return ftp_ctrl_identify_by_payload(payload, len, curdir);
+}
+
+struct ftp_context *ftp_data_identify(struct session *sess, struct ftp_decoder_env *fenv)
+{
+    struct ftp_link_key hash_key = {};
+    if (ftp_build_hashkey_from_session(sess, &hash_key) < 0)
+    {
+        return NULL;
+    }
+    int thread_id = session_get_current_thread_id(sess);
+    struct ftp_context *declare_fctx = (struct ftp_context *)ftp_hash_search(fenv->data_link_table[thread_id], &hash_key);
+    if (NULL == declare_fctx)
+    {
+        return NULL;
+    }
+    // use tuple3 as hash key, so free it immediately after found!
+    ftp_hash_del(fenv->data_link_table[thread_id], &hash_key);
+    return declare_fctx;
+}
+
+static const struct ftp_interact_parser *ftp_fetch_cmd_parser(const struct ftp_interact_parser *table, const struct ftp_interact_line *cmd_result)
+{
+    for (int i = 0; table[i].cmd_name != NULL; i++)
+    {
+        if (0 == strncmp(table[i].cmd_name, (char *)cmd_result->cmd_refer.iov_base, table[i].cmd_len))
+        {
+            return &table[i];
+        }
+    }
+    return NULL;
+}
+
+int ftp_cmd_process(struct session *sess, struct ftp_context *fctx, struct ftp_decoder_env *fenv, int curdir)
+{
+    const struct ftp_interact_parser *parser;
+    if (PACKET_DIRECTION_C2S == curdir)
+    {
+        parser = ftp_fetch_cmd_parser(g_ftp_client_cmd_tuple, &fctx->cmd_result);
+    }
+    else
+    {
+        parser = ftp_fetch_cmd_parser(g_ftp_server_response_tuple, &fctx->cmd_result);
+    }
+    if (NULL == parser)
+    {
+        ftp_runtime_log(RLOG_LV_INFO, "ftp_fetch_cmd_parser() failed, '%s', not support cmd %.*s", session_get0_readable_addr(sess), IOVEC_PRINT(fctx->cmd_result.cmd_refer));
+        return -1;
+    }
+    if (parser->cmd_handler(sess, fctx, fenv) != -1)
+    {
+        fctx->parse_result.push_result_flags[parser->cmd_msg_type] = 0;
+        ftp_decoder_push_msg(sess, fctx, fenv);
+        ftp_decoder_stat_incrby(session_get_current_thread_id(sess), fenv, FTPD_STAT_CTRL_CMD, 1);
+    }
+    return 0;
+}
diff --git a/src/ftp_decoder_stat.cpp b/src/ftp_decoder_stat.cpp
new file mode 100644
index 0000000..ac5f1ce
--- /dev/null
+++ b/src/ftp_decoder_stat.cpp
@@ -0,0 +1,54 @@
+#include <assert.h>
+#include <fieldstat/fieldstat_easy.h>
+#include "ftp_decoder_stat.h"
+#include "ftp_decoder_inner.h"
+#include "ftp_decoder_util.h"
+
+static const struct ftpd_stat_config_tuple g_ftpd_stat_tuple[] =
+    {
+        {FTPD_STAT_CTRL_LINK_OPEN, "clink_open"},
+        {FTPD_STAT_CTRL_LINK_CLOSE, "clink_close"},
+        {FTPD_STAT_DATA_LINK_OPEN, "dlink_open"},
+        {FTPD_STAT_DATA_LINK_CLOSE, "dlink_close"},
+        {FTPD_STAT_NEGOTIATE_DATA_LINK, "negotiate_dlink"},
+        {FTPD_STAT_CTRL_LINK_BYTES_C2S, "clink_bytes_c2s"},
+        {FTPD_STAT_CTRL_LINK_BYTES_S2C, "clink_bytes_s2c"},
+        {FTPD_STAT_DATA_LINK_BYTES_C2S, "dlink_bytes_c2s"},
+        {FTPD_STAT_DATA_LINK_BYTES_S2C, "dlink_bytes_s2c"},
+        {FTPD_STAT_CTRL_CMD, "ctrl_cmd"},
+};
+
+void ftp_decoder_stat_incrby(int thread_idx, struct ftp_decoder_env *fenv, enum ftp_decoder_stat_type stattype, long long increment)
+{
+    if (fenv->fstat.fs4_instance)
+    {
+        fieldstat_easy_counter_incrby(fenv->fstat.fs4_instance, thread_idx, fenv->fstat.fs4_counter_id[stattype], NULL, 0, increment);
+    }
+}
+
+int ftp_decoder_stat_init(struct ftp_decoder_env *fenv)
+{
+    assert(sizeof(g_ftpd_stat_tuple) / sizeof(struct ftpd_stat_config_tuple) == FTPD_STAT_MAX);
+    fenv->fstat.fs4_instance = fieldstat_easy_new(fenv->thread_count, FTP_DECODER_FIELDSTAT_NAME, NULL, 0);
+
+    for (int i = 0; i < FTPD_STAT_MAX; i++)
+    {
+        fenv->fstat.fs4_counter_id[i] = fieldstat_easy_register_counter(fenv->fstat.fs4_instance, g_ftpd_stat_tuple[i].name);
+    }
+    ftp_mkdir_p("./metrics", 0644);
+    int ret = fieldstat_easy_enable_auto_output(fenv->fstat.fs4_instance, FTP_DECODER_FIELDSTAT_OUTPUT_FILE, FTP_DECODER_FIELDSTAT_OUTPUT_INTERVAL);
+    if (ret < 0)
+    {
+        fprintf(stderr, "fieldstat_easy_enable_auto_output '%s' failed\n", FTP_DECODER_FIELDSTAT_OUTPUT_FILE);
+        fieldstat_easy_free(fenv->fstat.fs4_instance);
+        fenv->fstat.fs4_instance = NULL;
+        return -1;
+    }
+    return 0;
+}
+
+void ftp_decoder_stat_free(struct ftp_decoder_env *fenv)
+{
+    fieldstat_easy_free(fenv->fstat.fs4_instance);
+    fenv->fstat.fs4_instance = NULL;
+}
\ No newline at end of file
diff --git a/src/ftp_decoder_stat.h b/src/ftp_decoder_stat.h
new file mode 100644
index 0000000..bfca60c
--- /dev/null
+++ b/src/ftp_decoder_stat.h
@@ -0,0 +1,28 @@
+#pragma once
+#include <fieldstat/fieldstat_easy.h>
+
+enum ftp_decoder_stat_type{
+    FTPD_STAT_CTRL_LINK_OPEN = 0,
+    FTPD_STAT_CTRL_LINK_CLOSE,
+    FTPD_STAT_DATA_LINK_OPEN,
+    FTPD_STAT_DATA_LINK_CLOSE,
+    FTPD_STAT_NEGOTIATE_DATA_LINK, //Negotiating by PORT or PASV command in ctrl link
+    FTPD_STAT_CTRL_LINK_BYTES_C2S,
+    FTPD_STAT_CTRL_LINK_BYTES_S2C,
+    FTPD_STAT_DATA_LINK_BYTES_C2S,
+    FTPD_STAT_DATA_LINK_BYTES_S2C,    
+    FTPD_STAT_CTRL_CMD,
+
+    FTPD_STAT_MAX,
+};
+
+
+struct ftpd_stat_config_tuple
+{
+    enum ftp_decoder_stat_type type;
+    const char *name;
+};
+
+int ftp_decoder_stat_init(struct ftp_decoder_env *fenv);
+void ftp_decoder_stat_free(struct ftp_decoder_env *fenv);
+void ftp_decoder_stat_incrby(int thread_idx, struct ftp_decoder_env *fenv, enum ftp_decoder_stat_type stattype, long long increment);
\ No newline at end of file
diff --git a/src/ftp_decoder_util.cpp b/src/ftp_decoder_util.cpp
new file mode 100644
index 0000000..8956887
--- /dev/null
+++ b/src/ftp_decoder_util.cpp
@@ -0,0 +1,163 @@
+#include "ftp_decoder_inner.h"
+#include "ftp_decoder.h"
+#include "ftp_decoder_util.h"
+#include <ctype.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <errno.h>
+#include <libgen.h>
+
+void fstring_dup(const fstring *src, fstring *dst)
+{
+    dst->iov_base = (char *)malloc(src->iov_len);
+    assert(dst->iov_base);
+    memcpy((void *)dst->iov_base, src->iov_base, src->iov_len);
+    dst->iov_len = src->iov_len;
+}
+void fstring_safe_dup(const fstring *src, fstring *dst)
+{
+    if (dst->iov_base)
+    {
+        free(dst->iov_base); // free it if exist
+    }
+    fstring_dup(src, dst);
+}
+
+void ftp_parse_result_free(struct ftp_parse_result *parse_result)
+{
+    for (int i = 0; i < FTP_MSG_MAX; i++)
+    {
+        if (i != FTP_INVENTORY && i != FTP_FILE_CONTENT)
+        {
+            if (parse_result->result_array[i].iov_base)
+            {
+                free(parse_result->result_array[i].iov_base);
+            }
+        }
+    }
+}
+
+static void ftp_str_split(const char *payload, size_t len, fstring *left, fstring *rigth, int delim)
+{
+    const char *p = (char *)memchr(payload, delim, len);
+    if (NULL == p)
+    {
+        left->iov_base = (char *)payload;
+        left->iov_len = len;
+        rigth->iov_base = NULL;
+        rigth->iov_len = 0;
+    }
+    else
+    {
+        left->iov_base = (char *)payload;
+        left->iov_len = p - payload;
+        rigth->iov_base = (char *)p + 1;
+        rigth->iov_len = len - left->iov_len - 1;
+    }
+}
+
+long ftp_skip_tail_crlf(const char *payload, long len)
+{
+    long new_len = 0;
+    for (long i = 0; i < len; i++, new_len++)
+    {
+        if ((payload[i] == '\r' || payload[i] == '\n'))
+        {
+            break;
+        }
+    }
+    return new_len;
+}
+
+int ftp_cmd_readline(struct ftp_interact_line *line, const char *payload, size_t len)
+{
+    memset(line, 0, sizeof(struct ftp_interact_line));
+    long skip_crlf_len = ftp_skip_tail_crlf(payload, len);
+    ftp_str_split(payload, skip_crlf_len, &line->cmd_refer, &line->arg_refer, ' ');
+    return 0;
+}
+
+void ftp_strtolower(char *str, size_t len)
+{
+    for (size_t i = 0; i < len; i++)
+    {
+        str[i] = tolower(str[i]);
+    }
+}
+
+int ftp_mkdir_p(const char *path, mode_t mode)
+{
+    struct stat st;
+    errno = 0;
+
+    /* Try to make the directory */
+    if (mkdir(path, mode) == 0)
+        return 0;
+
+    /* If it fails for any reason but EEXIST, fail */
+    if (errno != EEXIST)
+        return -1;
+
+    /* Check if the existing path is a directory */
+    if (stat(path, &st) != 0)
+        return -1;
+
+    /* If not, fail with ENOTDIR */
+    if (!S_ISDIR(st.st_mode))
+    {
+        errno = ENOTDIR;
+        return -1;
+    }
+
+    errno = 0;
+    return 0;
+}
+
+int ftp_session_get_pkt_dir(struct session *sess)
+{
+    const struct packet *pkt = session_get0_current_packet(sess);
+    if (NULL == pkt)
+    {
+        return PACKET_DIRECTION_UNKNOWN;
+    }
+    return packet_get_direction(pkt);
+}
+
+struct ftp_context *ftp_decoder_context_deep_dup(const struct ftp_context *src)
+{
+    struct ftp_context *new_ctx = (struct ftp_context *)calloc(1, sizeof(struct ftp_context));
+    for (int i = 0; i < FTP_MSG_MAX; i++)
+    {
+        if (i != FTP_INVENTORY && i != FTP_FILE_CONTENT)
+        {
+            fstring_safe_dup(&src->parse_result.result_array[i], &new_ctx->parse_result.result_array[i]);
+        }
+    }
+    return new_ctx;
+}
+
+const char *ftp_message_type_to_string(enum ftp_msg_type fmsg_type)
+{
+    switch (fmsg_type)
+    {
+    case FTP_BANNER:
+        return "FTP_BANNER";
+    case FTP_ACCOUNT:
+        return "FTP_ACCOUNT";
+    case FTP_PASSWORD:
+        return "FTP_PASSWORD";
+    case FTP_URI:
+        return "FTP_URI";
+    case FTP_TRANS_MODE:
+        return "FTP_TRANS_MODE";
+    case FTP_TRANS_DIR:
+        return "FTP_TRANS_DIR";
+    case FTP_INVENTORY:
+        return "FTP_INVENTORY";
+    case FTP_FILE_CONTENT:
+        return "FTP_FILE_CONTENT";
+    case FTP_MSG_MAX:
+        return "UNKNOWN";
+    }
+    return "UNKNOWN";
+}
\ No newline at end of file
diff --git a/src/ftp_decoder_util.h b/src/ftp_decoder_util.h
new file mode 100644
index 0000000..4e83128
--- /dev/null
+++ b/src/ftp_decoder_util.h
@@ -0,0 +1,25 @@
+#pragma once
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <string.h>
+#include <bits/types/struct_iovec.h>
+
+#ifndef MAX
+#define MAX(a, b) ((a) >= (b) ? (a) : (b))
+#endif
+#ifndef MIN
+#define MIN(a, b) ((a) >= (b) ? (b) : (a))
+#endif
+
+#ifndef fstring
+typedef struct iovec fstring;
+#endif
+
+void fstring_dup(const fstring *src, fstring *dst);
+void fstring_safe_dup(const fstring *src, fstring *dst);
+void ftp_strtolower(char *str, size_t len);
+int ftp_mkdir_p(const char *path, mode_t mode);
+int ftp_session_get_pkt_dir(struct session *sess);
+struct ftp_context *ftp_decoder_context_deep_dup(const struct ftp_context *src);
+long ftp_skip_tail_crlf(const char *payload, long len);
+void ftp_parse_result_free(struct ftp_parse_result *parse_result);
\ No newline at end of file
diff --git a/src/version.map b/src/version.map
new file mode 100644
index 0000000..404c631
--- /dev/null
+++ b/src/version.map
@@ -0,0 +1,9 @@
+VERS_3.0{
+global: 
+	extern "C" {	
+		FTP_ONLOAD;
+		FTP_UNLOAD;
+		ftp_message_*;
+	};
+  local: *;
+};