rtp decoder initialdev-rtp-decoder

13 files changed, 1046 insertions, 2 deletions

diff --git a/decoders/CMakeLists.txt b/decoders/CMakeLists.txt
index efad779..4b931b6 100644
--- a/decoders/CMakeLists.txt
+++ b/decoders/CMakeLists.txt
@@ -3,4 +3,5 @@ add_subdirectory(lpi_plus)
 #add_subdirectory(http)
 #add_subdirectory(socks)
 #add_subdirectory(stratum)
-#add_subdirectory(session_flags)
\ No newline at end of file
+#add_subdirectory(session_flags)
+add_subdirectory(rtp)
diff --git a/decoders/rtp/CMakeLists.txt b/decoders/rtp/CMakeLists.txt
new file mode 100644
index 0000000..50475fd
--- /dev/null
+++ b/decoders/rtp/CMakeLists.txt
@@ -0,0 +1,9 @@
+add_library(
+    rtp
+    rtp.c
+)
+
+set_target_properties(
+    rtp PROPERTIES
+    LINK_FLAGS "-Wl,--version-script=${CMAKE_CURRENT_SOURCE_DIR}/version.map"
+)
diff --git a/decoders/rtp/rtp.c b/decoders/rtp/rtp.c
new file mode 100644
index 0000000..03bd10e
--- /dev/null
+++ b/decoders/rtp/rtp.c
@@ -0,0 +1,271 @@
+#include "stellar/packet.h"
+#include "stellar/utils.h"
+#include "stellar/mq.h"
+#include "stellar/session.h"
+#include "stellar/rtp.h"
+
+#include "rtp_internal.h"
+
+static void rtp_message_free(void *msg, void *arg)
+{
+    UNUSED(arg);
+    if (msg) {
+        free(msg);
+    }
+}
+
+static void rtp_message_dispatch(int topic, void *msg, on_msg_cb_func *msg_cb, void *arg, void *dispatch_arg)
+{
+    UNUSED(topic);
+    UNUSED(dispatch_arg);
+    struct rtp_message *rtp_msg;
+    rtp_callback_func *rtp_msg_cb;
+
+    rtp_msg = (struct rtp_message *)msg;
+    rtp_msg_cb = (rtp_callback_func *)(void *)msg_cb;
+    rtp_msg_cb(rtp_msg->sess, rtp_msg->hdr, rtp_msg->payload, rtp_msg->payload_len, arg);
+}
+
+static int rtp_decode(struct rtp_module_ctx *mod_ctx, struct rtp_exdata *exdata, struct session *sess, const struct packet *pkt)
+{
+    UNUSED(exdata);
+    int ret;
+    struct rtp_message *msg;
+
+    msg = (struct rtp_message *)calloc(1, sizeof(struct rtp_message));
+    msg->sess = sess;
+    msg->hdr = (struct rtp_header *)packet_get_payload_data(pkt);
+    msg->payload = (char *)msg->hdr + RTP_HEADER_LEN;
+    msg->payload_len = packet_get_payload_len(pkt) - RTP_HEADER_LEN;
+
+    ret = mq_runtime_publish_message(stellar_module_manager_get_mq_runtime(mod_ctx->mod_mgr), mod_ctx->topic_id, msg);
+    if (ret != 0) {
+        rtp_message_free(msg, NULL);
+    }
+
+    return ret;
+}
+
+static enum rtp_identify_state rtp_session_identify(struct rtp_module_ctx *mod_ctx, struct rtp_exdata *exdata, struct session *sess, const struct packet *pkt)
+{
+    UNUSED(mod_ctx);
+    UNUSED(sess);
+    unsigned short src_port, dst_port;
+    unsigned short last_seq, curr_seq;
+    unsigned int last_ssrc, curr_ssrc;
+    size_t pkt_len;
+    enum packet_direction pkt_dir;
+    const struct layer *packet_layer;
+    struct stun_header *stun_hdr;
+    struct rtp_header* rtp_hdr;
+
+    if (exdata->identify_state == RTP_IDENTIFY_STATE_FALSE ||
+        exdata->identify_state == RTP_IDENTIFY_STATE_TRUE) {
+        goto exit;
+    }
+
+    if (exdata->identify_times++ > RTP_IDENTIFY_TIMES_MAX) {
+        exdata->sess_ignore = 1;
+    }
+
+    pkt_len = packet_get_payload_len(pkt);
+    if (pkt_len < RTP_HEADER_LEN) {
+        exdata->identify_state = RTP_IDENTIFY_STATE_TRUE;
+        goto exit;
+    }
+
+    packet_layer = packet_get_layer_by_idx(pkt, packet_get_layer_count(pkt) - 1);
+    src_port = packet_layer->hdr.udp->source;
+    dst_port = packet_layer->hdr.udp->dest;
+    if (dst_port ==5060 || dst_port==5061 || dst_port==53 || dst_port==443 ||
+        src_port==5060 || src_port==5061 || src_port==53 || src_port==443) {
+        exdata->identify_state = RTP_IDENTIFY_STATE_FALSE;
+        goto exit;
+    }
+
+    stun_hdr = (struct stun_header *)packet_get_payload_data(pkt);
+    if(((ntohs(stun_hdr->type) == STUN_BINDING_REQUEST) ||
+            (ntohs(stun_hdr->type) == STUN_BINDING_INDECATION) ||
+            (ntohs(stun_hdr->type) == STUN_BINDING_RESPONSE) ||
+            (ntohs(stun_hdr->type) == STUN_BINDING_ERROR_RESPONSE)) &&
+            pkt_len >= STUN_HEADER_LEN && (ntohs(stun_hdr->len) == pkt_len - STUN_HEADER_LEN))  {
+        goto exit;
+    }
+
+    rtp_hdr = (struct rtp_header *)packet_get_payload_data(pkt);
+    if(rtp_hdr->version != 2 || rtp_hdr->padding != 0 || rtp_hdr->extension != 0 || rtp_hdr->csrc_len != 0) {
+        exdata->identify_state = RTP_IDENTIFY_STATE_FALSE;
+        goto exit;
+    }
+
+    curr_seq = (unsigned short)ntohs(rtp_hdr->seq);
+    curr_ssrc = (unsigned int)ntohl(rtp_hdr->ssrc);
+
+    pkt_dir = packet_get_direction(pkt);
+    switch (exdata->identify_state) {
+        case RTP_IDENTIFY_STATE_UNKNOWN:
+            exdata->first_pkt_dir = pkt_dir;
+            exdata->identify_state = RTP_IDENTIFY_STATE_HALF_TRUE;
+            break;
+        case RTP_IDENTIFY_STATE_HALF_TRUE:
+            if(exdata->first_pkt_dir == pkt_dir) {
+                last_seq = exdata->last_client_seq;
+                last_ssrc = exdata->last_client_ssrc;
+            } else {
+                last_seq = exdata->last_server_seq;
+                last_ssrc = exdata->last_server_ssrc;
+            }
+
+            if (last_ssrc == 0 || last_seq == 0) {
+                break;
+            }
+
+            if (curr_ssrc != last_ssrc) {
+                exdata->identify_state = RTP_IDENTIFY_STATE_FALSE;
+                break;
+            }
+
+            if (curr_seq < last_seq) {
+                exdata->identify_state = RTP_IDENTIFY_STATE_FALSE;
+                break;
+            }
+
+            if (curr_seq - last_seq == 1) {
+                exdata->identify_state = RTP_IDENTIFY_STATE_TRUE;
+                break;
+            }
+            break;
+        default:
+            break;
+    }
+
+    if (exdata->first_pkt_dir == pkt_dir) {
+        exdata->last_client_seq = curr_seq;
+        exdata->last_client_ssrc = curr_ssrc;
+    } else {
+        exdata->last_server_seq = curr_seq;
+        exdata->last_server_ssrc = curr_ssrc;
+    }
+
+exit:
+    return exdata->identify_state;
+}
+
+static void rtp_session_entry(struct session *sess, enum session_state state, struct packet *pkt, void *arg)
+{
+    UNUSED(state);
+    int ret;
+    enum rtp_identify_state identify_state;
+    struct rtp_exdata *exdata;
+    struct rtp_module_ctx *mod_ctx;
+
+    if (pkt == NULL) {
+        return;
+    }
+
+    mod_ctx = (struct rtp_module_ctx *)arg;
+    exdata = (struct rtp_exdata *)session_get_exdata(sess, mod_ctx->exdata_id);
+    if (exdata == NULL) {
+        exdata= (struct rtp_exdata *)calloc(1, sizeof(struct rtp_exdata));
+        session_set_exdata(sess, mod_ctx->exdata_id, exdata); 
+    }
+
+    if (exdata->sess_ignore) {
+        return;
+    }
+
+    identify_state = rtp_session_identify(mod_ctx, exdata, sess, (const struct packet *)pkt);
+    if (identify_state != RTP_IDENTIFY_STATE_TRUE) {
+        return;
+    }
+
+    ret = rtp_decode(mod_ctx, exdata, sess, (const struct packet *)pkt);
+    if (ret != 0) {
+        exdata->sess_ignore = 1;
+        return;
+    }
+}
+
+static void rtp_exdata_free(int idx, void *ex_ptr, void *arg)
+{
+    UNUSED(idx);
+    UNUSED(arg);
+    if (ex_ptr) {
+        free(ex_ptr);
+    }
+}
+
+int rtp_subscribe(struct stellar_module_manager *mod_mgr, rtp_callback_func *cb, void *arg)
+{
+    int topic;
+    struct mq_schema *schema;
+
+    if (mod_mgr == NULL) {
+        return -1;
+    }
+
+    schema = stellar_module_manager_get_mq_schema(mod_mgr);
+    if (schema == NULL) {
+        return -1;
+    }
+
+    topic = mq_schema_get_topic_id(schema, RTP_TOPIC_NAME);
+    if (topic < 0) {
+        topic = mq_schema_create_topic(schema, RTP_TOPIC_NAME, rtp_message_dispatch, NULL, rtp_message_free, NULL);
+    }
+
+    return mq_schema_subscribe(schema, topic, (on_msg_cb_func *)(void *)cb, arg);
+}
+
+void rtp_exit(struct stellar_module_manager *mod_mgr, struct stellar_module *mod)
+{
+    struct rtp_module_ctx *mod_ctx;
+
+    if (mod_mgr && mod) {
+        mod_ctx = (struct rtp_module_ctx *)stellar_module_get_ctx(mod);
+        if (mod_ctx) {
+            free(mod_ctx);
+        }
+
+        stellar_module_free(mod);
+    }
+}
+
+struct stellar_module* rtp_init(struct stellar_module_manager *mod_mgr)
+{
+    struct mq_schema *schema;
+    struct session_manager *sess_mgr;
+    struct stellar_module *mod;
+    struct rtp_module_ctx *mod_ctx;
+
+    mod_ctx = (struct rtp_module_ctx *)calloc(1, sizeof(struct rtp_module_ctx));
+    mod = stellar_module_new(RTP_MODULE_NAME, mod_ctx);
+    sess_mgr = stellar_module_get_session_manager(mod_mgr);
+    schema = stellar_module_manager_get_mq_schema(mod_mgr);
+
+    if (mod_mgr == NULL || sess_mgr == NULL || schema == NULL) {
+        goto exit;
+    }
+
+    mod_ctx->exdata_id = session_manager_new_session_exdata_index(sess_mgr, RTP_EXDATA_NAME, rtp_exdata_free, NULL);	
+    if (mod_ctx->exdata_id < 0) {
+        goto exit;
+    }
+
+    mod_ctx->topic_id = mq_schema_get_topic_id(schema, RTP_TOPIC_NAME);
+    if (mod_ctx->topic_id < 0) {
+        mod_ctx->topic_id = mq_schema_create_topic(schema, RTP_TOPIC_NAME, rtp_message_dispatch, mod_ctx, rtp_message_free, NULL);
+        if (mod_ctx->topic_id < 0) {
+            goto exit;
+        }
+    }
+
+    session_manager_subscribe_udp(sess_mgr, rtp_session_entry, mod_ctx);
+
+    return mod;
+exit:
+    rtp_exit(mod_mgr, mod);
+    return NULL;
+}
+
+
diff --git a/decoders/rtp/rtp_internal.h b/decoders/rtp/rtp_internal.h
new file mode 100644
index 0000000..5d8983d
--- /dev/null
+++ b/decoders/rtp/rtp_internal.h
@@ -0,0 +1,73 @@
+#pragma once
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#ifndef UNUSED
+#define UNUSED(x) (void)(x)
+#endif
+    
+#define RTP_MODULE_NAME     "RTP_MODULE"
+#define RTP_EXDATA_NAME     "RTP_EXDATA"
+#define RTP_TOPIC_NAME      "RTP_TOPIC"
+#define RTP_IDENTIFY_TIMES_MAX 128
+#define RTP_HEADER_LEN          12
+#define STUN_HEADER_LEN         20
+
+enum rtp_identify_state {
+    RTP_IDENTIFY_STATE_UNKNOWN,
+    RTP_IDENTIFY_STATE_HALF_TRUE,
+    RTP_IDENTIFY_STATE_TRUE,
+    RTP_IDENTIFY_STATE_FALSE,
+};
+
+enum stun_type {
+    STUN_BINDING_REQUEST            = 0x0001,
+    STUN_BINDING_INDECATION         = 0x0011,
+    STUN_BINDING_RESPONSE           = 0x0101,
+    STUN_BINDING_ERROR_RESPONSE     = 0x0111,
+};
+
+struct stun_header {
+    /* little-endian */
+    /* byte 0-1 */
+    unsigned short type;
+    /* byte 2-3 */
+    unsigned short len;
+    /* bytes 4-7 */
+    unsigned int magic_cookie;
+} __attribute__ ((packed));
+
+struct rtp_message {
+    struct session *sess;
+
+    struct rtp_header *hdr;
+    const char *payload;
+    size_t payload_len;
+};
+
+struct rtp_exdata {
+    enum rtp_identify_state identify_state;
+    int                     identify_times;
+
+    enum packet_direction   first_pkt_dir;
+    unsigned short          last_client_seq;
+    unsigned short          last_server_seq;
+    unsigned int            last_client_ssrc;
+    unsigned int            last_server_ssrc;
+
+    int             sess_ignore;
+};
+
+struct rtp_module_ctx {
+    int topic_id;
+    int exdata_id;
+
+    struct stellar_module_manager *mod_mgr;
+};
+
+#ifdef __cplusplus
+}
+#endif
diff --git a/decoders/rtp/version.map b/decoders/rtp/version.map
new file mode 100644
index 0000000..48516ec
--- /dev/null
+++ b/decoders/rtp/version.map
@@ -0,0 +1,11 @@
+VERS_2.4{
+global:
+extern "C" {
+        rtp_init;
+        rtp_exit;
+        rtp_subscribe;
+        GIT_VERSION_*;
+};
+
+local: *;
+};
diff --git a/include/stellar/rtp.h b/include/stellar/rtp.h
new file mode 100644
index 0000000..566ba6d
--- /dev/null
+++ b/include/stellar/rtp.h
@@ -0,0 +1,98 @@
+#pragma once
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#include "stellar/session.h"
+#include "stellar/module_manager.h"
+
+/*
+ *  http://www.iana.org/assignments/rtp-parameters
+ *  RTP Payload types 
+ */
+#define PT_PCMU           0  /* RFC 3551 */
+#define PT_1016           1  /* RFC 1890 (reserved in RFC 3551) */
+#define PT_G721           2  /* RFC 1890 (reserved in RFC 3551) */
+#define PT_GSM            3  /* RFC 3551 */
+#define PT_G723           4  /* From Vineet Kumar of Intel; see the Web page */
+#define PT_DVI4_8000      5  /* RFC 3551 */
+#define PT_DVI4_16000     6  /* RFC 3551 */
+#define PT_LPC            7  /* RFC 3551 */
+#define PT_PCMA           8  /* RFC 3551 */
+#define PT_G722           9  /* RFC 3551 */
+#define PT_L16_STEREO    10  /* RFC 3551 */
+#define PT_L16_MONO      11  /* RFC 3551 */
+#define PT_QCELP         12  /* Qualcomm Code Excited Linear Predictive coding? */
+#define PT_CN            13  /* RFC 3389 */
+#define PT_MPA           14  /* RFC 3551, RFC 2250 */
+#define PT_G728          15  /* RFC 3551 */
+#define PT_DVI4_11025    16  /* from Joseph Di Pol of Sun; see the Web page */
+#define PT_DVI4_22050    17  /* from Joseph Di Pol of Sun; see the Web page */
+#define PT_G729          18
+#define PT_CN_OLD        19  /* Payload type reserved (old version Comfort Noise) */
+#define PT_CELB          25  /* RFC 2029 */
+#define PT_JPEG          26  /* RFC 2435 */
+#define PT_NV            28  /* RFC 1890 */
+#define PT_H261          31  /* RFC 2032 */
+#define PT_MPV           32  /* RFC 2250 */
+#define PT_MP2T          33  /* RFC 2250 */
+#define PT_H263          34  /* from Chunrong Zhu of Intel; see the Web page */
+
+/* Added to by Alex Lindberg to cover port ranges 96-127 - Dynamic RTP
+   Some of these ports are used by Avaya for Modem and FAX support */
+#define PT_UNDF_96       96  /* RFC 3551 */
+#define PT_UNDF_97       97
+#define PT_UNDF_98       98
+#define PT_UNDF_99       99
+#define PT_UNDF_100     100
+#define PT_UNDF_101     101
+#define PT_UNDF_102     102
+#define PT_UNDF_103     103
+#define PT_UNDF_104     104
+#define PT_UNDF_105     105
+#define PT_UNDF_106     106
+#define PT_UNDF_107     107
+#define PT_UNDF_108     108
+#define PT_UNDF_109     109
+#define PT_UNDF_110     110
+#define PT_UNDF_111     111
+#define PT_UNDF_112     112
+#define PT_UNDF_113     113
+#define PT_UNDF_114     114
+#define PT_UNDF_115     115
+#define PT_UNDF_116     116
+#define PT_UNDF_117     117
+#define PT_UNDF_118     118
+#define PT_UNDF_119     119
+#define PT_UNDF_120     120
+#define PT_UNDF_121     121
+#define PT_UNDF_122     122
+#define PT_UNDF_123     123
+#define PT_UNDF_124     124
+#define PT_UNDF_125     125
+#define PT_UNDF_126     126
+#define PT_UNDF_127     127
+
+struct rtp_header {
+    unsigned char csrc_len:4;
+    unsigned char extension:1;
+    unsigned char padding:1;
+    unsigned char version:2;
+
+    unsigned char payload_type:7;
+    unsigned char marker:1;
+
+    unsigned short seq;
+    unsigned int timestamp;
+    unsigned int ssrc;
+} __attribute__ ((packed));
+
+
+typedef void rtp_callback_func(struct session *sess, const struct rtp_header *hdr, const char *payload, size_t payload_len, void *arg);
+int rtp_subscribe(struct stellar_module_manager *mod_mgr, rtp_callback_func *cb, void *arg);
+
+#ifdef __cplusplus
+}
+#endif
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
index a402bf4..6eb0986 100644
--- a/test/CMakeLists.txt
+++ b/test/CMakeLists.txt
@@ -5,4 +5,5 @@ add_subdirectory(lpi_plus)
 #add_subdirectory(decoders/http)
 #add_subdirectory(decoders/socks)
 #add_subdirectory(decoders/stratum)
