ftp decoder rebase develop-2.0dev-ftp-v2.0

12 files changed, 2012 insertions, 0 deletions

diff --git a/decoders/ftp/CMakeLists.txt b/decoders/ftp/CMakeLists.txt
new file mode 100644
index 0000000..578ee6a
--- /dev/null
+++ b/decoders/ftp/CMakeLists.txt
@@ -0,0 +1,10 @@
+add_definitions(-fPIC)
+include_directories(${CMAKE_SOURCE_DIR}/deps)
+
+set(FTP_SRC ftp_module.c ftp_decoder_entry.c ftp_decoder_hash.c ftp_decoder_proto.c ftp_decoder_stat.c ftp_decoder_util.c )
+
+add_library(ftp ${FTP_SRC})
+set_target_properties(ftp PROPERTIES LINK_FLAGS "-Wl,--version-script=${CMAKE_CURRENT_SOURCE_DIR}/version.map")
+target_include_directories(ftp PUBLIC ${CMAKE_SOURCE_DIR}/deps/)
+target_link_libraries(ftp fieldstat4 toml)
+set_target_properties(ftp PROPERTIES PREFIX "") 
diff --git a/decoders/ftp/ftp_decoder_entry.c b/decoders/ftp/ftp_decoder_entry.c
new file mode 100644
index 0000000..dfceadc
--- /dev/null
+++ b/decoders/ftp/ftp_decoder_entry.c
@@ -0,0 +1,243 @@
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+#include "ftp_decoder_inner.h"
+#include "stellar/ftp.h"
+#include "ftp_decoder_util.h"
+#include "stellar/log.h"
+#include <stellar/session.h>
+#include <sys/types.h>
+
+    void ftp_msg_free_cb(void *msg, void *msg_free_arg UNUSED)
+    {
+        struct ftp_message *ftp_msg = (struct ftp_message *)msg;
+        struct ftp_decoder_exdata *ftp_ext = ftp_msg->ftp_ext_ref;
+        if (ftp_msg->topic_type == FTP_TOPIC_CTRL_DTP && ftp_ext->link_type == FTP_LINK_CTRL)
+        {
+            /* multiple dtp messages in ctrl link */
+            ftp_exdata_dtp_free(ftp_msg->dtp_ref);
+        }
+        free(msg);
+    }
+
+    static inline int ftp_is_session_closed(enum session_state state, const char *tcp_payload)
+    {
+        if (unlikely((NULL == tcp_payload) &&
+                     (SESSION_STATE_CLOSING == state || SESSION_STATE_CLOSED == state)))
+        {
+            return 1;
+        }
+        return 0;
+    }
+
+    static struct ftp_decoder_exdata *ftp_exdata_new(struct session *sess, struct ftp_decoder *ftp_env,
+                                                     const char *tcp_payload, uint32_t tcp_payload_len, enum flow_type flow_dir)
+    {
+        uint64_t tcp_seg_num = session_get_stat(sess, flow_dir, STAT_TCP_SEGMENTS_RECEIVED);
+        if (tcp_seg_num > 1)
+        {
+            return NULL;
+        }
+
+        int is_ctrl_link = ftp_ctrl_identify(sess, tcp_payload, tcp_payload_len, flow_dir);
+        if (is_ctrl_link)
+        {
+            struct ftp_decoder_exdata *ctrl_ext = (struct ftp_decoder_exdata *)calloc(1, sizeof(struct ftp_decoder_exdata));
+            ctrl_ext->sess_ref = sess;
+            ctrl_ext->ftp_env_ref = ftp_env;
+            ctrl_ext->link_type = FTP_LINK_CTRL;
+            ctrl_ext->reference = 1;
+            snprintf(ctrl_ext->ctrl_ext.current_working_dir, sizeof(ctrl_ext->ctrl_ext.current_working_dir), "/");
+            int thread_idx = module_manager_get_thread_id(ftp_env->mod_mgr_ref);
+            ftp_decoder_stat_incrby(thread_idx, ftp_env, FTPD_STAT_CTRL_LINK_NEW, 1);
+            STELLAR_LOG_INFO(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp ctrl link new: %s", session_get_readable_addr(sess));
+            session_set_exdata(sess, ftp_env->exdata_id, ctrl_ext);
+            return ctrl_ext;
+        }
+
+        struct ftp_decoder_exdata *data_ext = ftp_data_identify(sess, ftp_env);
+        if (data_ext)
+        {
+            data_ext->link_type = FTP_LINK_DATA;
+            data_ext->sess_ref = sess;
+            data_ext->ftp_env_ref = ftp_env;
+            data_ext->reference++;
+            int thread_idx = module_manager_get_thread_id(ftp_env->mod_mgr_ref);
+            ftp_decoder_stat_incrby(thread_idx, ftp_env, FTPD_STAT_DATA_LINK_NEW, 1);
+            STELLAR_LOG_INFO(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp data link new, session: %s", session_get_readable_addr(sess));
+            session_set_exdata(sess, ftp_env->exdata_id, data_ext);
+            return data_ext;
+        }
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "not ftp session: %s", session_get_readable_addr(sess));
+        return NULL;
+    }
+
+    void ftp_decoder_do_exdata_free(struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env UNUSED)
+    {
+        if (NULL == ftp_ext)
+        {
+            return;
+        }
+        ftp_ext->reference--;
+        if (ftp_ext->reference > 0)
+        {
+            return;
+        }
+        ftp_exdata_dtp_free(ftp_ext->dtp);
+        ftp_login_info_free(&ftp_ext->ftp_login_pri);
+        free(ftp_ext);
+    }
+
+    void ftp_decoder_exdata_free_cb(int idx UNUSED, void *ex_ptr, void *arg)
+    {
+        if (NULL == ex_ptr)
+        {
+            return;
+        }
+        struct ftp_decoder *ftp_env = (struct ftp_decoder *)arg;
+        struct ftp_decoder_exdata *ftp_ext = (struct ftp_decoder_exdata *)ex_ptr;
+        int thread_idx = module_manager_get_thread_id(ftp_env->mod_mgr_ref);
+        if (ftp_ext->link_type == FTP_LINK_CTRL)
+        {
+            ftp_decoder_stat_incrby(thread_idx, (struct ftp_decoder *)ftp_env, FTPD_STAT_CTRL_LINK_FREE, 1);
+            STELLAR_LOG_INFO(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp ctrl link free: %s", session_get_readable_addr(ftp_ext->sess_ref));
+        }
+        else
+        {
+            ftp_decoder_stat_incrby(thread_idx, (struct ftp_decoder *)ftp_env, FTPD_STAT_DATA_LINK_FREE, 1);
+            STELLAR_LOG_INFO(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp data link free: %s", session_get_readable_addr(ftp_ext->sess_ref));
+            /* data link exdata freed in ftp_del_hash_item()
+             * maybe:
+             * 1. data link is not established success, ftp_del_hash_item() will free exdata
+             * 2. data link is established success, but ctrl link stor/retr command is not completed, so need keep the exdata in hash item, dtp information could be updated later
+             */
+        }
+        ftp_decoder_do_exdata_free(ftp_ext, ftp_env);
+    }
+
+    static void ftp_ctrl_link_entry(int thread_idx, struct session *sess, struct ftp_decoder_exdata *ftp_ext,
+                                    struct ftp_decoder *ftp_env, const char *tcp_payload, uint32_t tcp_payload_len, enum flow_type flow_dir)
+    {
+        if (unlikely(NULL == tcp_payload))
+        {
+            return;
+        }
+        if (ftp_cmd_readline(&ftp_ext->ctrl_ext.cmd_line, tcp_payload, tcp_payload_len) < 0)
+        {
+            STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp session:%s, command line parse error! %.*s",
+                              session_get_readable_addr(sess), (int)tcp_payload_len >= 32 ? 32 : tcp_payload_len, tcp_payload);
+            ftp_ext->ignore_session = 1;
+            return;
+        }
+
+        int thread_id = module_manager_get_thread_id(ftp_env->mod_mgr_ref);
+        ftp_decoder_stat_incrby(thread_id, ftp_env, FTPD_STAT_CTRL_CMD, 1);
+        if (FLOW_TYPE_C2S == flow_dir)
+        {
+            if (ftp_command_process(sess, ftp_ext, ftp_env) < 0)
+            {
+                STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp session:%s,command process error! %.*s",
+                                  session_get_readable_addr(sess), (int)tcp_payload_len >= 32 ? 32 : tcp_payload_len, tcp_payload);
+                ftp_ext->ignore_session = 1;
+                return;
+            }
+            ftp_decoder_push_msg(FTP_TOPIC_CTRL_REQ_LINE, sess, ftp_ext, ftp_env);
+            ftp_decoder_stat_incrby(thread_idx, ftp_env, FTPD_STAT_CTRL_LINK_BYTES_C2S, (long long)tcp_payload_len);
+        }
+        else
+        {
+            if (ftp_reply_process(sess, ftp_ext, ftp_env) < 0)
+            {
+                STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp session:%s, reply process error! %.*s",
+                                  session_get_readable_addr(sess), (int)tcp_payload_len >= 32 ? 32 : tcp_payload_len, tcp_payload);
+                ftp_ext->ignore_session = 1;
+                return;
+            }
+            ftp_decoder_push_msg(FTP_TOPIC_CTRL_RES_LINE, sess, ftp_ext, ftp_env);
+            ftp_decoder_stat_incrby(thread_idx, ftp_env, FTPD_STAT_CTRL_LINK_BYTES_S2C, (long long)tcp_payload_len);
+        }
+        if (ftp_ext->dtp != NULL)
+        {
+            ftp_decoder_push_msg(FTP_TOPIC_CTRL_DTP, sess, ftp_ext, ftp_env);
+            ftp_ext->dtp = NULL; // ownership move to message after push, update in next command
+        }
+        return;
+    }
+
+    static void ftp_data_link_entry(int thread_idx, struct session *sess, struct ftp_decoder_exdata *ftp_ext,
+                                    struct ftp_decoder *ftp_env, const char *tcp_payload UNUSED, uint32_t tcp_payload_len, enum flow_type flow_dir)
+    {
+        ftp_ext->data_ext.chunk = tcp_payload;
+        ftp_ext->data_ext.chunk_size = tcp_payload_len;
+        ftp_ext->data_ext.offset += tcp_payload_len;
+        if (ftp_is_session_closed(session_get_current_state(sess), tcp_payload))
+        {
+            ftp_ext->data_ext.is_finished = 1;
+            ftp_decoder_push_msg(FTP_TOPIC_DATA_DTP, sess, ftp_ext, ftp_env);
+            STELLAR_LOG_INFO(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp data link closed, %s", session_get_readable_addr(sess));
+            return;
+        }
+
+        ftp_decoder_push_msg(FTP_TOPIC_DATA_DTP, sess, ftp_ext, ftp_env);
+        if (flow_dir == FLOW_TYPE_C2S)
+        {
+            ftp_decoder_stat_incrby(thread_idx, ftp_env, FTPD_STAT_DATA_LINK_BYTES_C2S, (long long)tcp_payload_len);
+        }
+        else
+        {
+            ftp_decoder_stat_incrby(thread_idx, ftp_env, FTPD_STAT_DATA_LINK_BYTES_S2C, (long long)tcp_payload_len);
+        }
+    }
+
+    void ftp_on_tcp_stream_cb(struct session *sess, enum session_state state UNUSED, const char *tcp_payload, uint32_t tcp_payload_len, void *args)
+    {
+        enum flow_type flow_dir = session_get_flow_type(sess);
+        struct ftp_decoder *ftp_env = (struct ftp_decoder *)args;
+        struct ftp_decoder_exdata *ftp_ext = (struct ftp_decoder_exdata *)session_get_exdata(sess, ftp_env->exdata_id);
+        if (NULL == ftp_ext)
+        {
+            if (FLOW_TYPE_C2S != flow_dir && FLOW_TYPE_S2C != flow_dir) // maybe timeout in polling event
+            {
+                return;
+            }
+            ftp_ext = ftp_exdata_new(sess, ftp_env, tcp_payload, tcp_payload_len, flow_dir);
+        }
+        if (NULL == ftp_ext || ftp_ext->ignore_session) // not ftp session
+        {
+            return;
+        }
+        int thread_idx = module_manager_get_thread_id(ftp_env->mod_mgr_ref);
+        if (FTP_LINK_CTRL == ftp_ext->link_type)
+        {
+            ftp_ctrl_link_entry(thread_idx, sess, ftp_ext, ftp_env, tcp_payload, tcp_payload_len, flow_dir);
+        }
+        else
+        {
+            ftp_data_link_entry(thread_idx, sess, ftp_ext, ftp_env, tcp_payload, tcp_payload_len, flow_dir);
+        }
+    }
+
+    int ftp_decoder_push_msg(enum ftp_topic_type topic_type, struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        struct ftp_message *fmsg = (struct ftp_message *)calloc(1, sizeof(struct ftp_message));
+        fmsg->sess_ref = sess;
+        fmsg->topic_type = topic_type;
+        fmsg->ftp_ext_ref = ftp_ext;
+        fmsg->ftp_env_ref = ftp_env;
+        fmsg->dtp_ref = ftp_ext->dtp;
+
+        struct mq_runtime *mq_rt = module_manager_get_mq_runtime(ftp_env->mod_mgr_ref);
+        int ret = mq_runtime_publish_message(mq_rt, ftp_env->ftp_topic_mgr->topic_compose[topic_type].topic_id, fmsg);
+        if (ret < 0)
+        {
+            STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp publist message failed, topic:%s, session:%s",
+                              ftp_env->ftp_topic_mgr->topic_compose[topic_type].topic_name, session_get_readable_addr(sess));
+            free(fmsg);
+        }
+        return ret;
+    }
+
+#ifdef __cplusplus
+}
+#endif
diff --git a/decoders/ftp/ftp_decoder_hash.c b/decoders/ftp/ftp_decoder_hash.c
new file mode 100644
index 0000000..3a03368
--- /dev/null
+++ b/decoders/ftp/ftp_decoder_hash.c
@@ -0,0 +1,138 @@
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#include "ftp_decoder_inner.h"
+#include "ftp_decoder_hash.h"
+#include "ftp_decoder_util.h"
+#include <arpa/inet.h>
+#include <netinet/tcp.h>
+#include <netinet/ip.h>
+#include <netinet/ip6.h>
+#include "uthash/uthash.h"
+#include "uthash/utlist.h"
+
+    static __thread struct ftp_datalink_htable *__thread_local_ftp_datalink_htable = NULL;
+    static __thread struct ftp_datalink_htable *__thread_local_ftp_datalink_htable_fifo_head = NULL;
+    static __thread char __ftp_hash_string_buf[FTP_HASH_STRING_BUF_SIZE];
+    static __thread int ftp_local_thread_idx;
+    static __thread struct ftp_decoder *ftp_local_env;
+
+    static void ftp_del_hash_item(struct ftp_datalink_htable *item)
+    {
+        ftp_decoder_stat_incrby(ftp_local_thread_idx, ftp_local_env, FTPD_STAT_DATA_LINK_HTABLE_ITEMS, -1);
+        HASH_DEL(__thread_local_ftp_datalink_htable, item);
+        DL_DELETE(__thread_local_ftp_datalink_htable_fifo_head, item);
+        ftp_decoder_do_exdata_free(item->ftp_ext, ftp_local_env);
+        free(item); // only free hash item, but not free ftp_ext
+    }
+
+    static void ftp_hash_cleanup_timeout(void)
+    {
+        struct ftp_datalink_htable *to_del, *tmp;
+        time_t now = time(NULL);
+        DL_FOREACH_SAFE(__thread_local_ftp_datalink_htable_fifo_head, to_del, tmp)
+        {
+            if (now - to_del->insert_htable_time > FTP_HASH_ITEM_TIMEOUT)
+            {
+                ftp_del_hash_item(to_del);
+            }
+        }
+    }
+
+    int ftp_hash_add(const ftp_hash_key_t *key, u_int32_t key_len, struct ftp_datalink_htable *new_item)
+    {
+        ftp_hash_cleanup_timeout();
+        struct ftp_datalink_htable *in_hash_item = NULL;
+        HASH_FIND(hh, __thread_local_ftp_datalink_htable, key, key_len, in_hash_item);
+        if (in_hash_item != NULL)
+        {
+            return -1; // duplicate
+        }
+        HASH_ADD(hh, __thread_local_ftp_datalink_htable, hkey, key_len, new_item);
+        DL_APPEND(__thread_local_ftp_datalink_htable_fifo_head, new_item);
+        return 0;
+    }
+
+    void ftp_hash_del(const ftp_hash_key_t *key)
+    {
+        struct ftp_datalink_htable *tmp = NULL;
+        HASH_FIND(hh, __thread_local_ftp_datalink_htable, key, sizeof(ftp_hash_key_t), tmp);
+        if (tmp)
+        {
+            ftp_del_hash_item(tmp);
+        }
+    }
+
+    void ftp_make_hkey_v4(ftp_hash_key_t *keyv4, uint32_t sip_net, uint32_t dip_net, uint16_t dip_port_net)
+    {
+        memset(keyv4, 0, sizeof(ftp_hash_key_t));
+        keyv4->af_inet = AF_INET;
+        keyv4->saddr4 = sip_net;
+        keyv4->daddr4 = dip_net;
+        keyv4->sport = 0;
+        keyv4->dport = dip_port_net;
+    }
+
+    void ftp_make_hkey_v6(ftp_hash_key_t *keyv6, const struct in6_addr *sip, const struct in6_addr *dip, uint16_t dip_port_net)
+    {
+        memset(keyv6, 0, sizeof(ftp_hash_key_t));
+        keyv6->af_inet = AF_INET6;
+        memcpy(&keyv6->saddr6, sip, sizeof(struct in6_addr));
+        memcpy(&keyv6->daddr6, dip, sizeof(struct in6_addr));
+        keyv6->sport = 0;
+        keyv6->dport = dip_port_net;
+    }
+
+    struct ftp_datalink_htable *ftp_hash_search(const ftp_hash_key_t *key)
+    {
+        ftp_hash_cleanup_timeout();
+        struct ftp_datalink_htable *tmp = NULL;
+        HASH_FIND(hh, __thread_local_ftp_datalink_htable, key, sizeof(ftp_hash_key_t), tmp);
+        return tmp;
+    }
+
+    const char *ftp_hash_key_to_str(const ftp_hash_key_t *hkey)
+    {
+        char sip_str[INET6_ADDRSTRLEN], dip_str[INET6_ADDRSTRLEN];
+        unsigned short sport_host, dport_host;
+        if (AF_INET == hkey->af_inet)
+        {
+            inet_ntop(AF_INET, &hkey->saddr4, sip_str, INET_ADDRSTRLEN);
+            inet_ntop(AF_INET, &hkey->daddr4, dip_str, INET_ADDRSTRLEN);
+            sport_host = ntohs(hkey->sport);
+            dport_host = ntohs(hkey->dport);
+        }
+        else
+        {
+            inet_ntop(AF_INET6, &hkey->saddr6, sip_str, INET6_ADDRSTRLEN);
+            inet_ntop(AF_INET6, &hkey->daddr6, dip_str, INET6_ADDRSTRLEN);
+            sport_host = ntohs(hkey->sport);
+            dport_host = ntohs(hkey->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;
+    }
+
+    struct module *ftp_on_thread_init(UNUSED struct module_manager *mod_mgr, UNUSED int thread_id, struct module *mod)
+    {
+        ftp_local_thread_idx = thread_id;
+        __thread_local_ftp_datalink_htable = NULL;
+        ftp_local_env = (struct ftp_decoder *)module_get_ctx(mod);
+        return mod;
+    }
+
+    void ftp_on_thread_exit(UNUSED struct module_manager *mod_mgr, UNUSED int thread_id, UNUSED struct module *mod)
+    {
+        struct ftp_datalink_htable *item, *tmp;
+        HASH_ITER(hh, __thread_local_ftp_datalink_htable, item, tmp)
+        {
+            ftp_del_hash_item(item);
+        }
+        HASH_CLEAR(hh, __thread_local_ftp_datalink_htable);
+    }
+
+#ifdef __cplusplus
+}
+#endif
diff --git a/decoders/ftp/ftp_decoder_hash.h b/decoders/ftp/ftp_decoder_hash.h
new file mode 100644
index 0000000..f8f5468
--- /dev/null
+++ b/decoders/ftp/ftp_decoder_hash.h
@@ -0,0 +1,54 @@
+#pragma once
+#include <stddef.h>
+#include <arpa/inet.h>
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+#include "stellar/session.h"
+#include "stellar/log.h"
+#include "stellar/module.h"
+#include "stellar/mq.h"
+#include "uthash/uthash.h"
+#ifdef __cplusplus
+}
+#endif
+
+#define FTP_HASH_STRING_BUF_SIZE (256)
+#define FTP_HASH_ITEM_TIMEOUT (30) // seconds
+
+struct ftp_session_addr /* network order */
+{
+    int af_inet; // AF_INET or AF_INET6
+    uint16_t sport;
+    uint16_t dport;
+    union
+    {
+        uint32_t saddr4;
+        struct in6_addr saddr6;
+    };
+    union
+    {
+        uint32_t daddr4;
+        struct in6_addr daddr6;
+    };
+} __attribute__((packed));
+typedef struct ftp_session_addr ftp_hash_key_t;
+
+struct ftp_datalink_htable
+{
+    UT_hash_handle hh;
+    struct ftp_datalink_htable *next, *prev; // for timeout fifo list
+    ftp_hash_key_t hkey;
+    time_t insert_htable_time;
+    struct ftp_decoder_exdata *ftp_ext; // data link exdata
+};
+
+int ftp_hash_add(const ftp_hash_key_t *key, u_int32_t key_len, struct ftp_datalink_htable *new_item);
+struct ftp_datalink_htable *ftp_hash_search(const ftp_hash_key_t *key);
+void ftp_hash_del(const ftp_hash_key_t *key);
+void ftp_session_get_addr(const struct session *sess, struct ftp_session_addr *addr);
+void ftp_make_hkey_v4(ftp_hash_key_t *keyv4, uint32_t sip_net, uint32_t dip_net, uint16_t dip_port_net);
+void ftp_make_hkey_v6(ftp_hash_key_t *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_session_addr *hkey);
+void ftp_session_get_addr(const struct session *sess, struct ftp_session_addr *addr);
\ No newline at end of file
diff --git a/decoders/ftp/ftp_decoder_inner.h b/decoders/ftp/ftp_decoder_inner.h
new file mode 100644
index 0000000..4ccaf46
--- /dev/null
+++ b/decoders/ftp/ftp_decoder_inner.h
@@ -0,0 +1,177 @@
+#pragma once
+#include <linux/limits.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <assert.h>
+#include <string.h>
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+#include "stellar/ftp.h"
+#include "stellar/log.h"
+#include "stellar/module.h"
+#include "stellar/mq.h"
+#include "stellar/session.h"
+#include "stellar/utils.h"
+
+#ifdef __cplusplus
+}
+#endif
+#include "ftp_decoder_stat.h"
+#include "ftp_decoder_hash.h"
+
+#define FTP_IDENTIRY_MIN_LEN 4
+#define FTP_IDENTIRY_MAX_LEN 32
+#define FTP_CMD_MAX_LENGTH 256
+#define FTP_URL_MAX_LEN 2048
+
+#define FTP_DECODER_FIELDSTAT_NAME "ftp_decoder"
+#define FTP_DECODER_DXDATA_NAME "ftp_decoder_exdata"
+
+#define FTP_DECODER_FIELDSTAT_OUTPUT_FILE "./log/ftp_decoder.fs4"
+#define FTP_DECODER_FIELDSTAT_OUTPUT_INTERVAL 1
+
+#ifndef UNUSED
+#define UNUSED __attribute__((unused))
+#endif
+
+#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_line;  // full line but no "\r\n", pointer to packet payload
+    fstring cmd_refer; // pointer to packet payload first word
+    fstring arg_refer; // pointer to packet payload after first word
+};
+
+struct ftp_login_internal
+{
+    fstring username; // iov_base is C string with '\0'
+    fstring password; // iov_base is C string with '\0'
+};
+
+struct ftp_decoder_ctrl_exdata
+{
+    char current_working_dir[PATH_MAX]; // default is "/"
+    enum ftp_command cmd_type;
+    enum ftp_reply_code reply_code;
+    struct ftp_interact_line cmd_line; // per tcp segment
+    ftp_hash_key_t last_data_link_key; // by cmd port or pasv
+    enum ftp_transfer_mode mode;
+    enum ftp_transfer_dir dir;
+};
+
+struct ftp_decoder_data_exdata
+{
+    const char *chunk; // refer to tcp payload
+    size_t chunk_size;
+    size_t offset;
+    int is_finished;
+};
+
+struct ftp_decoder_exdata // per session
+{
+    struct session *sess_ref;
+    struct ftp_decoder *ftp_env_ref;
+    enum ftp_link_type link_type;
+    int reference;
+    int ignore_session;
+    struct ftp_login_internal ftp_login_pri; // per session
+    struct ftp_dtp *dtp;                     // per transaction. data link dtp is deep copy from ctrl link, need be free when session close
+    struct ftp_decoder_ctrl_exdata ctrl_ext;
+    struct ftp_decoder_data_exdata data_ext;
+};
+
+struct ftp_interact_cmd_parser
+{
+    enum ftp_command cmd_type;
+    const char *cmd_name;
+    size_t cmd_len;
+    int (*cmd_handler)(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env);
+};
+
+struct ftp_interact_reply_parser
+{
+    enum ftp_reply_code reply_code_type;
+    const char *reply_code;
+    size_t reply_len;
+    int (*reply_handler)(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env);
+};
+
+enum ftp_topic_type
+{
+    FTP_TOPIC_CTRL_REQ_LINE = 0,
+    FTP_TOPIC_CTRL_RES_LINE,
+    FTP_TOPIC_CTRL_DTP,
+    FTP_TOPIC_DATA_DTP,
+    FTP_TOPIC_MAX,
+};
+
+struct ftp_topic_compose
+{
+    enum ftp_topic_type topic_type;
+    int topic_id;
+    const char *topic_name;
+};
+
+struct ftp_topic_manager
+{
+    struct ftp_topic_compose topic_compose[FTP_TOPIC_MAX];
+};
+
+struct ftp_decoder
+{
+    struct module_manager *mod_mgr_ref;
+    struct logger *logger_ref;
+    struct ftp_topic_manager *ftp_topic_mgr;
+    int exdata_id;
+    struct ftp_decoder_stat stat;
+};
+
+struct ftp_message
+{
+    struct session *sess_ref;
+    enum ftp_topic_type topic_type;
+    struct ftp_decoder_exdata *ftp_ext_ref;
+    struct ftp_decoder *ftp_env_ref;
+    struct ftp_dtp *dtp_ref;
+};
+
+void ftp_msg_free_cb(void *msg, void *msg_free_arg);
+void ftp_decoder_exdata_free_cb(int idx, void *ex_ptr, void *arg);
+void ftp_decoder_do_exdata_free(struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env);
+void ftp_on_tcp_stream_cb(struct session *sess, enum session_state state, const char *tcp_payload, uint32_t tcp_payload_len, void *args);
+int ftp_command_process(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env);
+int ftp_reply_process(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env);
+
+void ftp_exdata_dtp_free(struct ftp_dtp *dtp);
+void ftp_login_info_free(struct ftp_login_internal *login_info);
+int ftp_hash_table_create(struct ftp_decoder *fenv);
+void ftp_hash_table_destroy(struct ftp_decoder *fenv);
+int ftp_ctrl_identify_by_payload(const char *payload, size_t len, enum flow_type curdir);
+int ftp_ctrl_identify_by_addr(struct session *sess);
+int ftp_ctrl_identify(struct session *sess, const char *payload, size_t len, enum flow_type curdir);
+struct ftp_decoder_exdata *ftp_data_identify(struct session *sess, struct ftp_decoder *fenv);
+int ftp_cmd_readline(struct ftp_interact_line *line, const char *payload, size_t len);
+void ftp_ctrl_entry(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *fenv);
+void ftp_data_entry(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *fenv);
+int ftp_decoder_push_msg(enum ftp_topic_type topic_type, struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *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, struct ftp_decoder *ftp_env);
\ No newline at end of file
diff --git a/decoders/ftp/ftp_decoder_proto.c b/decoders/ftp/ftp_decoder_proto.c
new file mode 100644
index 0000000..064d176
--- /dev/null
+++ b/decoders/ftp/ftp_decoder_proto.c
@@ -0,0 +1,653 @@
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#include "ftp_decoder_inner.h"
+#include "ftp_decoder_util.h"
+#include "ftp_decoder_hash.h"
+#include <sys/types.h>
+#include <arpa/inet.h>
+#include <time.h>
+    /*
+        https://datatracker.ietf.org/doc/html/rfc959
+                                                -------------
+                                                |/---------\|
+                                                ||   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 int ftp_create_new_datalink_hash_item(struct ftp_decoder *ftp_env, struct ftp_decoder_exdata *ftp_ext)
+    {
+        int thread_id = module_manager_get_thread_id(ftp_env->mod_mgr_ref);
+        ftp_decoder_stat_incrby(thread_id, ftp_env, FTPD_STAT_DATA_LINK_HTABLE_ITEMS, 1);
+
+        struct ftp_datalink_htable *hitem = (struct ftp_datalink_htable *)calloc(1, sizeof(struct ftp_datalink_htable));
+        hitem->ftp_ext = ftp_exdata_deep_clone(ftp_ext);
+        hitem->ftp_ext->link_type = FTP_LINK_DATA;
+        hitem->insert_htable_time = time(NULL);
+        memcpy(&hitem->hkey, &ftp_ext->ctrl_ext.last_data_link_key, sizeof(ftp_hash_key_t));
+        memset(&ftp_ext->data_ext, 0, sizeof(struct ftp_decoder_data_exdata));
+        int ret = ftp_hash_add(&ftp_ext->ctrl_ext.last_data_link_key, sizeof(ftp_hash_key_t), hitem);
+        if (ret < 0)
+        {
+            STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp data link hash item add failed, session:%s, hashkey: %s",
+                              session_get_readable_addr(ftp_ext->sess_ref), ftp_hash_key_to_str(&ftp_ext->ctrl_ext.last_data_link_key));
+            ftp_decoder_do_exdata_free(hitem->ftp_ext, ftp_env);
+            free(hitem);
+            return -1;
+        }
+        return 0;
+    }
+
+    /* According to the CWD, server address, file name of the current command, concatenated into a complete URL */
+    static void ftp_assemble_uri(struct session *sess, struct ftp_decoder_exdata *ftp_ext)
+    {
+        char dip_str[INET6_ADDRSTRLEN] = {};
+        int url_len = 0;
+        struct ftp_session_addr ses_addr = {};
+        ftp_session_get_addr(sess, &ses_addr);
+        uint16_t server_port_host = ntohs(ses_addr.dport);
+        if (ses_addr.af_inet == AF_INET)
+        {
+            inet_ntop(AF_INET, &ses_addr.daddr4, dip_str, INET_ADDRSTRLEN);
+        }
+        else
+        {
+            inet_ntop(AF_INET6, &ses_addr.daddr6, dip_str, INET6_ADDRSTRLEN);
+        }
+
+        char file_name[PATH_MAX] = {};
+        char absolute_file_path[PATH_MAX] = {};
+        char tmp_url[PATH_MAX] = {};
+        snprintf(file_name, sizeof(file_name), "%.*s", IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.arg_refer));
+        ftp_join_absolute_path(ftp_ext->ctrl_ext.current_working_dir, (const char *)file_name, absolute_file_path, PATH_MAX);
+        if (server_port_host != 21)
+        {
+            url_len = snprintf(tmp_url, sizeof(tmp_url), "ftp://%s:%u%s", dip_str, server_port_host, absolute_file_path);
+        }
+        else
+        {
+            url_len = snprintf(tmp_url, sizeof(tmp_url), "ftp://%s%s", dip_str, absolute_file_path);
+        }
+
+        struct ftp_dtp *dtp = ftp_ext->dtp;
+        if (dtp->uri != NULL)
+        {
+            free((void *)dtp->uri); // update every time for retr/stor commands
+        }
+        dtp->uri = (char *)calloc(1, url_len + 1);
+        memcpy((char *)dtp->uri, tmp_url, url_len);
+        return;
+    }
+
+    int ftp_cmd_handler_non_implemented(struct session *sess UNUSED, struct ftp_decoder_exdata *ftp_ext UNUSED, struct ftp_decoder *ftp_env UNUSED)
+    {
+        return 0;
+    }
+
+    int ftp_cmd_handler_do_user(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        fstring_safe_dup(&ftp_ext->ftp_login_pri.username, &ftp_ext->ctrl_ext.cmd_line.arg_refer);
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "session: %s, USER: %.*s",
+                          session_get_readable_addr(sess), IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.arg_refer));
+        return 0;
+    }
+
+    int ftp_cmd_handler_do_pass(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        fstring_safe_dup(&ftp_ext->ftp_login_pri.password, &ftp_ext->ctrl_ext.cmd_line.arg_refer);
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "session: %s, PASS: %.*s",
+                          session_get_readable_addr(sess), IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.arg_refer));
+        return 0;
+    }
+
+    int ftp_cmd_handler_do_cwd(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        char tmp_path[PATH_MAX] = {};
+        snprintf(tmp_path, sizeof(tmp_path), "%.*s", IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.arg_refer));
+        ftp_join_absolute_path(ftp_ext->ctrl_ext.current_working_dir, tmp_path, ftp_ext->ctrl_ext.current_working_dir, sizeof(ftp_ext->ctrl_ext.current_working_dir));
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "session: %s, CWD: %.*s, absolute pwd: %s",
+                          session_get_readable_addr(sess), IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.arg_refer), ftp_ext->ctrl_ext.current_working_dir);
+        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[FTP_CMD_MAX_LENGTH + 1] = {};
+        size_t min_size = MIN(FTP_CMD_MAX_LENGTH, cmd_str->iov_len + 1);
+        snprintf(raw_cmd_str_tmp, min_size, "%.*s", (int)min_size, (char *)cmd_str->iov_base);
+        char only_integer_str_tmp[FTP_CMD_MAX_LENGTH + 1] = {};
+        if (sscanf(raw_cmd_str_tmp, "%*[a-zA-Z (]%255[0-9,]", only_integer_str_tmp) <= 0)
+        {
+            if (sscanf(raw_cmd_str_tmp, "%255[1234567890,]", only_integer_str_tmp) <= 0)
+            {
+                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)
+        {
+            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;
+    }
+
+    void ftp_login_info_free(struct ftp_login_internal *login_info)
+    {
+        if (login_info->username.iov_base)
+        {
+            free((void *)login_info->username.iov_base);
+        }
+        if (login_info->password.iov_base)
+        {
+            free((void *)login_info->password.iov_base);
+        }
+    }
+
+    void ftp_exdata_dtp_free(struct ftp_dtp *dtp)
+    {
+        if (NULL == dtp)
+        {
+            return;
+        }
+        if (dtp->uri)
+        {
+            free((void *)dtp->uri);
+        }
+        free(dtp);
+    }
+
+    static void ftp_exdata_dtp_renew(struct ftp_decoder_exdata *ftp_ext)
+    {
+        if (ftp_ext->dtp)
+        {
+            ftp_exdata_dtp_free(ftp_ext->dtp);
+        }
+        ftp_ext->dtp = (struct ftp_dtp *)calloc(1, sizeof(struct ftp_dtp));
+        ftp_ext->dtp->mode = ftp_ext->ctrl_ext.mode;
+        ftp_ext->dtp->dir = ftp_ext->ctrl_ext.dir;
+        ftp_ext->dtp->cmd = ftp_ext->ctrl_ext.cmd_type;
+        ftp_ext->dtp->uri = NULL;
+        return;
+    }
+
+    /* PORT command only support IPv4, arg pattern: h1,h2,h3,h4,p1,p2 */
+    int ftp_cmd_handler_do_port(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        ftp_ext->ctrl_ext.mode = FTP_TRANSFER_PORT;
+        unsigned int dst_ip_net;
+        unsigned short dst_port_net;
+        if (ftp_parse_ipv4_port_style(&ftp_ext->ctrl_ext.cmd_line.arg_refer, &dst_ip_net, &dst_port_net) < 0)
+        {
+            STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "do port cmd, parse error: %.*s", IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.cmd_line));
+            return -1;
+        }
+        struct ftp_session_addr ses_addr = {};
+        ftp_session_get_addr(sess, &ses_addr);
+        // in active mode, new data link src ip address is server ip address
+        ftp_make_hkey_v4(&ftp_ext->ctrl_ext.last_data_link_key, ses_addr.daddr4, dst_ip_net, dst_port_net);
+
+        ftp_create_new_datalink_hash_item(ftp_env, ftp_ext);
+        int thread_idx = module_manager_get_thread_id(ftp_env->mod_mgr_ref);
+        ftp_decoder_stat_incrby(thread_idx, ftp_env, FTPD_STAT_NEGOTIATE_DATA_LINK, 1);
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "do port cmd: %.*s, parsed dip:%x, dport:%u, hashkey:%s",
+                          IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.arg_refer), dst_ip_net, dst_port_net,
+                          ftp_hash_key_to_str(&ftp_ext->ctrl_ext.last_data_link_key));
+        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_decoder_exdata *ftp_ext UNUSED, struct ftp_decoder *ftp_env)
+    {
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "'%s' PASV command", session_get_readable_addr(sess));
+        return 0;
+    }
+
+    static void ftp_update_datalink_dtp(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        struct ftp_datalink_htable *hitem = ftp_hash_search(&ftp_ext->ctrl_ext.last_data_link_key);
+        if (NULL == hitem)
+        {
+            STELLAR_LOG_WARN(ftp_env->logger_ref, FTP_MODULE_NAME, "data link not found, session:%s, key:%s",
+                             session_get_readable_addr(sess), ftp_hash_key_to_str(&ftp_ext->ctrl_ext.last_data_link_key));
+            return;
+        }
+        hitem->ftp_ext->link_type = FTP_LINK_DATA;
+        hitem->ftp_ext->dtp = ftp_dtp_deep_clone(ftp_ext->dtp);
+    }
+
+    int ftp_cmd_handler_do_list(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        ftp_exdata_dtp_renew(ftp_ext);
+        // no uri for list command
+        ftp_update_datalink_dtp(sess, ftp_ext, ftp_env);
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "'%s' LIST command", session_get_readable_addr(sess));
+        return 0;
+    }
+
+    int ftp_cmd_handler_do_stor(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        ftp_ext->ctrl_ext.dir = FTP_TRANSFER_STOR;
+        ftp_exdata_dtp_renew(ftp_ext);
+        ftp_assemble_uri(sess, ftp_ext);
+        ftp_update_datalink_dtp(sess, ftp_ext, ftp_env);
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "'%s', STOR uri:%.*s",
+                          session_get_readable_addr(sess), IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.arg_refer));
+        return 0;
+    }
+
+    int ftp_cmd_handler_do_retr(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        ftp_ext->ctrl_ext.dir = FTP_TRANSFER_RETR;
+        ftp_exdata_dtp_renew(ftp_ext);
+        ftp_assemble_uri(sess, ftp_ext);
+        ftp_update_datalink_dtp(sess, ftp_ext, ftp_env);
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "'%s', RETR uri:%.*s",
+                          session_get_readable_addr(sess), IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.arg_refer));
+        return 0;
+    }
+
+    int ftp_parse_eprt_ipport_style(const fstring *arg_str, struct in6_addr *ipd_addr, unsigned short *port_net, struct ftp_decoder *ftp_env)
+    {
+        unsigned int port_host;
+        int inet_proto;
+        char ip6_addr_str[INET6_ADDRSTRLEN] = {};
+        char raw_cmd_str_tmp[arg_str->iov_len + 1];
+        memset(raw_cmd_str_tmp, 0, sizeof(raw_cmd_str_tmp));
+        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)
+        {
+            STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp EPRT command parse error: %.*s", IOVEC_PRINT_PTR(arg_str));
+            return -1;
+        }
+        inet_proto = atoi(ptr);
+        if (2 != inet_proto)
+        {
+            STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp EPRT command 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)
+        {
+            STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp EPRT command 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)
+        {
+            STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp EPRT command 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_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        ftp_ext->ctrl_ext.mode = FTP_TRANSFER_PORT;
+        struct in6_addr ipd_addr;
+        unsigned short port_net;
+        // fstring_safe_dup(&ftp_ext->cmd_result.cmd_refer, &ftp_ext->parse_result.result_array[FTP_TRANS_MODE]);
+        if (ftp_parse_eprt_ipport_style(&ftp_ext->ctrl_ext.cmd_line.arg_refer, &ipd_addr, &port_net, ftp_env) < 0)
+        {
+            return -1;
+        }
+
+        struct ftp_session_addr ses_addr = {};
+        ftp_session_get_addr(sess, &ses_addr);
+        assert(ses_addr.af_inet == AF_INET6);
+        ftp_make_hkey_v6(&ftp_ext->ctrl_ext.last_data_link_key, (struct in6_addr *)&ses_addr.daddr6, &ipd_addr, port_net);
+
+        ftp_create_new_datalink_hash_item(ftp_env, ftp_ext);
+        int thread_id = module_manager_get_thread_id(ftp_env->mod_mgr_ref);
+        ftp_decoder_stat_incrby(thread_id, ftp_env, FTPD_STAT_NEGOTIATE_DATA_LINK, 1);
+
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "'%s': EPRT command, port:%u", session_get_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_decoder_exdata *ftp_ext UNUSED, struct ftp_decoder *ftp_env)
+    {
+        // todo
+        STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, " '%s': LPRT command not support yet!", session_get_readable_addr(sess));
+        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_decoder_exdata *ftp_ext UNUSED, struct ftp_decoder *ftp_env)
+    {
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "'%s' EPSV command", session_get_readable_addr(sess));
+        return 0;
+    }
+
+    /* 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_decoder_exdata *ftp_ext UNUSED, struct ftp_decoder *ftp_env)
+    {
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "'%s' LPSV command", session_get_readable_addr(sess));
+        return 0;
+    }
+
+    static const struct ftp_interact_cmd_parser g_ftp_c2s_cmd_tuple[] =
+        {
+            {FTP_COMMAND_SYST, "SYST", 4, ftp_cmd_handler_non_implemented},
+            {FTP_COMMAND_USER, "USER", 4, ftp_cmd_handler_do_user},
+            {FTP_COMMAND_PASS, "PASS", 4, ftp_cmd_handler_do_pass},
+            {FTP_COMMAND_CWD, "CWD", 3, ftp_cmd_handler_do_cwd},
+            {FTP_COMMAND_PORT, "PORT", 4, ftp_cmd_handler_do_port},
+            {FTP_COMMAND_PASV, "PASV", 4, ftp_cmd_handler_do_pasv},
+            {FTP_COMMAND_EPRT, "EPRT", 4, ftp_cmd_handler_do_eprt},
+            {FTP_COMMAND_EPSV, "EPSV", 4, ftp_cmd_handler_do_epsv},
+            {FTP_COMMAND_LPRT, "LPRT", 4, ftp_cmd_handler_do_lprt},
+            {FTP_COMMAND_LPSV, "LPSV", 4, ftp_cmd_handler_do_lpsv},
+            {FTP_COMMAND_LIST, "LIST", 4, ftp_cmd_handler_do_list},
+            {FTP_COMMAND_STOR, "STOR", 4, ftp_cmd_handler_do_stor},
+            {FTP_COMMAND_RETR, "RETR", 4, ftp_cmd_handler_do_retr},
+            {FTP_COMMAND_OTHERS, NULL, 0, NULL}};
+
+    const char *ftp_command_type_to_string(enum ftp_command cmd_type)
+    {
+        for (size_t i = 0; i < sizeof(g_ftp_c2s_cmd_tuple) / sizeof(struct ftp_interact_cmd_parser); i++)
+        {
+            if (g_ftp_c2s_cmd_tuple[i].cmd_type == cmd_type && cmd_type != FTP_COMMAND_OTHERS)
+            {
+                return g_ftp_c2s_cmd_tuple[i].cmd_name;
+            }
+        }
+        return "OTHERS";
+    }
+
+    int ftp_res_handler_do_220(struct session *sess, struct ftp_decoder_exdata *ftp_ext UNUSED, struct ftp_decoder *ftp_env)
+    {
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, " '%s': reply 220", session_get_readable_addr(sess));
+        return 0;
+    }
+
+    int ftp_res_handler_do_200(struct session *sess, struct ftp_decoder_exdata *ftp_ext UNUSED, struct ftp_decoder *ftp_env)
+    {
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, " '%s': reply 200", session_get_readable_addr(sess));
+        return 0;
+    }
+
+    /*
+      example: 227 Entering Passive Mode (218,13,32,6,78,40).
+     */
+    int ftp_res_handler_do_227(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        ftp_ext->ctrl_ext.mode = FTP_TRANSFER_PASV;
+        unsigned int dst_ip_net;
+        unsigned short dst_port_net;
+        if (ftp_parse_ipv4_port_style(&ftp_ext->ctrl_ext.cmd_line.arg_refer, &dst_ip_net, &dst_port_net) < 0)
+        {
+            STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp_parse_ipv4_port_style parse error: %.*s",
+                              IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.cmd_line));
+            return -1;
+        }
+
+        struct ftp_session_addr ses_addr = {};
+        ftp_session_get_addr(sess, &ses_addr);
+        // in passive mode, new data link src ip address is client ip address
+        ftp_make_hkey_v4(&ftp_ext->ctrl_ext.last_data_link_key, ses_addr.saddr4, dst_ip_net, dst_port_net);
+
+        ftp_create_new_datalink_hash_item(ftp_env, ftp_ext);
+        int thread_id = module_manager_get_thread_id(ftp_env->mod_mgr_ref);
+        ftp_decoder_stat_incrby(thread_id, ftp_env, FTPD_STAT_NEGOTIATE_DATA_LINK, 1);
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "'%s': response 227, %.*s parsed dip:0x%x, dport:%u, hashkey:%s",
+                          session_get_readable_addr(sess), IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.arg_refer),
+                          ntohl(dst_ip_net), ntohs(dst_port_net),
+                          ftp_hash_key_to_str(&ftp_ext->ctrl_ext.last_data_link_key));
+        return 0;
+    }
+
+    /*
+      LPSV reply,
+      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_decoder_exdata *ftp_ext UNUSED, struct ftp_decoder *ftp_env)
+    {
+        ftp_ext->ctrl_ext.mode = FTP_TRANSFER_PASV;
+        // todo
+        STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "'%s': response 228 command not support yet!", session_get_readable_addr(sess));
+        return 0;
+    }
+
+    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];
+        char raw_cmd_str_tmp[FTP_CMD_MAX_LENGTH + 1];
+        memset(raw_cmd_str_tmp, 0, sizeof(raw_cmd_str_tmp));
+        memcpy(raw_cmd_str_tmp, cmd_str->iov_base, cmd_str->iov_len);
+
+        char only_integer_str_tmp[FTP_CMD_MAX_LENGTH + 1];
+        memset(only_integer_str_tmp, 0, sizeof(only_integer_str_tmp));
+        if (sscanf(raw_cmd_str_tmp, "%*[a-zA-Z (|]%255[0-9]", only_integer_str_tmp) <= 0)
+        {
+            if (sscanf(raw_cmd_str_tmp, "|||%255s|", only_integer_str_tmp) <= 0)
+            {
+                return -1;
+            }
+        }
+        int ret = sscanf(only_integer_str_tmp, "%u", &port_host);
+        if (ret != 1)
+        {
+            return -1;
+        }
+        *port_net = htons((unsigned short)port_host);
+        return 0;
+    }
+    /*
+      reply of EPSV command,
+      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_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        ftp_ext->ctrl_ext.mode = FTP_TRANSFER_PASV;
+        unsigned short port_net;
+        if (ftp_parse_ipv6_port_style(&ftp_ext->ctrl_ext.cmd_line.arg_refer, &port_net) < 0)
+        {
+            STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "'%s': response 229 parse error, %.*s",
+                              session_get_readable_addr(sess), IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.arg_refer));
+            return -1;
+        }
+
+        struct ftp_session_addr ses_addr = {};
+        ftp_session_get_addr(sess, &ses_addr);
+        if (AF_INET6 != ses_addr.af_inet && AF_INET != ses_addr.af_inet)
+        {
+            STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "session: %s, parse response 229 error! %.*s",
+                              session_get_readable_addr(sess), IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.arg_refer));
+            return -1;
+        }
+        // in passive mode, new data link src ip address is client ip address
+        if (AF_INET == ses_addr.af_inet)
+        {
+            ftp_make_hkey_v4(&ftp_ext->ctrl_ext.last_data_link_key, ses_addr.saddr4, ses_addr.daddr4, port_net);
+        }
+        else
+        {
+            ftp_make_hkey_v6(&ftp_ext->ctrl_ext.last_data_link_key, (struct in6_addr *)&ses_addr.saddr6, (struct in6_addr *)&ses_addr.daddr6, port_net);
+        }
+        ftp_create_new_datalink_hash_item(ftp_env, ftp_ext);
+        int thread_id = module_manager_get_thread_id(ftp_env->mod_mgr_ref);
+        ftp_decoder_stat_incrby(thread_id, ftp_env, FTPD_STAT_NEGOTIATE_DATA_LINK, 1);
+        STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "'%s': EPSV response 229, port is:%u", session_get_readable_addr(sess), ntohs(port_net));
+        return 0;
+    }
+
+    static const struct ftp_interact_reply_parser g_ftp_s2c_cmd_tuple[] =
+        {
+            {FTP_REPLY_CMD_OK, "220", 3, ftp_res_handler_do_220},
+            {FTP_REPLY_SERVICE_READY, "200", 3, ftp_res_handler_do_200},
+            {FTP_REPLY_ENTER_PASV_MODE, "227", 3, ftp_res_handler_do_227},
+            {FTP_REPLY_ENTER_LONG_PASV_MODE, "228", 3, ftp_res_handler_do_228},
+            {FTP_REPLY_ENTER_EXTEND_PASSIVE_MODE, "229", 3, ftp_res_handler_do_229},
+            {FTP_REPLY_OTHRES, NULL, 0, NULL}};
+
+    const char *ftp_reply_type_to_string(enum ftp_reply_code reply_type)
+    {
+        for (size_t i = 0; i < sizeof(g_ftp_s2c_cmd_tuple) / sizeof(struct ftp_interact_reply_parser); i++)
+        {
+            if (g_ftp_s2c_cmd_tuple[i].reply_code_type == reply_type && reply_type != FTP_REPLY_OTHRES)
+            {
+                return g_ftp_s2c_cmd_tuple[i].reply_code;
+            }
+        }
+        return "OTHERS";
+    }
+
+    int ftp_ctrl_identify(struct session *sess UNUSED, const char *payload, size_t len, enum flow_type curdir)
+    {
+#if 0        
+        if(ftp_ctrl_identify_by_addr(sess)){
+            return 1;
+        }
+#endif
+        return ftp_ctrl_identify_by_payload(payload, len, curdir);
+    }
+
+    struct ftp_decoder_exdata *ftp_data_identify(struct session *sess, struct ftp_decoder *ftp_env)
+    {
+        ftp_hash_key_t hash_key = {};
+        ftp_session_get_addr(sess, &hash_key);
+        if (hash_key.af_inet == 0)
+        {
+            STELLAR_LOG_WARN(ftp_env->logger_ref, FTP_MODULE_NAME, "data link identify(): get session addr fail: %s", session_get_readable_addr(sess));
+            return NULL;
+        }
+        /* sport is random value in passive mode, so set sport=0 for all active and passive mode */
+        hash_key.sport = 0;
+        struct ftp_datalink_htable *hitem = ftp_hash_search(&hash_key);
+        if (NULL == hitem)
+        {
+            STELLAR_LOG_DEBUG(ftp_env->logger_ref, FTP_MODULE_NAME, "not ftp session. hashkey: %s, session: %s",
+                              ftp_hash_key_to_str(&hash_key), session_get_readable_addr(sess));
+            return NULL;
+        }
+        struct ftp_decoder_exdata *data_ext = hitem->ftp_ext;
+        return data_ext;
+    }
+
+    static const struct ftp_interact_cmd_parser *ftp_fetch_cmd_parser(const struct ftp_interact_cmd_parser *table,
+                                                                      size_t table_item_max, const struct ftp_interact_line *cmd_result)
+    {
+        for (size_t i = 0; i < table_item_max && 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 &table[table_item_max - 1];
+    }
+
+    static const struct ftp_interact_reply_parser *ftp_fetch_reply_parser(const struct ftp_interact_reply_parser *table,
+                                                                          size_t table_item_max, const struct ftp_interact_line *cmd_result)
+    {
+        for (size_t i = 0; i < table_item_max && table[i].reply_code != NULL; i++)
+        {
+            if (0 == strncmp(table[i].reply_code, (char *)cmd_result->cmd_refer.iov_base, table[i].reply_len))
+            {
+                return &table[i];
+            }
+        }
+        return &table[table_item_max - 1];
+    }
+
+    int ftp_command_process(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        const struct ftp_interact_cmd_parser *cmd_parser = NULL;
+        cmd_parser = ftp_fetch_cmd_parser(g_ftp_c2s_cmd_tuple,
+                                          sizeof(g_ftp_c2s_cmd_tuple) / sizeof(struct ftp_interact_cmd_parser),
+                                          &ftp_ext->ctrl_ext.cmd_line);
+        ftp_ext->ctrl_ext.cmd_type = cmd_parser->cmd_type;
+        if (NULL == cmd_parser->cmd_handler)
+        {
+            STELLAR_LOG_INFO(ftp_env->logger_ref, FTP_MODULE_NAME, "fetch parser() failed, '%s', not support cmd %.*s",
+                             session_get_readable_addr(sess), IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.cmd_line));
+            return 0;
+        }
+        return cmd_parser->cmd_handler(sess, ftp_ext, ftp_env);
+    }
+
+    int ftp_reply_process(struct session *sess, struct ftp_decoder_exdata *ftp_ext, struct ftp_decoder *ftp_env)
+    {
+        const struct ftp_interact_reply_parser *reply_parser = NULL;
+
+        reply_parser = ftp_fetch_reply_parser(g_ftp_s2c_cmd_tuple,
+                                              sizeof(g_ftp_s2c_cmd_tuple) / sizeof(struct ftp_interact_reply_parser),
+                                              &ftp_ext->ctrl_ext.cmd_line);
+        ftp_ext->ctrl_ext.reply_code = reply_parser->reply_code_type;
+        if (NULL == reply_parser->reply_handler)
+        {
+            STELLAR_LOG_INFO(ftp_env->logger_ref, FTP_MODULE_NAME, "fetch parser() failed, '%s', not support cmd %.*s",
+                             session_get_readable_addr(sess), IOVEC_PRINT(ftp_ext->ctrl_ext.cmd_line.cmd_line));
+            return 0;
+        }
+        return reply_parser->reply_handler(sess, ftp_ext, ftp_env);
+    }
+
+#ifdef __cplusplus
+}
+#endif
diff --git a/decoders/ftp/ftp_decoder_stat.c b/decoders/ftp/ftp_decoder_stat.c
new file mode 100644
index 0000000..f06e5bf
--- /dev/null
+++ b/decoders/ftp/ftp_decoder_stat.c
@@ -0,0 +1,55 @@
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+#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_name_id_tuple g_ftpd_stat_tuple[] =
+        {
+            {FTPD_STAT_CTRL_LINK_NEW, "clink_open"},
+            {FTPD_STAT_CTRL_LINK_FREE, "clink_close"},
+            {FTPD_STAT_DATA_LINK_NEW, "dlink_open"},
+            {FTPD_STAT_DATA_LINK_FREE, "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_DATA_LINK_HTABLE_ITEMS, "dlink_hash_items"},
+            {FTPD_STAT_CTRL_CMD, "ctrl_commands"},
+    };
+
+    void ftp_decoder_stat_incrby(int thread_idx, struct ftp_decoder *ftp_env, enum ftp_decoder_stat_type stattype, long long increment)
+    {
+        if (ftp_env->stat.fs4_instance)
+        {
+            fieldstat_easy_counter_incrby(ftp_env->stat.fs4_instance, thread_idx, ftp_env->stat.fs4_counter_id[stattype], NULL, 0, increment);
+        }
+    }
+
+    int ftp_decoder_stat_init(struct module_manager *mod_mgr, struct ftp_decoder *ftp_env)
+    {
+        assert(sizeof(g_ftpd_stat_tuple) / sizeof(struct ftpd_stat_name_id_tuple) == FTPD_STAT_MAX);
+        int thread_count = module_manager_get_max_thread_num(mod_mgr);
+        ftp_env->stat.fs4_instance = fieldstat_easy_new(thread_count, FTP_DECODER_FIELDSTAT_NAME, NULL, 0);
+        for (int i = 0; i < FTPD_STAT_MAX; i++)
+        {
+            ftp_env->stat.fs4_counter_id[i] = fieldstat_easy_register_counter(ftp_env->stat.fs4_instance, g_ftpd_stat_tuple[i].name);
+        }
+        fieldstat_easy_enable_auto_output(ftp_env->stat.fs4_instance, FTP_DECODER_FIELDSTAT_OUTPUT_FILE, FTP_DECODER_FIELDSTAT_OUTPUT_INTERVAL);
+        return 0;
+    }
+
+    void ftp_decoder_stat_free(struct ftp_decoder *ftp_env)
+    {
+        fieldstat_easy_free(ftp_env->stat.fs4_instance);
+        ftp_env->stat.fs4_instance = NULL;
+    }
+
+#ifdef __cplusplus
+}
+#endif
\ No newline at end of file
diff --git a/decoders/ftp/ftp_decoder_stat.h b/decoders/ftp/ftp_decoder_stat.h
new file mode 100644
index 0000000..b477950
--- /dev/null
+++ b/decoders/ftp/ftp_decoder_stat.h
@@ -0,0 +1,43 @@
+#pragma once
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#include <fieldstat/fieldstat_easy.h>
+#include "stellar/ftp.h"
+
+    enum ftp_decoder_stat_type
+    {
+        FTPD_STAT_CTRL_LINK_NEW = 0,
+        FTPD_STAT_CTRL_LINK_FREE,
+        FTPD_STAT_DATA_LINK_NEW,
+        FTPD_STAT_DATA_LINK_FREE,
+        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_DATA_LINK_HTABLE_ITEMS,
+        FTPD_STAT_CTRL_CMD,
+        FTPD_STAT_MAX,
+    };
+
+    struct ftpd_stat_name_id_tuple
+    {
+        enum ftp_decoder_stat_type type;
+        const char *name;
+    };
+
+    struct ftp_decoder_stat
+    {
+        struct fieldstat_easy *fs4_instance;
+        int fs4_counter_id[FTPD_STAT_MAX];
+    };
+    int ftp_decoder_stat_init(struct module_manager *mod_mgr, struct ftp_decoder *ftp_env);
+    void ftp_decoder_stat_free(struct ftp_decoder *ftp_env);
+    void ftp_decoder_stat_incrby(int thread_idx, struct ftp_decoder *ftp_env, enum ftp_decoder_stat_type stattype, long long increment);
+
+#ifdef __cplusplus
+}
+#endif
diff --git a/decoders/ftp/ftp_decoder_util.c b/decoders/ftp/ftp_decoder_util.c
new file mode 100644
index 0000000..b0e1d3c
--- /dev/null
+++ b/decoders/ftp/ftp_decoder_util.c
@@ -0,0 +1,400 @@
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#include "ftp_decoder_inner.h"
+#include "ftp_decoder_util.h"
+#include "ftp_decoder_hash.h"
+#include "stellar/session.h"
+#include "stellar/packet.h"
+#include <ctype.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <errno.h>
+#include <libgen.h>
+
+    void fstring_dup(fstring *dst, const fstring *src)
+    {
+        dst->iov_base = (char *)calloc(1, src->iov_len + 1);
+        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(fstring *dst, const fstring *src)
+    {
+        if (dst->iov_base)
+        {
+            free(dst->iov_base); // free it if exist
+        }
+        fstring_dup(dst, src);
+    }
+
+    void ftp_cmd_line_parse(const char *payload, size_t len, struct ftp_interact_line *cmd_line, int delim)
+    {
+        cmd_line->cmd_line.iov_base = (char *)payload;
+        cmd_line->cmd_line.iov_len = len;
+
+        const char *p = (char *)memchr(payload, delim, len);
+        if (NULL == p)
+        {
+            cmd_line->cmd_refer.iov_base = (char *)payload;
+            cmd_line->cmd_refer.iov_len = len;
+            cmd_line->arg_refer.iov_base = NULL;
+            cmd_line->arg_refer.iov_len = 0;
+        }
+        else
+        {
+            cmd_line->cmd_refer.iov_base = (char *)payload;
+            cmd_line->cmd_refer.iov_len = (p - payload);
+            cmd_line->arg_refer.iov_base = (char *)p + 1;
+            cmd_line->arg_refer.iov_len = len - (p + 1 - payload);
+        }
+    }
+
+    size_t ftp_strip_crlf(const char *payload, size_t len)
+    {
+        size_t new_len = 0;
+        for (size_t 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 *cmd_line, const char *payload, size_t len)
+    {
+        memset(cmd_line, 0, sizeof(struct ftp_interact_line));
+        size_t strip_crlf_len = ftp_strip_crlf(payload, len);
+        if (strip_crlf_len == 0)
+        {
+            return -1;
+        }
+        ftp_cmd_line_parse(payload, strip_crlf_len, cmd_line, ' ');
+        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;
+    }
+
+    struct ftp_dtp *ftp_dtp_deep_clone(const struct ftp_dtp *src)
+    {
+        if (NULL == src)
+        {
+            return NULL;
+        }
+        struct ftp_dtp *new_dtp = (struct ftp_dtp *)calloc(1, sizeof(struct ftp_dtp));
+        memcpy(new_dtp, src, sizeof(struct ftp_dtp));
+        if (src->uri)
+        {
+            new_dtp->uri = strdup(src->uri);
+        }
+        else
+        {
+            new_dtp->uri = NULL;
+        }
+        return new_dtp;
+    }
+
+    /*
+        from ctrl link to data link.
+    */
+    struct ftp_decoder_exdata *ftp_exdata_deep_clone(const struct ftp_decoder_exdata *src)
+    {
+        struct ftp_decoder_exdata *new_ext = (struct ftp_decoder_exdata *)calloc(1, sizeof(struct ftp_decoder_exdata));
+        memcpy(new_ext, src, sizeof(struct ftp_decoder_exdata));
+        fstring_dup(&new_ext->ftp_login_pri.username, &src->ftp_login_pri.username);
+        fstring_dup(&new_ext->ftp_login_pri.password, &src->ftp_login_pri.password);
+        new_ext->dtp = NULL; // update in retr/stor commands
+        new_ext->reference = 1;
+        return new_ext;
+    }
+
+    static void ftp_set_tcp_addr(const struct tcphdr *tcph, struct ftp_session_addr *addr, enum flow_type fdir)
+    {
+        if (FLOW_TYPE_C2S == fdir)
+        {
+            addr->sport = 0; // tcph->th_sport;
+            addr->dport = tcph->th_dport;
+        }
+        else
+        {
+            addr->sport = 0; // tcph->th_dport;
+            addr->dport = tcph->th_sport;
+        }
+    }
+    static void ftp_set_ipv4_addr(const struct ip *ip4h, struct ftp_session_addr *addr, enum flow_type fdir)
+    {
+        addr->af_inet = AF_INET;
+        if (FLOW_TYPE_C2S == fdir)
+        {
+            addr->saddr4 = ip4h->ip_src.s_addr;
+            addr->daddr4 = ip4h->ip_dst.s_addr;
+        }
+        else
+        {
+            addr->saddr4 = ip4h->ip_dst.s_addr;
+            addr->daddr4 = ip4h->ip_src.s_addr;
+        }
+    }
+    static void ftp_set_ipv6_addr(const struct ip6_hdr *ip6h, struct ftp_session_addr *addr, enum flow_type fdir)
+    {
+        addr->af_inet = AF_INET6;
+        if (FLOW_TYPE_C2S == fdir)
+        {
+            memcpy(&addr->saddr6, &ip6h->ip6_src, sizeof(struct in6_addr));
+            memcpy(&addr->daddr6, &ip6h->ip6_dst, sizeof(struct in6_addr));
+        }
+        else
+        {
+            memcpy(&addr->saddr6, &ip6h->ip6_dst, sizeof(struct in6_addr));
+            memcpy(&addr->daddr6, &ip6h->ip6_src, sizeof(struct in6_addr));
+        }
+    }
+
+    void ftp_session_get_addr(const struct session *sess, struct ftp_session_addr *addr)
+    {
+        if (sess == NULL || addr == NULL)
+        {
+            return;
+        }
+        enum flow_type fdir = session_get_flow_type(sess);
+        const struct packet *raw_pkt = session_get_first_packet(sess, fdir);
+        if (NULL == raw_pkt)
+        {
+            addr->af_inet = 0;
+            return;
+        }
+
+        int count = packet_get_layer_count(raw_pkt);
+        for (int i = count - 1; i >= 0; i--)
+        {
+            const struct layer *layer = packet_get_layer_by_idx(raw_pkt, i);
+            if (layer->proto == LAYER_PROTO_TCP)
+            {
+                ftp_set_tcp_addr(layer->hdr.tcp, addr, fdir);
+            }
+            else if (layer->proto == LAYER_PROTO_IPV4)
+            {
+                ftp_set_ipv4_addr(layer->hdr.ip4, addr, fdir);
+                break;
+            }
+            else if (layer->proto == LAYER_PROTO_IPV6)
+            {
+                ftp_set_ipv6_addr(layer->hdr.ip6, addr, fdir);
+                break;
+            }
+        }
+    }
+
+    int ftp_ctrl_identify_by_payload(const char *payload, size_t len, enum flow_type curdir)
+    {
+        if (NULL == payload || len < FTP_IDENTIRY_MIN_LEN)
+        {
+            return 0;
+        }
+        if (curdir == FLOW_TYPE_C2S)
+        {
+            if ((strncasecmp(payload, "USER", 4) == 0) ||
+                (strncasecmp(payload, "OPTS", 4) == 0) ||
+                (strncasecmp(payload, "SYST", 4) == 0) ||
+                (strncasecmp(payload, "STAT", 4) == 0))
+            {
+                return 1;
+            }
+        }
+        else
+        {
+            if (memcmp(payload, "220", 3) == 0)
+            {
+                int tmplen = MIN(len, FTP_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;
+    }
+#if 0
+    /* return value need free after used */
+    const char *ftp_path_update(const char *current_path, const char *new_cwd_path)
+    {
+        size_t current_path_len = current_path ? strlen(current_path) : 0;
+        size_t new_path_len = new_cwd_path ? strlen(new_cwd_path) : 0;
+
+        if (NULL == new_cwd_path || new_path_len == 0)
+        {
+            return strdup(current_path);
+        }
+
+        if (NULL == current_path || current_path_len == 0)
+        {
+            if (new_cwd_path)
+            {
+                if ('/' == new_cwd_path[0])
+                {
+                    return strdup(new_cwd_path);
+                }
+                else
+                {
+                    size_t tmp_len = strlen(new_cwd_path) + 2;
+                    char *new_path = (char *)calloc(1, tmp_len);
+                    snprintf(new_path, tmp_len, "/%s", new_cwd_path);
+                    return new_path;
+                }
+            }
+            return NULL;
+        }
+
+        char *new_path = (char *)calloc(1, current_path_len + new_path_len + 1);
+        if (current_path[0] == '/')
+        {
+            snprintf(new_path, PATH_MAX, "%s/%s", current_path, new_cwd_path);
+        }
+        else
+        {
+            snprintf(new_path, PATH_MAX, "/%s/%s", current_path, new_cwd_path);
+        }
+        return new_path;
+    }
+#endif
+
+    static void ftp_normalize_path(const char *input, char *output, size_t size)
+    {
+        if (!input || !output || size == 0)
+        {
+            if (output)
+            {
+                output[0] = '\0';
+            }
+            return;
+        }
+
+        char *stack[PATH_MAX]; // Use pointers for components, no deep copying.
+        int stack_index = 0;
+
+        // Temporary buffer for tokenization
+        char temp_path[PATH_MAX];
+        strncpy(temp_path, input, sizeof(temp_path) - 1);
+        temp_path[sizeof(temp_path) - 1] = '\0';
+
+        // Tokenize the path and process components
+        char *token = strtok(temp_path, "/");
+        while (token)
+        {
+            if (strcmp(token, "..") == 0)
+            {
+                if (stack_index > 0)
+                    stack_index--; // Pop from stack
+            }
+            else if (strcmp(token, ".") != 0 && strlen(token) > 0)
+            {
+                stack[stack_index++] = token; // Push valid component
+            }
+            token = strtok(NULL, "/");
+        }
+
+        // Build the normalized path
+        size_t len = 0;
+        output[0] = '\0';
+        for (int i = 0; i < stack_index; i++)
+        {
+            len += snprintf(output + len, size - len, "/%s", stack[i]);
+            if (len >= size)
+                break; // Avoid buffer overflow
+        }
+
+        // Ensure at least "/" is returned
+        if (stack_index == 0)
+            strncpy(output, "/", size - 1);
+
+        output[size - 1] = '\0'; // Null-terminate the output
+    }
+
+    /*
+     * According to the cwd, the server address, the file name of stor/retr command,
+     * concatenated them into a complete absolute uri path.
+     */
+    void ftp_join_absolute_path(const char *dir, const char *file, char *result, size_t size)
+    {
+        if (!dir || !file || !result || size == 0)
+        {
+            if (result)
+            {
+                result[0] = '\0';
+            }
+            return;
+        }
+        char combined_path[PATH_MAX] = {};
+        // If file is absolute path, use it directly
+        if (file[0] == '/')
+        {
+            strncpy(combined_path, file, sizeof(combined_path) - 1);
+            combined_path[sizeof(combined_path) - 1] = '\0';
+        }
+        else
+        {
+            if (dir[0] == '\0')
+            {
+                snprintf(combined_path, sizeof(combined_path), "/%s", file);
+            }
+            else
+            {
+                // Otherwise, join dir and file
+                if (dir[strlen(dir) - 1] == '/')
+                {
+                    snprintf(combined_path, sizeof(combined_path), "%s%s", dir, file);
+                }
+                else
+                {
+                    snprintf(combined_path, sizeof(combined_path), "%s/%s", dir, file);
+                }
+            }
+        }
+        // Normalize the combined path
+        ftp_normalize_path(combined_path, result, size);
+    }
+
+#ifdef __cplusplus
+}
+#endif
\ No newline at end of file
diff --git a/decoders/ftp/ftp_decoder_util.h b/decoders/ftp/ftp_decoder_util.h
new file mode 100644
index 0000000..004b922
--- /dev/null
+++ b/decoders/ftp/ftp_decoder_util.h
@@ -0,0 +1,35 @@
+#pragma once
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <string.h>
+#include <bits/types/struct_iovec.h>
+#include "stellar/ftp.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(fstring *dst, const fstring *src);
+    void fstring_safe_dup(fstring *dst, const fstring *src);
+    void ftp_strtolower(char *str, size_t len);
+    int ftp_mkdir_p(const char *path, mode_t mode);
+    struct ftp_decoder_exdata *ftp_exdata_deep_clone(const struct ftp_decoder_exdata *src);
+    size_t ftp_strip_crlf(const char *payload, size_t len);
+    const char *ftp_command_type_to_string(enum ftp_command cmd_type);
+    const char *ftp_reply_type_to_string(enum ftp_reply_code reply_type);
+    struct ftp_dtp *ftp_dtp_deep_clone(const struct ftp_dtp *src);
+    void ftp_join_absolute_path(const char *dir, const char *file, char *result, size_t size);
+#ifdef __cplusplus
+}
+#endif
\ No newline at end of file
diff --git a/decoders/ftp/ftp_module.c b/decoders/ftp/ftp_module.c
new file mode 100644
index 0000000..8819d60
--- /dev/null
+++ b/decoders/ftp/ftp_module.c
@@ -0,0 +1,192 @@
+#include <stdio.h>
+#include <assert.h>
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+#include "stellar/session.h"
+#include "stellar/module.h"
+#include "stellar/mq.h"
+#include "ftp_decoder_inner.h"
+#include "ftp_decoder_stat.h"
+
+    static void ftp_on_msg_dispatch(int topic_id UNUSED, void *msg, on_msg_cb_func *on_msg_cb,
+                                    void *on_msg_cb_arg, void *dispatch_arg UNUSED)
+    {
+        assert(msg != NULL && on_msg_cb != NULL);
+        struct ftp_message *ftpmsg = (struct ftp_message *)msg;
+        switch (ftpmsg->topic_type)
+        {
+        case FTP_TOPIC_CTRL_REQ_LINE:
+        {
+            ftp_on_control_request_cb *on_ctrl_req_cb = (ftp_on_control_request_cb *)((void *)on_msg_cb);
+            on_ctrl_req_cb(ftpmsg->sess_ref, ftpmsg->ftp_ext_ref->ctrl_ext.cmd_type,
+                           (const char *)ftpmsg->ftp_ext_ref->ctrl_ext.cmd_line.arg_refer.iov_base, ftpmsg->ftp_ext_ref->ctrl_ext.cmd_line.arg_refer.iov_len,
+                           (const char *)ftpmsg->ftp_ext_ref->ctrl_ext.cmd_line.cmd_line.iov_base, ftpmsg->ftp_ext_ref->ctrl_ext.cmd_line.cmd_line.iov_len,
+                           on_msg_cb_arg);
+        }
+        break;
+        case FTP_TOPIC_CTRL_RES_LINE:
+        {
+            ftp_on_control_response_cb *on_ctrl_res_cb = (ftp_on_control_response_cb *)((void *)on_msg_cb);
+            on_ctrl_res_cb(ftpmsg->sess_ref, ftpmsg->ftp_ext_ref->ctrl_ext.reply_code,
+                           (const char *)ftpmsg->ftp_ext_ref->ctrl_ext.cmd_line.cmd_line.iov_base, ftpmsg->ftp_ext_ref->ctrl_ext.cmd_line.cmd_line.iov_len,
+                           on_msg_cb_arg);
+        }
+        break;
+        case FTP_TOPIC_CTRL_DTP:
+        {
+            struct ftp_login local_login_info = {};
+            ftp_on_control_dtp_cb *on_ctrl_dtp_cb = (ftp_on_control_dtp_cb *)((void *)on_msg_cb);
+            local_login_info.username = (const char *)ftpmsg->ftp_ext_ref->ftp_login_pri.username.iov_base;
+            local_login_info.password = (const char *)ftpmsg->ftp_ext_ref->ftp_login_pri.password.iov_base;
+            on_ctrl_dtp_cb(ftpmsg->sess_ref, &local_login_info, ftpmsg->ftp_ext_ref->dtp, on_msg_cb_arg);
+        }
+        break;
+        case FTP_TOPIC_DATA_DTP:
+        {
+            struct ftp_login local_login_info = {};
+            ftp_on_data_connection_cb *on_data_dtp_cb = (ftp_on_data_connection_cb *)((void *)on_msg_cb);
+            local_login_info.username = (const char *)ftpmsg->ftp_ext_ref->ftp_login_pri.username.iov_base;
+            local_login_info.password = (const char *)ftpmsg->ftp_ext_ref->ftp_login_pri.password.iov_base;
+            on_data_dtp_cb(ftpmsg->sess_ref, &local_login_info, ftpmsg->ftp_ext_ref->dtp,
+                           ftpmsg->ftp_ext_ref->data_ext.chunk, ftpmsg->ftp_ext_ref->data_ext.chunk_size,
+                           ftpmsg->ftp_ext_ref->data_ext.offset, ftpmsg->ftp_ext_ref->data_ext.is_finished,
+                           on_msg_cb_arg);
+        }
+        break;
+        default:
+            assert(0);
+            break;
+        }
+    }
+
+    static int ftp_create_topic_nx(struct module_manager *mod_mgr, const char *topic_name)
+    {
+        struct mq_schema *mq_s = module_manager_get_mq_schema(mod_mgr);
+        assert(mq_s != NULL);
+        int topic_id = mq_schema_get_topic_id(mq_s, topic_name);
+        if (topic_id >= 0)
+        {
+            return topic_id;
+        }
+        topic_id = mq_schema_create_topic(mq_s, topic_name, (on_msg_dispatch_cb_func *)ftp_on_msg_dispatch,
+                                          NULL, ftp_msg_free_cb, NULL);
+        return topic_id;
+    }
+
+    static const struct ftp_topic_compose ftp_topic_schema[FTP_TOPIC_MAX] =
+        {
+            {FTP_TOPIC_CTRL_REQ_LINE, -1, "FTP_TOPIC_CTRL_REQ_LINE"},
+            {FTP_TOPIC_CTRL_RES_LINE, -1, "FTP_TOPIC_CTRL_RES_LINE"},
+            {FTP_TOPIC_CTRL_DTP, -1, "FTP_TOPIC_CTRL_DTP"},
+            {FTP_TOPIC_DATA_DTP, -1, "FTP_TOPIC_DATA_DTP"},
+    };
+
+    static struct ftp_topic_manager *ftp_topic_mgr_init(struct module_manager *mod_mgr)
+    {
+        struct ftp_topic_manager *http_topic_mgr = (struct ftp_topic_manager *)calloc(1, sizeof(struct ftp_topic_manager));
+        assert(http_topic_mgr != NULL);
+        struct ftp_topic_compose *topic_compose = http_topic_mgr->topic_compose;
+        for (int i = 0; i < (int)FTP_TOPIC_MAX; i++)
+        {
+            topic_compose[i].topic_type = ftp_topic_schema[i].topic_type;
+            topic_compose[i].topic_name = ftp_topic_schema[i].topic_name;
+            topic_compose[i].topic_id = ftp_create_topic_nx(mod_mgr, ftp_topic_schema[i].topic_name);
+        }
+        return http_topic_mgr;
+    }
+
+    static void ftp_topic_mgr_free(struct ftp_topic_manager *topic_mgr)
+    {
+        assert(topic_mgr != NULL);
+        FREE(topic_mgr);
+    }
+
+    static int ftp_subscribe_common(struct module_manager *mod_mgr, enum ftp_topic_type topic_type, void *cb, void *args)
+    {
+        struct module *ftp_mod = module_manager_get_module(mod_mgr, FTP_MODULE_NAME);
+        struct ftp_decoder *ftp_env = module_to_ftp_decoder(ftp_mod);
+        struct ftp_topic_manager *ftp_topic_mgr = ftp_env->ftp_topic_mgr;
+        if (ftp_topic_mgr == NULL)
+        {
+            ftp_topic_mgr = ftp_topic_mgr_init(mod_mgr);
+            ftp_env->ftp_topic_mgr = ftp_topic_mgr;
+        }
+        return mq_schema_subscribe(module_manager_get_mq_schema(mod_mgr),
+                                   ftp_topic_mgr->topic_compose[topic_type].topic_id,
+                                   (on_msg_cb_func *)cb, args);
+    }
+
+    int ftp_subscribe(struct ftp_decoder *ftp_decoder,
+                      ftp_on_control_request_cb *on_control_request_cb,
+                      ftp_on_control_response_cb *on_control_response_cb,
+                      ftp_on_control_dtp_cb *on_control_dtp_cb,
+                      ftp_on_data_connection_cb *on_data_connection_cb, void *arg)
+    {
+        assert(ftp_decoder != NULL);
+        struct module_manager *mod_mgr = ftp_decoder->mod_mgr_ref;
+        if (on_control_request_cb != NULL)
+        {
+            ftp_subscribe_common(mod_mgr, FTP_TOPIC_CTRL_REQ_LINE, (void *)on_control_request_cb, arg);
+        }
+        if (on_control_response_cb != NULL)
+        {
+            ftp_subscribe_common(mod_mgr, FTP_TOPIC_CTRL_RES_LINE, (void *)on_control_response_cb, arg);
+        }
+        if (on_control_dtp_cb != NULL)
+        {
+            ftp_subscribe_common(mod_mgr, FTP_TOPIC_CTRL_DTP, (void *)on_control_dtp_cb, arg);
+        }
+        if (on_data_connection_cb != NULL)
+        {
+            ftp_subscribe_common(mod_mgr, FTP_TOPIC_DATA_DTP, (void *)on_data_connection_cb, arg);
+        }
+
+        return 0;
+    }
+
+    struct module *ftp_init(struct module_manager *mod_mgr)
+    {
+        assert(mod_mgr != NULL);
+        struct ftp_decoder *ftp_env = (struct ftp_decoder *)calloc(1, sizeof(struct ftp_decoder));
+
+        ftp_decoder_stat_init(mod_mgr, ftp_env);
+        struct module *mod = module_new(FTP_MODULE_NAME, ftp_env);
+        ftp_env->mod_mgr_ref = mod_mgr;
+        ftp_env->logger_ref = module_manager_get_logger(mod_mgr);
+        assert(ftp_env->logger_ref != NULL);
+        struct module *sess_mod = module_manager_get_module(mod_mgr, SESSION_MANAGER_MODULE_NAME);
+        struct session_manager *sess_mgr = module_to_session_manager(sess_mod);
+        assert(sess_mgr != NULL);
+
+        struct ftp_topic_manager *ftp_topic_mgr = ftp_topic_mgr_init(mod_mgr);
+        assert(ftp_topic_mgr != NULL);
+        ftp_env->ftp_topic_mgr = ftp_topic_mgr;
+
+        session_manager_subscribe_tcp_stream(sess_mgr, ftp_on_tcp_stream_cb, ftp_env);
+        ftp_env->exdata_id = session_manager_new_session_exdata_index(sess_mgr, FTP_DECODER_DXDATA_NAME, ftp_decoder_exdata_free_cb, ftp_env);
+        STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp init success");
+        return mod;
+    }
+
+    struct ftp_decoder *module_to_ftp_decoder(struct module *ftp_mod)
+    {
+        assert(ftp_mod);
+        return (struct ftp_decoder *)module_get_ctx(ftp_mod);
+    }
+
+    void ftp_exit(struct module_manager *mod_mgr UNUSED, struct module *mod)
+    {
+        assert(mod != NULL);
+        struct ftp_decoder *ftp_env = (struct ftp_decoder *)module_get_ctx(mod);
+        ftp_decoder_stat_free(ftp_env);
+        ftp_topic_mgr_free(ftp_env->ftp_topic_mgr);
+        STELLAR_LOG_FATAL(ftp_env->logger_ref, FTP_MODULE_NAME, "ftp exit!");
+        FREE(ftp_env);
+        module_free(mod);
+    }
+
+#ifdef __cplusplus
+}
+#endif
diff --git a/decoders/ftp/version.map b/decoders/ftp/version.map
new file mode 100644
index 0000000..5acdfc3
--- /dev/null
+++ b/decoders/ftp/version.map
@@ -0,0 +1,12 @@
+VERS_3.0{
+global: 
+	extern "C" {	
+		ftp_init;
+		ftp_exit;
+		ftp_on_thread_init;
+		ftp_on_thread_exit;
+		ftp_subscribe;
+		module_to_ftp_decoder
+	};
+  local: *;
+};