Suricata-7.0 源码分析之流表建立FlowWorker

一、什么是Flow？
二、Flow是怎么建立的？
三、Flow建立的具体过程是什么？

一、什么是Flow？
??在Suricata 7.0中，流Flow是指所有相同五元组（协议，源IP，目的IP，源端口，目的端口）的数据包Packet分组，它们都属于同一流Flow，属于流Flow的数据包Packet在内部连接到它。
??流Flow建立是在FlowWorker（）函数中，在解码Decode模块中调用函数FlowSetupPacket（），基于五元组（协议，源IP，目的IP，源端口，目的端口）生成流Flow的hash值（p->flow_hash），并且设置标志位PKT_WANTS_FLOW，表示这个Packet数据包是由网络收包经过解码Decode模块而来的Packet数据包。

/*
*这段代码是一个函数，用于在处理网络数据包时设置相关的流处理标记和流哈希值。

1. `p->flags |= PKT_WANTS_FLOW;`：将数据包的标志中加入了 `PKT_WANTS_FLOW` 标记。这个标记通常用于表示当前数据包需要进行流处理，即需要为该数据包创建或更新对应的流对象。

2. `p->flow_hash = FlowGetHash(p);`：调用 `FlowGetHash` 函数来计算当前数据包的流哈希值，并将结果赋给数据包的 `flow_hash` 属性。流哈希值通常用于在流表中快速查找或定位对应的流对象。

总的来说，这段代码的作用是在处理网络数据包时，标记该数据包需要进行流处理，并计算并保存该数据包对应的流哈希值，以便后续创建或更新对应的流对象。
*/
void FlowSetupPacket(Packet *p)
{
    p->flags |= PKT_WANTS_FLOW;
    p->flow_hash = FlowGetHash(p);
}。

Flow结构体如下：

/**
 *  \brief Flow data structure.
 *
 *  The flow is a global data structure that is created for new packets of a
 *  flow and then looked up for the following packets of a flow.
 *
 *  Locking
 *
 *  The flow is updated/used by multiple packets at the same time. This is why
 *  there is a flow-mutex. It's a mutex and not a spinlock because some
 *  operations on the flow can be quite expensive, thus spinning would be
 *  too expensive.
 *
 *  The flow "header" (addresses, ports, proto, recursion level) are static
 *  after the initialization and remain read-only throughout the entire live
 *  of a flow. This is why we can access those without protection of the lock.
 * 
 Flow的结构体，用于表示流数据的结构。这个结构体的各个成员的含义：

这个结构体定义了用于表示网络数据流的各种属性和状态信息，包括地址、端口、协议类型、超时信息、线程信息、锁信息、协议特定数据指针等。
 */