-#add_subdirectory(decoders/session_flags)
\ No newline at end of file
+#add_subdirectory(decoders/session_flags)
+add_subdirectory(decoders/rtp)
diff --git a/test/decoders/rtp/CMakeLists.txt b/test/decoders/rtp/CMakeLists.txt
new file mode 100644
index 0000000..c9389e8
--- /dev/null
+++ b/test/decoders/rtp/CMakeLists.txt
@@ -0,0 +1,46 @@
+set(TEST_NAME rtp_test)
+set(TEST_MAIN ${CMAKE_CURRENT_BINARY_DIR}/${TEST_NAME})
+file(GLOB TEST_SRC "${TEST_NAME}*.cpp")
+
+add_executable(
+    ${TEST_NAME}
+    ${TEST_SRC}
+)
+
+target_include_directories(
+    ${TEST_NAME} PRIVATE
+    ${CMAKE_SOURCE_DIR}/deps/
+    ${CMAKE_SOURCE_DIR}/decoders/
+)
+
+target_link_libraries(
+    ${TEST_NAME}
+    rtp
+    stellar_lib
+    cjson-static
+    dl "-rdynamic"
+    gtest
+    gmock
+)
+
+add_test(
+    NAME ${TEST_NAME}.SETUP
+    COMMAND sh -c " 
+	cat ${CMAKE_CURRENT_SOURCE_DIR}/conf/stellar.toml > ${CMAKE_CURRENT_BINARY_DIR}/stellar.toml &&
+        cat ${CMAKE_CURRENT_SOURCE_DIR}/conf/spec.toml >> ${CMAKE_CURRENT_BINARY_DIR}/stellar.toml &&
+        cat ${CMAKE_CURRENT_SOURCE_DIR}/result/rtp_result.json >> ${CMAKE_CURRENT_BINARY_DIR}/rtp_result.json
+    "
+)
+                                            
+add_test(
+    NAME ${TEST_NAME}
+    COMMAND sh -c "
+        find ${CMAKE_CURRENT_SOURCE_DIR}/pcap/ -type f | sort -V >  ${CMAKE_CURRENT_BINARY_DIR}/pcaplist.txt &&
+        ${TEST_MAIN}
+    "
+)
+
+set_tests_properties(
+    ${TEST_NAME} PROPERTIES
+    FIXTURES_REQUIRED ${TEST_NAME}.SETUP
+)
diff --git a/test/decoders/rtp/conf/spec.toml b/test/decoders/rtp/conf/spec.toml
new file mode 100644
index 0000000..1360252
--- /dev/null
+++ b/test/decoders/rtp/conf/spec.toml
@@ -0,0 +1,11 @@
+# stellar_plugin.toml
+#
+[[module]]
+path = ""
+init = "rtp_init"
+exit = "rtp_exit"
+
+[[module]]
+path = ""
+init = "rtp_test_init"
+exit = "rtp_test_exit"
diff --git a/test/decoders/rtp/conf/stellar.toml b/test/decoders/rtp/conf/stellar.toml
new file mode 100644
index 0000000..c7ce725
--- /dev/null
+++ b/test/decoders/rtp/conf/stellar.toml
@@ -0,0 +1,79 @@
+[instance]
+    id = 1 # range: [0, 4095] (12 bit)
+
+[packet_io]
+    mode = "pcaplist"                      # pcapfile, pcaplist, marsio
+    app_symbol = "stellar"
+    dev_symbol = "nf_0_fw"
+    pcap_path = "pcaplist.txt"
+    thread_num = 1                         # range: [1, 256]
+    cpu_mask = [5, 6, 7, 8, 9, 10, 11, 12]
+    idle_yield_ms = 900                    # range: [0, 60000] (ms)
+
+    [packet_io.packet_pool]
+        capacity = 1024       # range: [1, 4294967295]
+
+    [packet_io.ip_reassembly]
+        fail_action = 1       # 0: bypass, 1: drop
+        timeout_ms = 1000     # range: [1, 60000] (ms)
+        frag_queue_num = 1024 # range: [1, 4294967295]
+        frag_queue_size = 64  # range: [2, 65535]
+
+[session_manager]
+    tcp_session_max = 50000
+    udp_session_max = 50000
+
+    evict_old_on_tcp_table_limit = 1 # range: [0, 1]
+    evict_old_on_udp_table_limit = 1 # range: [0, 1]
+
+    expire_period_ms = 0    # range: [0, 60000] (ms)
+    expire_batch_max = 1024 # range: [1, 1024]
+
+    [session_manager.tcp_timeout_ms]
+        init = 5000             # range: [1, 60000] (ms)
+        handshake = 5000        # range: [1, 60000] (ms)
+        data = 5000             # range: [1, 15999999000] (ms)
+        half_closed = 5000      # range: [1, 604800000] (ms)
+        time_wait = 5000        # range: [1, 600000] (ms)
+        discard_default = 10000 # range: [1, 15999999000] (ms)
+        unverified_rst = 5000   # range: [1, 600000] (ms)
+
+    [session_manager.udp_timeout_ms]
+        data = 5000            # range: [1, 15999999000] (ms)
+        discard_default = 5000 # range: [1, 15999999000] (ms)
+
+    [session_manager.duplicated_packet_bloom_filter]
+        enable = 1
+        capacity = 1000000     # range: [1, 4294967295]
+        time_window_ms = 10000 # range: [1, 60000] (ms)
+        error_rate = 0.00001   # range: [0.0, 1.0]
+
+    [session_manager.evicted_session_bloom_filter]
+        enable = 1             # range: [0, 1]
+        capacity = 1000000     # range: [1, 4294967295]
+        time_window_ms = 10000 # range: [1, 60000] (ms)
+        error_rate = 0.00001   # range: [0.0, 1.0]
+
+    [session_manager.tcp_reassembly]
+        enable = 1                  # range: [0, 1]
+        timeout_ms = 10000          # range: [1, 60000] (ms)
+        buffered_segments_max = 256 # range: [2, 4096] per flow
+
+[log]
+    output = "both"           # stderr, file, both
+    file = "stellar.log"
+    level = "INFO"            # TRACE, DEBUG, INFO, WARN, ERROR, FATAL
+
+[[module]]
+    path = ""
+    init = "packet_manager_on_init"
+    exit = "packet_manager_on_exit"
+    thread_init = "packet_manager_on_thread_init"
+    thread_exit = "packet_manager_on_thread_exit"
+
+[[module]]
+    path = ""
+    init = "session_manager_on_init"
+    exit = "session_manager_on_exit"
+    thread_init = "session_manager_on_thread_init"
+    thread_exit = "session_manager_on_thread_exit"
diff --git a/test/decoders/rtp/pcap/01-rtp-10-packet-PCMA.pcap b/test/decoders/rtp/pcap/01-rtp-10-packet-PCMA.pcap
new file mode 100644
index 0000000..a8d063f
--- /dev/null
+++ b/test/decoders/rtp/pcap/01-rtp-10-packet-PCMA.pcap
diff --git a/test/decoders/rtp/result/rtp_result.json b/test/decoders/rtp/result/rtp_result.json
new file mode 100644
index 0000000..fa626b7
--- /dev/null
+++ b/test/decoders/rtp/result/rtp_result.json
@@ -0,0 +1,119 @@
+[
+  {
+    "CSRC_LEN": 0,
+    "EXTENSION": 0,
+    "PADDING": 0,
+    "VERSION": 2,
+    "PAYLOAD_TYPE": "PCMA",
+    "MARKER": 0,
+    "SEQ": 2,
+    "TIMESTAMP": 320,
+    "SSRC": 3535621694,
+    "PAYLOAD_LEN": 160,
+    "PAYLOAD_SEQ": 0
+  },
+  {
+    "CSRC_LEN": 0,
+    "EXTENSION": 0,
+    "PADDING": 0,
+    "VERSION": 2,
+    "PAYLOAD_TYPE": "PCMA",
+    "MARKER": 0,
+    "SEQ": 3,
+    "TIMESTAMP": 480,
+    "SSRC": 3535621694,
+    "PAYLOAD_LEN": 160,
+    "PAYLOAD_SEQ": 1
+  },
+  {
+    "CSRC_LEN": 0,
+    "EXTENSION": 0,
+    "PADDING": 0,
+    "VERSION": 2,
+    "PAYLOAD_TYPE": "PCMA",
+    "MARKER": 0,
+    "SEQ": 4,
+    "TIMESTAMP": 640,
+    "SSRC": 3535621694,
+    "PAYLOAD_LEN": 160,
+    "PAYLOAD_SEQ": 2
+  },
+  {
+    "CSRC_LEN": 0,
+    "EXTENSION": 0,
+    "PADDING": 0,
+    "VERSION": 2,
+    "PAYLOAD_TYPE": "PCMA",
+    "MARKER": 0,
+    "SEQ": 5,
+    "TIMESTAMP": 800,
+    "SSRC": 3535621694,
+    "PAYLOAD_LEN": 160,
+    "PAYLOAD_SEQ": 3
+  },
+  {
+    "CSRC_LEN": 0,
+    "EXTENSION": 0,
+    "PADDING": 0,
+    "VERSION": 2,
+    "PAYLOAD_TYPE": "PCMA",
+    "MARKER": 0,
+    "SEQ": 6,
+    "TIMESTAMP": 960,
+    "SSRC": 3535621694,
+    "PAYLOAD_LEN": 160,
+    "PAYLOAD_SEQ": 4
+  },
+  {
+    "CSRC_LEN": 0,
+    "EXTENSION": 0,
+    "PADDING": 0,
+    "VERSION": 2,
+    "PAYLOAD_TYPE": "PCMA",
+    "MARKER": 0,
+    "SEQ": 7,
+    "TIMESTAMP": 9440,
+    "SSRC": 3535621694,
+    "PAYLOAD_LEN": 160,
+    "PAYLOAD_SEQ": 5
+  },
+  {
+    "CSRC_LEN": 0,
+    "EXTENSION": 0,
+    "PADDING": 0,
+    "VERSION": 2,
+    "PAYLOAD_TYPE": "PCMA",
+    "MARKER": 0,
+    "SEQ": 8,
+    "TIMESTAMP": 9600,
+    "SSRC": 3535621694,
+    "PAYLOAD_LEN": 160,
+    "PAYLOAD_SEQ": 6
+  },
+  {
+    "CSRC_LEN": 0,
+    "EXTENSION": 0,
+    "PADDING": 0,
+    "VERSION": 2,
+    "PAYLOAD_TYPE": "PCMA",
+    "MARKER": 0,
+    "SEQ": 9,
+    "TIMESTAMP": 9760,
+    "SSRC": 3535621694,
+    "PAYLOAD_LEN": 160,
+    "PAYLOAD_SEQ": 7
+  },
+  {
+    "CSRC_LEN": 0,
+    "EXTENSION": 0,
+    "PADDING": 0,
+    "VERSION": 2,
+    "PAYLOAD_TYPE": "PCMA",
+    "MARKER": 0,
+    "SEQ": 10,
+    "TIMESTAMP": 9920,
+    "SSRC": 3535621694,
+    "PAYLOAD_LEN": 160,
+    "PAYLOAD_SEQ": 8
+  }
+]
diff --git a/test/decoders/rtp/rtp_test.cpp b/test/decoders/rtp/rtp_test.cpp
new file mode 100644
index 0000000..ddd4d80
--- /dev/null
+++ b/test/decoders/rtp/rtp_test.cpp
@@ -0,0 +1,325 @@
+#include <stdio.h>
+#include <stdlib.h>
+
+#include "stellar/stellar.h"
+#include "stellar/module_manager.h"
+#include "stellar/session.h"
+#include "stellar/utils.h"
+#include "stellar/rtp.h"
+
+#include <gtest/gtest.h>
+#include "cjson/cJSON.h"
+
+#define RTP_TEST_MODULE_NAME "RTP_TEST_MODULE"
+#define RTP_TEST_RESULT_EXPECT_ENV "RTP_TEST_RESULT_EXPECT"
+#define RTP_TEST_STELLAR_CONFIG_FILE "RTP_TEST_STELLAR_CONFIG"
+
+struct rtp_test_result {
+    cJSON *test_json;
+    cJSON *expect_json;
+    int count;
+};
+
+struct rtp_test_module_ctx {
+    struct rtp_test_result *result;
+
+    // depends a single session pcap
+    int rtp_count;
+};
+
+static void rtp_test_result_add(struct rtp_test_result *result, cJSON *json)
+{
+    cJSON_AddItemToArray(result->test_json, json);
+}
+
+static void rtp_test_result_compare(struct rtp_test_result *result)
+{
+    EXPECT_TRUE(result->expect_json != NULL && result->test_json != NULL);
+
+    int i;
+    int test_result_count, expect_result_count;
+    char *test_str, *expect_str;
+    cJSON *tmp_test, *tmp_expect;
+
+    /*
+        expect_str = cJSON_Print(result->expect_json);
+        test_str = cJSON_Print(result->test_json);
+        printf("LOAD Raw:\n%s\n", expect_str);
+        printf("TEST Raw:\n%s\n", test_str);
+    */
+
+    test_result_count = cJSON_GetArraySize(result->test_json);
+    expect_result_count = cJSON_GetArraySize(result->expect_json);
+
+    EXPECT_EQ(test_result_count, expect_result_count);
+
+    for (i = 0; i < MIN(test_result_count, expect_result_count); i++) {
+        tmp_test = cJSON_GetArrayItem(result->test_json, i);
+        tmp_expect = cJSON_GetArrayItem(result->expect_json, i);
+        expect_str = cJSON_Print(tmp_expect);
+        test_str = cJSON_Print(tmp_test);
+
+        EXPECT_STREQ(expect_str, test_str);
+        free(expect_str);
+        free(test_str);
+        break;
+    }
+}
+
+static void rtp_test_result_exit(struct rtp_test_result *result)
+{
+    if (result->expect_json) {
+        cJSON_Delete(result->expect_json);
+    }
+    if (result->test_json) {
+        cJSON_Delete(result->test_json);
+    }
+}
+
+static struct rtp_test_result * rtp_test_result_init(void)
+{
+    long filesize;
+    char *buffer;
+    FILE *file;
+    struct rtp_test_result *result;
+
+    result = (struct rtp_test_result *)calloc(1, sizeof(struct rtp_test_result));
+
+    file = fopen(getenv(RTP_TEST_RESULT_EXPECT_ENV), "rb");
+    if (file) {
+        fseek(file, 0, SEEK_END);
+        filesize = ftell(file);
+        rewind(file);
+        buffer = (char *)calloc(filesize + 1, 1);
+        fread(buffer, 1, filesize, file);
+
+        result->expect_json = cJSON_Parse(buffer);
+
+        free(buffer);
+        fclose(file);
+    }
+
+    result->test_json = cJSON_CreateArray();
+
+    return result;
+}
+
+static const char* rtp_test_rtp_payload_to_string(unsigned char payload_type)
+{
+    switch (payload_type) {
+        case PT_PCMU:
+            return "PCMU";
+        case PT_1016:
+            return "1016";
+        case PT_G721:
+            return "G721";
+        case PT_GSM:
+            return "GSM";
+        case PT_G723:
+            return "G723";
+        case PT_DVI4_8000:
+            return "DVI4_8000";
+        case PT_DVI4_16000:
+            return "DVI4_16000";
+        case PT_LPC:
+            return "LPC";
+        case PT_PCMA:
+            return "PCMA";
+        case PT_G722:
+            return "G722";
+        case PT_L16_STEREO:
+            return "L16_STEREO";
+        case PT_L16_MONO:
+            return "L16_MONO";
+        case PT_QCELP:
+            return "QCELP";
+        case PT_CN:
+            return "CN";
+        case PT_MPA:
+            return "MPA";
+        case PT_G728:
+            return "G728";
+        case PT_DVI4_11025:
+            return "DVI4_11025";
+        case PT_DVI4_22050:
+            return "DVI4_22050";
+        case PT_G729:
+            return "G729";
+        case PT_CN_OLD:
+            return "CN_OLD";
+        case PT_CELB:
+            return "CELB";
+        case PT_JPEG:
+            return "JPEG";
+        case PT_NV:
+            return "NV";
+        case PT_H261:
+            return "H261";
+        case PT_MPV:
+            return "MPV";
+        case PT_MP2T:
+            return "MP2T";
+        case PT_H263:
+            return "H263";
+        case PT_UNDF_96:
+            return "UNDF_96";
+        case PT_UNDF_97:
+            return "UNDF_97";
+        case PT_UNDF_98:
+            return "UNDF_98";
+        case PT_UNDF_99:
+            return "UNDF_99";
+        case PT_UNDF_100:
+            return "UNDF_100";
+        case PT_UNDF_101:
+            return "UNDF_101";
+        case PT_UNDF_102:
+            return "UNDF_102";
+        case PT_UNDF_103:
+            return "UNDF_103";
+        case PT_UNDF_104:
+            return "UNDF_104";
+        case PT_UNDF_105:
+            return "UNDF_105";
+        case PT_UNDF_106:
+            return "UNDF_106";
+        case PT_UNDF_107:
+            return "UNDF_107";
+        case PT_UNDF_108:
+            return "UNDF_108";
+        case PT_UNDF_109:
+            return "UNDF_109";
+        case PT_UNDF_110:
+            return "UNDF_110";
+        case PT_UNDF_111:
+            return "UNDF_111";
+        case PT_UNDF_112:
+            return "UNDF_112";
+        case PT_UNDF_113:
+            return "UNDF_113";
+        case PT_UNDF_114:
+            return "UNDF_114";
+        case PT_UNDF_115:
+            return "UNDF_115";
+        case PT_UNDF_116:
+            return "UNDF_116";
+        case PT_UNDF_117:
+            return "UNDF_117";
+        case PT_UNDF_118:
+            return "UNDF_118";
+        case PT_UNDF_119:
+            return "UNDF_119";
+        case PT_UNDF_120:
+            return "UNDF_120";
+        case PT_UNDF_121:
+            return "UNDF_121";
+        case PT_UNDF_122:
+            return "UNDF_122";
+        case PT_UNDF_123:
+            return "UNDF_123";
+        case PT_UNDF_124:
+            return "UNDF_124";
+        case PT_UNDF_125:
+            return "UNDF_125";
+        case PT_UNDF_126:
+            return "UNDF_126";
+        case PT_UNDF_127:
+            return "UNDF_127";
+        default:
+            return "UNKNOWN";
+    }
+    return NULL;
+}
+
+static void rtp_test_rtp_callback(struct session *sess, const struct rtp_header *hdr, const char *payload, size_t payload_len, void *arg)
+{
+    (void)(sess);
+    (void)(payload);
+    struct rtp_test_module_ctx *mod_ctx = (struct rtp_test_module_ctx *)arg;
+
+    cJSON *json = cJSON_CreateObject();
+
+    cJSON_AddNumberToObject(json, "CSRC_LEN", hdr->csrc_len);
+    cJSON_AddNumberToObject(json, "EXTENSION", hdr->extension);
+    cJSON_AddNumberToObject(json, "PADDING", hdr->padding);
+    cJSON_AddNumberToObject(json, "VERSION", hdr->version);
+    cJSON_AddStringToObject(json, "PAYLOAD_TYPE", rtp_test_rtp_payload_to_string(hdr->payload_type));
+    cJSON_AddNumberToObject(json, "MARKER", hdr->marker);
+    cJSON_AddNumberToObject(json, "SEQ", ntohs(hdr->seq));
+    cJSON_AddNumberToObject(json, "TIMESTAMP", ntohl(hdr->timestamp));
+    cJSON_AddNumberToObject(json, "SSRC", ntohl(hdr->ssrc));
+
+    cJSON_AddNumberToObject(json, "PAYLOAD_LEN", payload_len);
+    cJSON_AddNumberToObject(json, "PAYLOAD_SEQ", mod_ctx->rtp_count++);
+
+    rtp_test_result_add(mod_ctx->result, json);
+    return;
+}
+
+extern "C" void rtp_test_exit(struct stellar_module_manager *mod_mgr, struct stellar_module *mod)
+{
+    struct rtp_test_module_ctx *mod_ctx;
+
+    if (mod_mgr && mod) {
+        mod_ctx = (struct rtp_test_module_ctx *)stellar_module_get_ctx(mod);
+        if (mod_ctx) {
+            rtp_test_result_compare(mod_ctx->result);
+            rtp_test_result_exit(mod_ctx->result);
+            free(mod_ctx);
+        }
+        stellar_module_free(mod);
+    }
+}
+
+extern "C" struct stellar_module *rtp_test_init(struct stellar_module_manager *mod_mgr)
+{
+    int ret;
+    struct stellar_module *mod;
+    struct rtp_test_module_ctx *mod_ctx;
+    struct session_manager *sess_mgr;
+
+    mod_ctx = (struct rtp_test_module_ctx *)calloc(1, sizeof(struct rtp_test_module_ctx));
+    mod = stellar_module_new(RTP_TEST_MODULE_NAME, mod_ctx);
+    sess_mgr = stellar_module_get_session_manager(mod_mgr);
+
+    if (mod_mgr == NULL || sess_mgr == NULL) {
+        goto exit;
+    }
+
+    mod_ctx->result = rtp_test_result_init();
+    if (mod_ctx->result == NULL) {
+        goto exit;
+    }
+
+    ret = rtp_subscribe(mod_mgr, rtp_test_rtp_callback, mod_ctx);
+    if (ret < 0) {
+        goto exit;
+    }
+
+    return mod;
+exit:
+    printf("rtp_test module init failed!\n");
+    rtp_test_exit(mod_mgr, mod);
+    return NULL;
+}
+
+TEST(rtp, rtp_module)
+{
+    struct stellar *st;
+
+    setenv(RTP_TEST_RESULT_EXPECT_ENV, "./rtp_result.json", 0);
+
+    st = stellar_new(getenv(RTP_TEST_STELLAR_CONFIG_FILE));
+
+    stellar_run(st);
+    stellar_free(st);
+}
+
+int main(int argc, char ** argv)
+{
+    setenv(RTP_TEST_STELLAR_CONFIG_FILE, "./stellar.toml", 0);
+
+    ::testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}
+