typedef struct Flow_
{
    /* flow "header", used for hashing and flow lookup. Static after init,
     * so safe to look at without lock */
    FlowAddress src, dst; /*表示流的源地址和目的地址。*/

/* sp,dp:表示源端口和目的端口。*/
    union {
        Port sp;        /**< tcp/udp source port */
        struct {
            uint8_t type;   /**< icmp type */
            uint8_t code;   /**< icmp code */
        } icmp_s;

        struct {
            uint32_t spi; /**< esp spi */
        } esp;
    };
    union {
        Port dp;        /**< tcp/udp destination port */
        struct {
            uint8_t type;   /**< icmp type */
            uint8_t code;   /**< icmp code */
        } icmp_d;
    };
    uint8_t proto; /*表示协议类型。*/
    uint8_t recursion_level;/*表示递归级别。 */
    uint16_t vlan_id[VLAN_MAX_LAYERS]; /* 表示 VLAN ID 和 VLAN 索引。*/

    uint8_t vlan_idx;

    /* track toserver/toclient flow timeout needs 
    ffr_ts、ffr_tc、ffr
用于跟踪流的超时需求。
*/
    union {
        struct {
            uint8_t ffr_ts:4;
            uint8_t ffr_tc:4;
        };
        uint8_t ffr;
    };

    /** timestamp in seconds of the moment this flow will timeout
     *  according to the timeout policy. Does *not* take emergency
     *  mode into account.
timeout_at:表示流的超时时间戳。
 */
    uint32_t timeout_at; 

    /** Thread ID for the stream/detect portion of this flow
thread_id:用于存储流的线程 ID。
*/
    FlowThreadId thread_id[2];

    struct Flow_ *next; /* (hash) list next 指向下一个流的指针。 */
    /** Incoming interface 
livedev:表示流的输入接口。
*/
    struct LiveDevice_ *livedev;

    /** flow hash - the flow hash before hash table size mod.
flow_hash：表示流的哈希值。
 */
    uint32_t flow_hash;

    /** timeout policy value in seconds to add to the lastts.tv_sec
     *  when a packet has been received.
timeout_policy：表示超时策略的数值。
*/
    uint32_t timeout_policy;

    /* time stamp of last update (last packet). Set/updated under the
     * flow and flow hash row locks, safe to read under either the
     * flow lock or flow hash row lock. 
lastts：表示流的最后更新时间戳。
*/
    SCTime_t lastts;

/* flow_state：表示流的状态类型。*/
    FlowStateType flow_state;

    /** flow tenant id, used to setup flow timeout and stream pseudo
     *  packets with the correct tenant id set 
tenant_id：表示租户 ID。
*/
    uint32_t tenant_id;

	/*
 probing_parser_toserver_alproto_masks、probing_parser_toclient_alproto_masks：用于探测解析器的掩码。
	*/
    uint32_t probing_parser_toserver_alproto_masks;
    uint32_t probing_parser_toclient_alproto_masks;

    uint32_t flags;         /**< generic flags flags：表示流的通用标志。*/

    uint16_t file_flags;    /**< file tracking/extraction flags file_flags：表示文件跟踪/提取标志。*/

    /** destination port to be used in protocol detection. This is meant
     *  for use with STARTTLS and HTTP CONNECT detection 
protodetect_dp：用于协议检测的目的端口。
	*/
    uint16_t protodetect_dp; /**< 0 if not used */

    /* Parent flow id for protocol like ftp 
	parent_id：表示协议的父流 ID。
	*/
    int64_t parent_id;

/*
r、m：用于流锁的读写锁或互斥锁。
*/
#ifdef FLOWLOCK_RWLOCK
    SCRWLock r;
#elif defined FLOWLOCK_MUTEX
    SCMutex m;
#else
    #error Enable FLOWLOCK_RWLOCK or FLOWLOCK_MUTEX
#endif

    /** protocol specific data pointer, e.g. for TcpSession 
	protoctx：指向协议特定数据的指针。
	*/
    void *protoctx;

    /** mapping to Flow's protocol specific protocols for timeouts
        and state and free functions. 
	protomap：用于超时和状态函数的协议映射。
	*/
    uint8_t protomap;

	/* flow_end_flags：流结束标志。*/
    uint8_t flow_end_flags;
    /* coccinelle: Flow:flow_end_flags:FLOW_END_FLAG_ */

	/*alproto、alproto_ts、alproto_tc、alproto_orig、alproto_expect：表示应用层协议类型。*/
    AppProto alproto; /**< \brief application level protocol */
    AppProto alproto_ts;
    AppProto alproto_tc;

    /** original application level protocol. Used to indicate the previous
       protocol when changing to another protocol , e.g. with STARTTLS. 
	 alproto_orig：这是一个枚举类型的变量，用于表示原始的应用层协议。当协议发生更改时，例如通过使用STARTTLS进行加密通信，该变量用于指示之前使用的协议。

        alproto_expect：这是一个枚举类型的变量，用于表示预期的应用层协议。它在处理协议更改或升级的情况下很有用，例如在使用STARTTLS进行加密通信后，它可以指定在更改或升级后预期使用的协议。
	*/
    AppProto alproto_orig;
    /** expected app protocol: used in protocol change/upgrade like in
     *  STARTTLS. */
    AppProto alproto_expect;

    /** detection engine ctx version used to inspect this flow. Set at initial
     *  inspection. If it doesn't match the currently in use de_ctx, the
     *  stored sgh ptrs are reset.
	de_ctx_version：表示检测引擎上下文的版本。
	*/
    uint32_t de_ctx_version;

    /** ttl tracking 
	min_ttl_toserver、max_ttl_toserver、min_ttl_toclient、		 max_ttl_toclient：用于跟踪 TTL（生存时间）。
	*/
    uint8_t min_ttl_toserver;
    uint8_t max_ttl_toserver;
    uint8_t min_ttl_toclient;
    uint8_t max_ttl_toclient;

    /** application level storage ptrs.
     *alparser、alstate：用于应用层解析器的内部状态和应用层状态的指针。
     */
    AppLayerParserState *alparser;     /**< parser internal state */
    void *alstate;      /**< application layer state */

    /** toclient sgh for this flow. Only use when FLOW_SGH_TOCLIENT flow flag
     *  has been set. 
	sgh_toclient、sgh_toserver：指向流的客户端和服务器端的 SigGroupHead 结构的指针。
	*/
    const struct SigGroupHead_ *sgh_toclient;
    
    /** toserver sgh for this flow. Only use when FLOW_SGH_TOSERVER flow flag
     *  has been set. */
    const struct SigGroupHead_ *sgh_toserver;

    /* pointer to the var list 
	flowvar：指向变量列表的指针。
	*/
    GenericVar *flowvar;

	/* fb：指向流桶的指针。*/
    struct FlowBucket_ *fb;

	/* startts：表示流的开始时间戳。*/
    SCTime_t startts;

	/*todstpktcnt、tosrcpktcnt、todstbytecnt、tosrcbytecnt：用于跟踪数据包和字节计数。*/
    uint32_t todstpktcnt;
    uint32_t tosrcpktcnt;
    uint64_t todstbytecnt;
    uint64_t tosrcbytecnt;
} Flow;

二、Flow是怎么建立的？
??主要流程，如下所示：
??FlowWorker
????-->FlowHandlePacket
??????-->FlowGetFlowFromHash
????????-->FlowGetNew（新建流）
??????????-->FlowQueuePrivateGetFromTop（从flow_spare_pool中申请流）
??????????-->FlowAlloc（flow_spare_pool不够，直接alloc申请）
?????????-->MoveToWorkQueue（已有流超时的，从哈希桶中删除，并放入work_queue或者evicted链表）

????????-->FlowWorkerProcessInjectedFlows（取出flow_queue中的flow放入到work_queue）
????????-->FlowWorkerProcessLocalFlows
??????????-->CheckWorkQueue
????????????-->FlowClearMemory（清除流信息）
????????????-->FlowSparePoolReturnFlow（将流归还到flow_spare_pool中）

??Flow建立详细过程，如下图所示：