自学教程：C++ FLOWLOCK_WRLOCK函数代码示例

/** * /brief Search tags for src and dst. Update entries of the tag, remove if necessary * * /param de_ctx Detect context * /param det_ctx Detect thread context * /param p packet * */void TagHandlePacket(DetectEngineCtx *de_ctx,        DetectEngineThreadCtx *det_ctx, Packet *p){    /* If there's no tag, get out of here */    unsigned int current_tags = SC_ATOMIC_GET(num_tags);    if (current_tags == 0)        return;    /* First update and get session tags */    if (p->flow != NULL) {        FLOWLOCK_WRLOCK(p->flow);        TagHandlePacketFlow(p->flow, p);        FLOWLOCK_UNLOCK(p->flow);    }    Host *src = HostLookupHostFromHash(&p->src);    if (src) {        if (src->tag != NULL) {            TagHandlePacketHost(src,p);        }        HostRelease(src);    }    Host *dst = HostLookupHostFromHash(&p->dst);    if (dst) {        if (dst->tag != NULL) {            TagHandlePacketHost(dst,p);        }        HostRelease(dst);    }}

static TmEcode OutputStreamingLog(ThreadVars *tv, Packet *p, void *thread_data, PacketQueue *pq, PacketQueue *postpq){    BUG_ON(thread_data == NULL);    BUG_ON(list == NULL);    OutputLoggerThreadData *op_thread_data = (OutputLoggerThreadData *)thread_data;    OutputStreamingLogger *logger = list;    OutputLoggerThreadStore *store = op_thread_data->store;    StreamerCallbackData streamer_cbdata = { logger, store, tv, p , 0};    BUG_ON(logger == NULL && store != NULL);    BUG_ON(logger != NULL && store == NULL);    BUG_ON(logger == NULL && store == NULL);    uint8_t flags = 0;    Flow * const f = p->flow;    /* no flow, no streaming */    if (f == NULL) {        SCReturnInt(TM_ECODE_OK);    }    if (p->flowflags & FLOW_PKT_TOCLIENT)        flags |= OUTPUT_STREAMING_FLAG_TOCLIENT;    else        flags |= OUTPUT_STREAMING_FLAG_TOSERVER;    FLOWLOCK_WRLOCK(f);    if (op_thread_data->loggers & (1<<STREAMING_TCP_DATA)) {        TcpSession *ssn = f->protoctx;        if (ssn) {            int close = (ssn->state >= TCP_CLOSED);            close |= ((p->flags & PKT_PSEUDO_STREAM_END) ? 1 : 0);            SCLogDebug("close ? %s", close ? "yes" : "no");            TcpStream *stream = flags & OUTPUT_STREAMING_FLAG_TOSERVER ? &ssn->client : &ssn->server;            streamer_cbdata.type = STREAMING_TCP_DATA;            StreamIterator(p->flow, stream, close, (void *)&streamer_cbdata, flags);        }    }    if (op_thread_data->loggers & (1<<STREAMING_HTTP_BODIES)) {        if (f->alproto == ALPROTO_HTTP && f->alstate != NULL) {            int close = 0;            TcpSession *ssn = f->protoctx;            if (ssn) {                close = (ssn->state >= TCP_CLOSED);                close |= ((p->flags & PKT_PSEUDO_STREAM_END) ? 1 : 0);            }            SCLogDebug("close ? %s", close ? "yes" : "no");            streamer_cbdata.type = STREAMING_HTTP_BODIES;            HttpBodyIterator(f, close, (void *)&streamer_cbdata, flags);        }    }    FLOWLOCK_UNLOCK(f);    return TM_ECODE_OK;}

static int JsonTlsLogger(ThreadVars *tv, void *thread_data, const Packet *p) {    JsonTlsLogThread *aft = (JsonTlsLogThread *)thread_data;    MemBuffer *buffer = (MemBuffer *)aft->buffer;    OutputTlsCtx *tls_ctx = aft->tlslog_ctx;    if (unlikely(p->flow == NULL)) {        return 0;    }    /* check if we have TLS state or not */    FLOWLOCK_WRLOCK(p->flow);    uint16_t proto = FlowGetAppProtocol(p->flow);    if (proto != ALPROTO_TLS)        goto end;    SSLState *ssl_state = (SSLState *)FlowGetAppState(p->flow);    if (unlikely(ssl_state == NULL)) {        goto end;    }    if (ssl_state->server_connp.cert0_issuerdn == NULL || ssl_state->server_connp.cert0_subject == NULL)        goto end;    json_t *js = CreateJSONHeader((Packet *)p, 0, "tls");//TODO    if (unlikely(js == NULL))        goto end;    json_t *tjs = json_object();    if (tjs == NULL) {        free(js);        goto end;    }    /* reset */    MemBufferReset(buffer);    /* tls.subject */    json_object_set_new(tjs, "subject",                        json_string(ssl_state->server_connp.cert0_subject));    /* tls.issuerdn */    json_object_set_new(tjs, "issuerdn",                        json_string(ssl_state->server_connp.cert0_issuerdn));    if (tls_ctx->flags & LOG_TLS_EXTENDED) {        LogTlsLogExtendedJSON(tjs, ssl_state);    }    json_object_set_new(js, "tls", tjs);    OutputJSONBuffer(js, tls_ctx->file_ctx, buffer);    json_object_clear(js);    json_decref(js);    /* we only log the state once */    ssl_state->flags |= SSL_AL_FLAG_STATE_LOGGED;end:    FLOWLOCK_UNLOCK(p->flow);    return 0;}

/** *  /brief Get a new flow * *  Get a new flow. We're checking memcap first and will try to make room *  if the memcap is reached. * *  /param tv thread vars *  /param dtv decode thread vars (for flow log api thread data) * *  /retval f *LOCKED* flow on succes, NULL on error. */static Flow *FlowGetNew(ThreadVars *tv, DecodeThreadVars *dtv, const Packet *p){    Flow *f = NULL;    if (FlowCreateCheck(p) == 0) {        return NULL;    }    /* get a flow from the spare queue */    f = FlowDequeue(&flow_spare_q);    if (f == NULL) {        /* If we reached the max memcap, we get a used flow */        if (!(FLOW_CHECK_MEMCAP(sizeof(Flow) + FlowStorageSize()))) {            /* declare state of emergency */            if (!(SC_ATOMIC_GET(flow_flags) & FLOW_EMERGENCY)) {                SC_ATOMIC_OR(flow_flags, FLOW_EMERGENCY);                FlowTimeoutsEmergency();                /* under high load, waking up the flow mgr each time leads                 * to high cpu usage. Flows are not timed out much faster if                 * we check a 1000 times a second. */                FlowWakeupFlowManagerThread();            }            f = FlowGetUsedFlow(tv, dtv);            if (f == NULL) {                /* max memcap reached, so increments the counter */                if (tv != NULL && dtv != NULL) {                    StatsIncr(tv, dtv->counter_flow_memcap);                }                /* very rare, but we can fail. Just giving up */                return NULL;            }            /* freed a flow, but it's unlocked */        } else {            /* now see if we can alloc a new flow */            f = FlowAlloc();            if (f == NULL) {                if (tv != NULL && dtv != NULL) {                    StatsIncr(tv, dtv->counter_flow_memcap);                }                return NULL;            }            /* flow is initialized but *unlocked* */        }    } else {        /* flow has been recycled before it went into the spare queue */        /* flow is initialized (recylced) but *unlocked* */    }    FLOWLOCK_WRLOCK(f);    FlowUpdateCounter(tv, dtv, p->proto);    return f;}

static TmEcode LogFileLogWrap(ThreadVars *tv, Packet *p, void *data, PacketQueue *pq, PacketQueue *postpq, int ipver){    SCEnter();    LogFileLogThread *aft = (LogFileLogThread *)data;    uint8_t flags = 0;    /* no flow, no htp state */    if (p->flow == NULL) {        SCReturnInt(TM_ECODE_OK);    }    if (p->flowflags & FLOW_PKT_TOCLIENT)        flags |= STREAM_TOCLIENT;    else        flags |= STREAM_TOSERVER;    int file_close = (p->flags & PKT_PSEUDO_STREAM_END) ? 1 : 0;    int file_trunc = 0;    FLOWLOCK_WRLOCK(p->flow);    file_trunc = StreamTcpReassembleDepthReached(p);    FileContainer *ffc = AppLayerParserGetFiles(IPPROTO_TCP, p->flow->alproto,                                                p->flow->alstate, flags);    SCLogDebug("ffc %p", ffc);    if (ffc != NULL) {        File *ff;        for (ff = ffc->head; ff != NULL; ff = ff->next) {            if (ff->flags & FILE_LOGGED)                continue;            if (FileForceMagic() && ff->magic == NULL) {                FilemagicGlobalLookup(ff);            }            SCLogDebug("ff %p", ff);            if (file_trunc && ff->state < FILE_STATE_CLOSED)                ff->state = FILE_STATE_TRUNCATED;            if (ff->state == FILE_STATE_CLOSED ||                    ff->state == FILE_STATE_TRUNCATED || ff->state == FILE_STATE_ERROR ||                    (file_close == 1 && ff->state < FILE_STATE_CLOSED))            {                LogFileWriteJsonRecord(aft, p, ff, ipver);                ff->flags |= FILE_LOGGED;                aft->file_cnt++;            }        }        FilePrune(ffc);    }    FLOWLOCK_UNLOCK(p->flow);    SCReturnInt(TM_ECODE_OK);}

void FlowAlertSidSet(Flow *f, uint32_t sid) {    FLOWLOCK_WRLOCK(f);    FlowAlertSid *fb = FlowAlertSidGet(f, sid);    if (fb == NULL) {        FlowAlertSidAdd(f, sid);    }    FLOWLOCK_UNLOCK(f);}

void FlowBitUnset(Flow *f, uint16_t idx) {    FLOWLOCK_WRLOCK(f);    FlowBit *fb = FlowBitGet(f, idx);    if (fb != NULL) {        FlowBitRemove(f, idx);    }    FLOWLOCK_UNLOCK(f);}

void FlowAlertSidUnset(Flow *f, uint32_t sid) {    FLOWLOCK_WRLOCK(f);    FlowAlertSid *fb = FlowAlertSidGet(f, sid);    if (fb != NULL) {        FlowAlertSidRemove(f, sid);    }    FLOWLOCK_UNLOCK(f);}

void FlowBitSet(Flow *f, uint16_t idx) {    FLOWLOCK_WRLOCK(f);    FlowBit *fb = FlowBitGet(f, idx);    if (fb == NULL) {        FlowBitAdd(f, idx);    }    FLOWLOCK_UNLOCK(f);}

void FlowBitToggle(Flow *f, uint16_t idx) {    FLOWLOCK_WRLOCK(f);    FlowBit *fb = FlowBitGet(f, idx);    if (fb != NULL) {        FlowBitRemove(f, idx);    } else {        FlowBitAdd(f, idx);    }    FLOWLOCK_UNLOCK(f);}

void FlowAlertSidToggle(Flow *f, uint32_t sid) {    FLOWLOCK_WRLOCK(f);    FlowAlertSid *fb = FlowAlertSidGet(f, sid);    if (fb != NULL) {        FlowAlertSidRemove(f, sid);    } else {        FlowAlertSidAdd(f, sid);    }    FLOWLOCK_UNLOCK(f);}

/** * /test Test the deallocation of app layer parser memory on occurance of *       error in the parsing process. */static int AppLayerParserTest01(void){    AppLayerParserBackupParserTable();    int result = 0;    Flow *f = NULL;    uint8_t testbuf[] = { 0x11 };    uint32_t testlen = sizeof(testbuf);    TcpSession ssn;    AppLayerParserThreadCtx *alp_tctx = AppLayerParserThreadCtxAlloc();    memset(&ssn, 0, sizeof(ssn));    /* Register the Test protocol state and parser functions */    AppLayerParserRegisterParser(IPPROTO_TCP, ALPROTO_TEST, STREAM_TOSERVER,                      TestProtocolParser);    AppLayerParserRegisterStateFuncs(IPPROTO_TCP, ALPROTO_TEST,                          TestProtocolStateAlloc, TestProtocolStateFree);    f = UTHBuildFlow(AF_INET, "1.2.3.4", "4.3.2.1", 20, 40);    if (f == NULL)        goto end;    f->protoctx = &ssn;    f->alproto = ALPROTO_TEST;    f->proto = IPPROTO_TCP;    StreamTcpInitConfig(TRUE);    FLOWLOCK_WRLOCK(f);    int r = AppLayerParserParse(NULL, alp_tctx, f, ALPROTO_TEST,                                STREAM_TOSERVER | STREAM_EOF, testbuf,                                testlen);    if (r != -1) {        printf("returned %" PRId32 ", expected -1: ", r);        FLOWLOCK_UNLOCK(f);        goto end;    }    FLOWLOCK_UNLOCK(f);    if (!(ssn.flags & STREAMTCP_FLAG_APP_LAYER_DISABLED)) {        printf("flag should have been set, but is not: ");        goto end;    }    result = 1; end:    AppLayerParserRestoreParserTable();    StreamTcpFreeConfig(TRUE);    UTHFreeFlow(f);    return result;}

/** *  /brief post-match function for filestore * *  /param t thread local vars *  /param det_ctx pattern matcher thread local data *  /param p packet * *  The match function for filestore records store candidates in the det_ctx. *  When we are sure all parts of the signature matched, we run this function *  to finalize the filestore. */int DetectFilestorePostMatch(ThreadVars *t, DetectEngineThreadCtx *det_ctx, Packet *p, Signature *s){    uint8_t flags = 0;    SCEnter();    if (det_ctx->filestore_cnt == 0) {        SCReturnInt(0);    }    if (s->filestore_sm == NULL || p->flow == NULL) {#ifndef DEBUG        SCReturnInt(0);#else        BUG_ON(1);#endif    }    if (p->flowflags & FLOW_PKT_TOCLIENT)        flags |= STREAM_TOCLIENT;    else        flags |= STREAM_TOSERVER;    if (det_ctx->flow_locked == 0)        FLOWLOCK_WRLOCK(p->flow);    FileContainer *ffc = AppLayerParserGetFiles(p->flow->proto, p->flow->alproto,                                                p->flow->alstate, flags);    /* filestore for single files only */    if (s->filestore_sm->ctx == NULL) {        uint16_t u;        for (u = 0; u < det_ctx->filestore_cnt; u++) {            FileStoreFileById(ffc, det_ctx->filestore[u].file_id);        }    } else {        DetectFilestoreData *filestore = (DetectFilestoreData *)s->filestore_sm->ctx;        uint16_t u;        for (u = 0; u < det_ctx->filestore_cnt; u++) {            FilestorePostMatchWithOptions(p, p->flow, filestore, ffc,                    det_ctx->filestore[u].file_id, det_ctx->filestore[u].tx_id);        }    }    if (det_ctx->flow_locked == 0)        FLOWLOCK_UNLOCK(p->flow);    SCReturnInt(0);}

/** * /brief This function is used to add a tag to a session (type session) *        or update it if it's already installed. The number of times to *        allow an update is limited by DETECT_TAG_MATCH_LIMIT. This way *        repetitive matches to the same rule are limited of setting tags, *        to avoid DOS attacks * * /param p pointer to the current packet * /param tde pointer to the new DetectTagDataEntry * * /retval 0 if the tde was added succesfuly * /retval 1 if an entry of this sid/gid already exist and was updated */int TagFlowAdd(Packet *p, DetectTagDataEntry *tde) {    uint8_t updated = 0;    uint16_t num_tags = 0;    DetectTagDataEntry *iter = NULL;    if (p->flow == NULL)        return 1;    FLOWLOCK_WRLOCK(p->flow);    if (p->flow->tag_list != NULL) {        iter = p->flow->tag_list;        /* First iterate installed entries searching a duplicated sid/gid */        for (; iter != NULL; iter = iter->next) {            num_tags++;            if (iter->sid == tde->sid && iter->gid == tde->gid) {                iter->cnt_match++;                /* If so, update data, unless the maximum MATCH limit is                 * reached. This prevents possible DOS attacks */                if (iter->cnt_match < DETECT_TAG_MATCH_LIMIT) {                    /* Reset time and counters */                    iter->first_ts = iter->last_ts = tde->first_ts;                    iter->packets = 0;                    iter->bytes = 0;                }                updated = 1;                break;            }        }    }    /* If there was no entry of this rule, prepend the new tde */    if (updated == 0 && num_tags < DETECT_TAG_MAX_TAGS) {        DetectTagDataEntry *new_tde = DetectTagDataCopy(tde);        if (new_tde != NULL) {            new_tde->next = p->flow->tag_list;            p->flow->tag_list = new_tde;            (void) SC_ATOMIC_ADD(num_tags, 1);        }    } else if (num_tags == DETECT_TAG_MAX_TAGS) {        SCLogDebug("Max tags for sessions reached (%"PRIu16")", num_tags);    }    FLOWLOCK_UNLOCK(p->flow);    return updated;}

/** * /internal * /brief Forces reassembly for flows that need it. * * When this function is called we're running in virtually dead engine, * so locking the flows is not strictly required. The reasons it is still * done are: * - code consistency * - silence complaining profilers * - allow us to aggressively check using debug valdation assertions * - be robust in case of future changes * - locking overhead if neglectable when no other thread fights us * * /param q The queue to process flows from. */static inline void FlowForceReassemblyForHash(void){    Flow *f;    TcpSession *ssn;    int client_ok = 0;    int server_ok = 0;    uint32_t idx = 0;    for (idx = 0; idx < flow_config.hash_size; idx++) {        FlowBucket *fb = &flow_hash[idx];        PacketPoolWaitForN(9);        FBLOCK_LOCK(fb);        /* get the topmost flow from the QUEUE */        f = fb->head;        /* we need to loop through all the flows in the queue */        while (f != NULL) {            PacketPoolWaitForN(3);            FLOWLOCK_WRLOCK(f);            /* Get the tcp session for the flow */            ssn = (TcpSession *)f->protoctx;            /* /todo Also skip flows that shouldn't be inspected */            if (ssn == NULL) {                FLOWLOCK_UNLOCK(f);                f = f->hnext;                continue;            }            if (FlowForceReassemblyNeedReassembly(f, &server_ok, &client_ok) == 1) {                FlowForceReassemblyForFlow(f, server_ok, client_ok);            }            FLOWLOCK_UNLOCK(f);            /* next flow in the queue */            f = f->hnext;        }        FBLOCK_UNLOCK(fb);    }    return;}

static int DetectTlsStoreMatch (ThreadVars *t, DetectEngineThreadCtx *det_ctx, Flow *f, uint8_t flags, void *state, Signature *s, SigMatch *m){    SCEnter();    SSLState *ssl_state = (SSLState *)state;    if (ssl_state == NULL) {        SCLogDebug("no tls state, no match");        SCReturnInt(1);    }    FLOWLOCK_WRLOCK(f);    if (s->flags & SIG_FLAG_TLSSTORE) {        ssl_state->server_connp.cert_log_flag |= SSL_TLS_LOG_PEM;    }    FLOWLOCK_UNLOCK(f);    SCReturnInt(1);}

/** * /test Test the deallocation of app layer parser memory on occurance of *       error in the parsing process for UDP. */static int AppLayerParserTest02(void){    AppLayerParserBackupParserTable();    int result = 1;    Flow *f = NULL;    uint8_t testbuf[] = { 0x11 };    uint32_t testlen = sizeof(testbuf);    AppLayerParserThreadCtx *alp_tctx = AppLayerParserThreadCtxAlloc();    /* Register the Test protocol state and parser functions */    AppLayerParserRegisterParser(IPPROTO_UDP, ALPROTO_TEST, STREAM_TOSERVER,                      TestProtocolParser);    AppLayerParserRegisterStateFuncs(IPPROTO_UDP, ALPROTO_TEST,                          TestProtocolStateAlloc, TestProtocolStateFree);    f = UTHBuildFlow(AF_INET, "1.2.3.4", "4.3.2.1", 20, 40);    if (f == NULL)        goto end;    f->alproto = ALPROTO_TEST;    f->proto = IPPROTO_UDP;    f->protomap = FlowGetProtoMapping(f->proto);    StreamTcpInitConfig(TRUE);    FLOWLOCK_WRLOCK(f);    int r = AppLayerParserParse(NULL, alp_tctx, f, ALPROTO_TEST,                                STREAM_TOSERVER | STREAM_EOF, testbuf,                                testlen);    if (r != -1) {        printf("returned %" PRId32 ", expected -1: /n", r);        result = 0;        FLOWLOCK_UNLOCK(f);        goto end;    }    FLOWLOCK_UNLOCK(f); end:    AppLayerParserRestoreParserTable();    StreamTcpFreeConfig(TRUE);    UTHFreeFlow(f);    return result;}

/** /test Check the signature working to alert when http_stat_msg is used with *        negated content . */static int DetectHttpStatMsgSigTest03(void){    int result = 0;    Flow f;    uint8_t httpbuf1[] = "POST / HTTP/1.0/r/nUser-Agent: Mozilla/1.0/r/n/r/n";    uint32_t httplen1 = sizeof(httpbuf1) - 1; /* minus the /0 */    uint8_t httpbuf2[] = "HTTP/1.0 200 OK/r/n/r/n";    uint32_t httplen2 = sizeof(httpbuf2) - 1; /* minus the /0 */    TcpSession ssn;    Packet *p = NULL;    Signature *s = NULL;    ThreadVars th_v;    DetectEngineThreadCtx *det_ctx = NULL;    HtpState *http_state = NULL;    AppLayerParserThreadCtx *alp_tctx = AppLayerParserThreadCtxAlloc();    memset(&th_v, 0, sizeof(th_v));    memset(&f, 0, sizeof(f));    memset(&ssn, 0, sizeof(ssn));    p = UTHBuildPacket(NULL, 0, IPPROTO_TCP);    FLOW_INITIALIZE(&f);    f.protoctx = (void *)&ssn;    f.proto = IPPROTO_TCP;    f.flags |= FLOW_IPV4;    p->flow = &f;    p->flowflags |= FLOW_PKT_TOCLIENT;    p->flowflags |= FLOW_PKT_ESTABLISHED;    p->flags |= PKT_HAS_FLOW|PKT_STREAM_EST;    f.alproto = ALPROTO_HTTP;    StreamTcpInitConfig(TRUE);    DetectEngineCtx *de_ctx = DetectEngineCtxInit();    if (de_ctx == NULL) {        goto end;    }    de_ctx->flags |= DE_QUIET;    s = de_ctx->sig_list = SigInit(de_ctx,"alert http any any -> any any (msg:"                                   "/"HTTP status message/"; content:/"ok/"; "                                   "nocase; http_stat_msg; sid:1;)");    if (s == NULL) {        goto end;    }    s->next = SigInit(de_ctx,"alert http any any -> any any (msg:/"HTTP "                        "Status message nocase/"; content:!/"Not/"; "                        "http_stat_msg; sid:2;)");    if (s->next == NULL) {        goto end;    }    SigGroupBuild(de_ctx);    DetectEngineThreadCtxInit(&th_v, (void *)de_ctx, (void *)&det_ctx);    FLOWLOCK_WRLOCK(&f);    int r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_HTTP,                                STREAM_TOSERVER, httpbuf1, httplen1);    if (r != 0) {        printf("toserver chunk 1 returned %" PRId32 ", expected 0: ", r);        result = 0;        FLOWLOCK_UNLOCK(&f);        goto end;    }    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_HTTP,                            STREAM_TOCLIENT, httpbuf2, httplen2);    if (r != 0) {        printf("toclient chunk 1 returned %" PRId32 ", expected 0: ", r);        result = 0;        FLOWLOCK_UNLOCK(&f);        goto end;    }    FLOWLOCK_UNLOCK(&f);    http_state = f.alstate;    if (http_state == NULL) {        printf("no http state: ");        result = 0;        goto end;    }    /* do detect */    SigMatchSignatures(&th_v, de_ctx, det_ctx, p);    if (! PacketAlertCheck(p, 1)) {        printf("sid 1 didn't matched but should have: ");        goto end;    }    if (! PacketAlertCheck(p, 2)) {        printf("sid 2 didn't matched but should have: ");        goto end;    }//.........这里部分代码省略.........

static TmEcode FlowWorker(ThreadVars *tv, Packet *p, void *data, PacketQueue *preq, PacketQueue *unused){    FlowWorkerThreadData *fw = data;    void *detect_thread = SC_ATOMIC_GET(fw->detect_thread);    SCLogDebug("packet %"PRIu64, p->pcap_cnt);    /* update time */    if (!(PKT_IS_PSEUDOPKT(p))) {        TimeSetByThread(tv->id, &p->ts);    }    /* handle Flow */    if (p->flags & PKT_WANTS_FLOW) {        FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_FLOW);        FlowHandlePacket(tv, fw->dtv, p);        if (likely(p->flow != NULL)) {            DEBUG_ASSERT_FLOW_LOCKED(p->flow);            if (FlowUpdate(p) == TM_ECODE_DONE) {                FLOWLOCK_UNLOCK(p->flow);                return TM_ECODE_OK;            }        }        /* Flow is now LOCKED */        FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_FLOW);    /* if PKT_WANTS_FLOW is not set, but PKT_HAS_FLOW is, then this is a     * pseudo packet created by the flow manager. */    } else if (p->flags & PKT_HAS_FLOW) {        FLOWLOCK_WRLOCK(p->flow);    }    SCLogDebug("packet %"PRIu64" has flow? %s", p->pcap_cnt, p->flow ? "yes" : "no");    /* handle TCP and app layer */    if (p->flow && PKT_IS_TCP(p)) {        SCLogDebug("packet %"PRIu64" is TCP. Direction %s", p->pcap_cnt, PKT_IS_TOSERVER(p) ? "TOSERVER" : "TOCLIENT");        DEBUG_ASSERT_FLOW_LOCKED(p->flow);        /* if detect is disabled, we need to apply file flags to the flow         * here on the first packet. */        if (detect_thread == NULL &&                ((PKT_IS_TOSERVER(p) && (p->flowflags & FLOW_PKT_TOSERVER_FIRST)) ||                 (PKT_IS_TOCLIENT(p) && (p->flowflags & FLOW_PKT_TOCLIENT_FIRST))))        {            DisableDetectFlowFileFlags(p->flow);        }        FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_STREAM);        StreamTcp(tv, p, fw->stream_thread, &fw->pq, NULL);        FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_STREAM);        if (FlowChangeProto(p->flow)) {            StreamTcpDetectLogFlush(tv, fw->stream_thread, p->flow, p, &fw->pq);        }        /* Packets here can safely access p->flow as it's locked */        SCLogDebug("packet %"PRIu64": extra packets %u", p->pcap_cnt, fw->pq.len);        Packet *x;        while ((x = PacketDequeue(&fw->pq))) {            SCLogDebug("packet %"PRIu64" extra packet %p", p->pcap_cnt, x);            // TODO do we need to call StreamTcp on these pseudo packets or not?            //StreamTcp(tv, x, fw->stream_thread, &fw->pq, NULL);            if (detect_thread != NULL) {                FLOWWORKER_PROFILING_START(x, PROFILE_FLOWWORKER_DETECT);                Detect(tv, x, detect_thread, NULL, NULL);                FLOWWORKER_PROFILING_END(x, PROFILE_FLOWWORKER_DETECT);            }            //  Outputs            OutputLoggerLog(tv, x, fw->output_thread);            /* put these packets in the preq queue so that they are             * by the other thread modules before packet 'p'. */            PacketEnqueue(preq, x);        }    /* handle the app layer part of the UDP packet payload */    } else if (p->flow && p->proto == IPPROTO_UDP) {        FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_APPLAYERUDP);        AppLayerHandleUdp(tv, fw->stream_thread->ra_ctx->app_tctx, p, p->flow);        FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_APPLAYERUDP);    }    /* handle Detect */    DEBUG_ASSERT_FLOW_LOCKED(p->flow);    SCLogDebug("packet %"PRIu64" calling Detect", p->pcap_cnt);    if (detect_thread != NULL) {        FLOWWORKER_PROFILING_START(p, PROFILE_FLOWWORKER_DETECT);        Detect(tv, p, detect_thread, NULL, NULL);        FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_DETECT);    }    // Outputs.    OutputLoggerLog(tv, p, fw->output_thread);//.........这里部分代码省略.........

//.........这里部分代码省略.........    s = DetectEngineAppendSig(de_ctx, "alert tcp any any -> any any "                              "(msg:/"ssl state/"; ssl_state:client_hello; "                              "sid:1;)");    FAIL_IF_NULL(s);    s = DetectEngineAppendSig(de_ctx, "alert tcp any any -> any any "                              "(msg:/"ssl state/"; "                              "ssl_state:server_hello; "                              "sid:2;)");    FAIL_IF_NULL(s);    s = DetectEngineAppendSig(de_ctx, "alert tcp any any -> any any "                              "(msg:/"ssl state/"; "                              "ssl_state:client_keyx; "                              "sid:3;)");    FAIL_IF_NULL(s);    s = DetectEngineAppendSig(de_ctx, "alert tcp any any -> any any "                              "(msg:/"ssl state/"; "                              "ssl_state:server_keyx; "                              "sid:4;)");    FAIL_IF_NULL(s);    s = DetectEngineAppendSig(de_ctx, "alert tcp any any -> any any "                              "(msg:/"ssl state/"; "                              "ssl_state:!client_hello; "                              "sid:5;)");    FAIL_IF_NULL(s);    SigGroupBuild(de_ctx);    DetectEngineThreadCtxInit(&th_v, (void *)de_ctx, (void *)&det_ctx);    FLOWLOCK_WRLOCK(&f);    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_TLS,                            STREAM_TOSERVER | STREAM_START, chello_buf,                            chello_buf_len);    FAIL_IF(r != 0);    FLOWLOCK_UNLOCK(&f);    ssl_state = f.alstate;    FAIL_IF(ssl_state == NULL);    /* do detect */    p->alerts.cnt = 0;    SigMatchSignatures(&th_v, de_ctx, det_ctx, p);    FAIL_IF(!PacketAlertCheck(p, 1));    FAIL_IF(PacketAlertCheck(p, 2));    FAIL_IF(PacketAlertCheck(p, 3));    FAIL_IF(PacketAlertCheck(p, 4));    FAIL_IF(PacketAlertCheck(p, 5));    FLOWLOCK_WRLOCK(&f);    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_TLS, STREAM_TOCLIENT,                            shello_buf, shello_buf_len);    FAIL_IF(r != 0);    FLOWLOCK_UNLOCK(&f);    /* do detect */    p->alerts.cnt = 0;    p->flowflags = (FLOW_PKT_TOCLIENT | FLOW_PKT_ESTABLISHED);    SigMatchSignatures(&th_v, de_ctx, det_ctx, p);    FAIL_IF(PacketAlertCheck(p, 1));

static int JsonSshLogger(ThreadVars *tv, void *thread_data, const Packet *p){    JsonSshLogThread *aft = (JsonSshLogThread *)thread_data;    MemBuffer *buffer = (MemBuffer *)aft->buffer;    OutputSshCtx *ssh_ctx = aft->sshlog_ctx;    if (unlikely(p->flow == NULL)) {        return 0;    }    /* check if we have SSH state or not */    FLOWLOCK_WRLOCK(p->flow);    uint16_t proto = FlowGetAppProtocol(p->flow);    if (proto != ALPROTO_SSH)        goto end;    SshState *ssh_state = (SshState *)FlowGetAppState(p->flow);    if (unlikely(ssh_state == NULL)) {        goto end;    }    if (ssh_state->cli_hdr.software_version == NULL || ssh_state->srv_hdr.software_version == NULL)        goto end;    json_t *js = CreateJSONHeader((Packet *)p, 1, "ssh");//TODO    if (unlikely(js == NULL))        goto end;    json_t *tjs = json_object();    if (tjs == NULL) {        free(js);        goto end;    }    /* reset */    MemBufferReset(buffer);    json_t *cjs = json_object();    if (cjs != NULL) {        json_object_set_new(cjs, "proto_version",                json_string((char *)ssh_state->cli_hdr.proto_version));        json_object_set_new(cjs, "software_version",                json_string((char *)ssh_state->cli_hdr.software_version));    }    json_object_set_new(tjs, "client", cjs);    json_t *sjs = json_object();    if (sjs != NULL) {        json_object_set_new(sjs, "proto_version",                json_string((char *)ssh_state->srv_hdr.proto_version));        json_object_set_new(sjs, "software_version",                json_string((char *)ssh_state->srv_hdr.software_version));    }    json_object_set_new(tjs, "server", sjs);    json_object_set_new(js, "ssh", tjs);    OutputJSONBuffer(js, ssh_ctx->file_ctx, buffer);    json_object_clear(js);    json_decref(js);    /* we only log the state once */    ssh_state->cli_hdr.flags |= SSH_FLAG_STATE_LOGGED;end:    FLOWLOCK_UNLOCK(p->flow);    return 0;}

/** /test Send a get request in three chunks + more data. */static int DetectSshVersionTestDetect03(void){    int result = 0;    Flow f;    uint8_t sshbuf1[] = "SSH-1.";    uint32_t sshlen1 = sizeof(sshbuf1) - 1;    uint8_t sshbuf2[] = "7-PuTTY_2.123" ;    uint32_t sshlen2 = sizeof(sshbuf2) - 1;    uint8_t sshbuf3[] = "/n";    uint32_t sshlen3 = sizeof(sshbuf3) - 1;    uint8_t sshbuf4[] = "whatever...";    uint32_t sshlen4 = sizeof(sshbuf4) - 1;    TcpSession ssn;    Packet *p = NULL;    Signature *s = NULL;    ThreadVars th_v;    DetectEngineThreadCtx *det_ctx = NULL;    AppLayerParserThreadCtx *alp_tctx = AppLayerParserThreadCtxAlloc();    memset(&th_v, 0, sizeof(th_v));    memset(&f, 0, sizeof(f));    memset(&ssn, 0, sizeof(ssn));    p = UTHBuildPacket(NULL, 0, IPPROTO_TCP);    FLOW_INITIALIZE(&f);    f.protoctx = (void *)&ssn;    p->flow = &f;    p->flowflags |= FLOW_PKT_TOSERVER;    p->flowflags |= FLOW_PKT_ESTABLISHED;    p->flags |= PKT_HAS_FLOW|PKT_STREAM_EST;    f.alproto = ALPROTO_SSH;    f.proto = IPPROTO_TCP;    StreamTcpInitConfig(TRUE);    DetectEngineCtx *de_ctx = DetectEngineCtxInit();    if (de_ctx == NULL) {        goto end;    }    de_ctx->flags |= DE_QUIET;    s = de_ctx->sig_list = SigInit(de_ctx,"alert ssh any any -> any any (msg:/"SSH/"; ssh.protoversion:2_compat; sid:1;)");    if (s == NULL) {        goto end;    }    SigGroupBuild(de_ctx);    DetectEngineThreadCtxInit(&th_v, (void *)de_ctx, (void *)&det_ctx);    FLOWLOCK_WRLOCK(&f);    int r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_SSH,                                STREAM_TOSERVER, sshbuf1, sshlen1);    if (r != 0) {        printf("toserver chunk 1 returned %" PRId32 ", expected 0: ", r);        FLOWLOCK_UNLOCK(&f);        goto end;    }    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOSERVER,                            sshbuf2, sshlen2);    if (r != 0) {        printf("toserver chunk 2 returned %" PRId32 ", expected 0: ", r);        FLOWLOCK_UNLOCK(&f);        goto end;    }    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOSERVER,                            sshbuf3, sshlen3);    if (r != 0) {        printf("toserver chunk 3 returned %" PRId32 ", expected 0: ", r);        FLOWLOCK_UNLOCK(&f);        goto end;    }    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOSERVER,                            sshbuf4, sshlen4);    if (r != 0) {        printf("toserver chunk 4 returned %" PRId32 ", expected 0: ", r);        FLOWLOCK_UNLOCK(&f);        goto end;    }    FLOWLOCK_UNLOCK(&f);    SshState *ssh_state = f.alstate;    if (ssh_state == NULL) {        printf("no ssh state: ");        goto end;    }    /* do detect */    SigMatchSignatures(&th_v, de_ctx, det_ctx, p);    if (PacketAlertCheck(p, 1)) {        printf("Error, 1.7 version is not 2 compat, so the sig should not match: ");        goto end;    }//.........这里部分代码省略.........

static TmEcode OutputTxLog(ThreadVars *tv, Packet *p, void *thread_data, PacketQueue *pq, PacketQueue *postpq){    BUG_ON(thread_data == NULL);    BUG_ON(list == NULL);    OutputLoggerThreadData *op_thread_data = (OutputLoggerThreadData *)thread_data;    OutputTxLogger *logger = list;    OutputLoggerThreadStore *store = op_thread_data->store;    BUG_ON(logger == NULL && store != NULL);    BUG_ON(logger != NULL && store == NULL);    BUG_ON(logger == NULL && store == NULL);    if (p->flow == NULL)        return TM_ECODE_OK;    Flow * const f = p->flow;    FLOWLOCK_WRLOCK(f); /* WRITE lock before we updated flow logged id */    AppProto alproto = f->alproto;    if (AppLayerParserProtocolIsTxAware(p->proto, alproto) == 0)        goto end;    if (AppLayerParserProtocolHasLogger(p->proto, alproto) == 0)        goto end;    void *alstate = f->alstate;    if (alstate == NULL) {        SCLogDebug("no alstate");        goto end;    }    uint64_t total_txs = AppLayerParserGetTxCnt(p->proto, alproto, alstate);    uint64_t tx_id = AppLayerParserGetTransactionLogId(f->alparser);    for (; tx_id < total_txs; tx_id++)    {        int logger_not_logged = 0;        void *tx = AppLayerParserGetTx(p->proto, alproto, alstate, tx_id);        if (tx == NULL) {            SCLogDebug("tx is NULL not logging");            continue;        }        int tx_progress_ts = AppLayerParserGetStateProgress(p->proto, alproto,                tx, FlowGetDisruptionFlags(f, STREAM_TOSERVER));        int tx_progress_tc = AppLayerParserGetStateProgress(p->proto, alproto,                tx, FlowGetDisruptionFlags(f, STREAM_TOCLIENT));        // call each logger here (pseudo code)        logger = list;        store = op_thread_data->store;        while (logger && store) {            BUG_ON(logger->LogFunc == NULL);            SCLogDebug("logger %p", logger);            if (logger->alproto == alproto) {                SCLogDebug("alproto match, logging tx_id %ju", tx_id);                if (AppLayerParserGetTxLogged(p->proto, alproto, alstate, tx,                        logger->id)) {                    SCLogDebug("logger has already logged this transaction");                    goto next;                }                if (!(AppLayerParserStateIssetFlag(f->alparser,                                                   APP_LAYER_PARSER_EOF))) {                    if (logger->LogCondition) {                        int r = logger->LogCondition(tv, p, alstate, tx, tx_id);                        if (r == FALSE) {                            SCLogDebug("conditions not met, not logging");                            logger_not_logged = 1;                            goto next;                        }                    } else {                        if (tx_progress_tc < logger->tc_log_progress) {                            SCLogDebug("progress not far enough, not logging");                            logger_not_logged = 1;                            goto next;                        }                        if (tx_progress_ts < logger->ts_log_progress) {                            SCLogDebug("progress not far enough, not logging");                            logger_not_logged = 1;                            goto next;                        }                    }                }                PACKET_PROFILING_TMM_START(p, logger->module_id);                logger->LogFunc(tv, store->thread_data, p, f, alstate, tx, tx_id);                PACKET_PROFILING_TMM_END(p, logger->module_id);                AppLayerParserSetTxLogged(p->proto, alproto, alstate, tx,                                          logger->id);            }//.........这里部分代码省略.........

//.........这里部分代码省略.........        fb->tail = f;        /* got one, now lock, initialize and return */        FlowInit(f, p);        f->flow_hash = hash;        f->fb = fb;        FlowUpdateState(f, FLOW_STATE_NEW);        FlowReference(dest, f);        FBLOCK_UNLOCK(fb);        return f;    }    /* ok, we have a flow in the bucket. Let's find out if it is our flow */    f = fb->head;    /* see if this is the flow we are looking for */    if (FlowCompare(f, p) == 0) {        Flow *pf = NULL; /* previous flow */        while (f) {            pf = f;            f = f->hnext;            if (f == NULL) {                f = pf->hnext = FlowGetNew(tv, dtv, p);                if (f == NULL) {                    FBLOCK_UNLOCK(fb);                    return NULL;                }                fb->tail = f;                /* flow is locked */                f->hprev = pf;                /* initialize and return */                FlowInit(f, p);                f->flow_hash = hash;                f->fb = fb;                FlowUpdateState(f, FLOW_STATE_NEW);                FlowReference(dest, f);                FBLOCK_UNLOCK(fb);                return f;            }            if (FlowCompare(f, p) != 0) {                /* we found our flow, lets put it on top of the                 * hash list -- this rewards active flows */                if (f->hnext) {                    f->hnext->hprev = f->hprev;                }                if (f->hprev) {                    f->hprev->hnext = f->hnext;                }                if (f == fb->tail) {                    fb->tail = f->hprev;                }                f->hnext = fb->head;                f->hprev = NULL;                fb->head->hprev = f;                fb->head = f;                /* found our flow, lock & return */                FLOWLOCK_WRLOCK(f);                if (unlikely(TcpSessionPacketSsnReuse(p, f, f->protoctx) == 1)) {                    f = TcpReuseReplace(tv, dtv, fb, f, hash, p);                    if (f == NULL) {                        FBLOCK_UNLOCK(fb);                        return NULL;                    }                }                FlowReference(dest, f);                FBLOCK_UNLOCK(fb);                return f;            }        }    }    /* lock & return */    FLOWLOCK_WRLOCK(f);    if (unlikely(TcpSessionPacketSsnReuse(p, f, f->protoctx) == 1)) {        f = TcpReuseReplace(tv, dtv, fb, f, hash, p);        if (f == NULL) {            FBLOCK_UNLOCK(fb);            return NULL;        }    }    FlowReference(dest, f);    FBLOCK_UNLOCK(fb);    return f;}

//.........这里部分代码省略.........    /* get the key to our bucket */    uint32_t key = FlowGetKey(p);    /* get our hash bucket and lock it */    FlowBucket *fb = &flow_hash[key];    FBLOCK_LOCK(fb);    SCLogDebug("fb %p fb->head %p", fb, fb->head);    FlowHashCountIncr;    /* see if the bucket already has a flow */    if (fb->head == NULL) {        f = FlowGetNew(p);        if (f == NULL) {            FBLOCK_UNLOCK(fb);            FlowHashCountUpdate;            return NULL;        }        /* flow is locked */        fb->head = f;        fb->tail = f;        /* got one, now lock, initialize and return */        FlowInit(f, p);        f->fb = fb;        FBLOCK_UNLOCK(fb);        FlowHashCountUpdate;        return f;    }    /* ok, we have a flow in the bucket. Let's find out if it is our flow */    f = fb->head;    /* see if this is the flow we are looking for */    if (FlowCompare(f, p) == 0) {        Flow *pf = NULL; /* previous flow */        while (f) {            FlowHashCountIncr;            pf = f;            f = f->hnext;            if (f == NULL) {                f = pf->hnext = FlowGetNew(p);                if (f == NULL) {                    FBLOCK_UNLOCK(fb);                    FlowHashCountUpdate;                    return NULL;                }                fb->tail = f;                /* flow is locked */                f->hprev = pf;                /* initialize and return */                FlowInit(f, p);                f->fb = fb;                FBLOCK_UNLOCK(fb);                FlowHashCountUpdate;                return f;            }            if (FlowCompare(f, p) != 0) {                /* we found our flow, lets put it on top of the                 * hash list -- this rewards active flows */                if (f->hnext) {                    f->hnext->hprev = f->hprev;                }                if (f->hprev) {                    f->hprev->hnext = f->hnext;                }                if (f == fb->tail) {                    fb->tail = f->hprev;                }                f->hnext = fb->head;                f->hprev = NULL;                fb->head->hprev = f;                fb->head = f;                /* found our flow, lock & return */                FLOWLOCK_WRLOCK(f);                FBLOCK_UNLOCK(fb);                FlowHashCountUpdate;                return f;            }        }    }    /* lock & return */    FLOWLOCK_WRLOCK(f);    FBLOCK_UNLOCK(fb);    FlowHashCountUpdate;    return f;}

/** * /internal * /brief Forces reassembly for flows that need it. * * When this function is called we're running in virtually dead engine, * so locking the flows is not strictly required. The reasons it is still * done are: * - code consistency * - silence complaining profilers * - allow us to aggressively check using debug valdation assertions * - be robust in case of future changes * - locking overhead if neglectable when no other thread fights us * * /param q The queue to process flows from. */static inline void FlowForceReassemblyForHash(void){    Flow *f;    TcpSession *ssn;    int client_ok;    int server_ok;    int tcp_needs_inspection;    uint32_t idx = 0;    /* We use this packet just for reassembly purpose */    Packet *reassemble_p = PacketGetFromAlloc();    if (reassemble_p == NULL)        return;    for (idx = 0; idx < flow_config.hash_size; idx++) {        FlowBucket *fb = &flow_hash[idx];        FBLOCK_LOCK(fb);        /* get the topmost flow from the QUEUE */        f = fb->head;        /* we need to loop through all the flows in the queue */        while (f != NULL) {            PACKET_RECYCLE(reassemble_p);            FLOWLOCK_WRLOCK(f);            /* Get the tcp session for the flow */            ssn = (TcpSession *)f->protoctx;            /* /todo Also skip flows that shouldn't be inspected */            if (ssn == NULL) {                FLOWLOCK_UNLOCK(f);                f = f->hnext;                continue;            }            /* ah ah!  We have some unattended toserver segments */            if ((client_ok = StreamHasUnprocessedSegments(ssn, 0)) == 1) {                StreamTcpThread *stt = SC_ATOMIC_GET(stream_pseudo_pkt_stream_tm_slot->slot_data);                ssn->client.last_ack = (ssn->client.seg_list_tail->seq +                        ssn->client.seg_list_tail->payload_len);                FlowForceReassemblyPseudoPacketSetup(reassemble_p, 1, f, ssn, 1);                StreamTcpReassembleHandleSegment(stream_pseudo_pkt_stream_TV,                        stt->ra_ctx, ssn, &ssn->server,                        reassemble_p, NULL);                FlowDeReference(&reassemble_p->flow);                if (StreamTcpReassembleProcessAppLayer(stt->ra_ctx) < 0) {                    SCLogDebug("shutdown flow timeout "                               "StreamTcpReassembleProcessAppLayer() erroring "                               "over something");                }            }            /* oh oh!  We have some unattended toclient segments */            if ((server_ok = StreamHasUnprocessedSegments(ssn, 1)) == 1) {                StreamTcpThread *stt = SC_ATOMIC_GET(stream_pseudo_pkt_stream_tm_slot->slot_data);                ssn->server.last_ack = (ssn->server.seg_list_tail->seq +                        ssn->server.seg_list_tail->payload_len);                FlowForceReassemblyPseudoPacketSetup(reassemble_p, 0, f, ssn, 1);                StreamTcpReassembleHandleSegment(stream_pseudo_pkt_stream_TV,                        stt->ra_ctx, ssn, &ssn->client,                        reassemble_p, NULL);                FlowDeReference(&reassemble_p->flow);                if (StreamTcpReassembleProcessAppLayer(stt->ra_ctx) < 0) {                    SCLogDebug("shutdown flow timeout "                               "StreamTcpReassembleProcessAppLayer() erroring "                               "over something");                }            }            if (ssn->state >= TCP_ESTABLISHED && ssn->state != TCP_CLOSED)                tcp_needs_inspection = 1;            else                tcp_needs_inspection = 0;            FLOWLOCK_UNLOCK(f);            /* insert a pseudo packet in the toserver direction */            if (client_ok || tcp_needs_inspection)//.........这里部分代码省略.........

static int DetectSslVersionTestDetect02(void){    Flow f;    uint8_t sslbuf1[] = { 0x16 };    uint32_t ssllen1 = sizeof(sslbuf1);    uint8_t sslbuf2[] = { 0x03 };    uint32_t ssllen2 = sizeof(sslbuf2);    uint8_t sslbuf3[] = { 0x01 };    uint32_t ssllen3 = sizeof(sslbuf3);    uint8_t sslbuf4[] = { 0x01, 0x00, 0x00, 0xad, 0x03, 0x02 };    uint32_t ssllen4 = sizeof(sslbuf4);    TcpSession ssn;    Packet *p = NULL;    Signature *s = NULL;    ThreadVars th_v;    DetectEngineThreadCtx *det_ctx = NULL;    AppLayerParserThreadCtx *alp_tctx = AppLayerParserThreadCtxAlloc();    memset(&th_v, 0, sizeof(th_v));    memset(&f, 0, sizeof(f));    memset(&ssn, 0, sizeof(ssn));    p = UTHBuildPacket(NULL, 0, IPPROTO_TCP);    FLOW_INITIALIZE(&f);    f.protoctx = (void *)&ssn;    f.proto = IPPROTO_TCP;    p->flow = &f;    p->flowflags |= FLOW_PKT_TOSERVER;    p->flowflags |= FLOW_PKT_ESTABLISHED;    p->flags |= PKT_HAS_FLOW | PKT_STREAM_EST;    f.alproto = ALPROTO_TLS;    StreamTcpInitConfig(TRUE);    DetectEngineCtx *de_ctx = DetectEngineCtxInit();    FAIL_IF_NULL(de_ctx);    de_ctx->flags |= DE_QUIET;    s = de_ctx->sig_list = SigInit(de_ctx,"alert tls any any -> any any (msg:/"TLS/"; ssl_version:tls1.0; sid:1;)");    FAIL_IF_NULL(s);    SigGroupBuild(de_ctx);    DetectEngineThreadCtxInit(&th_v, (void *)de_ctx, (void *)&det_ctx);    FLOWLOCK_WRLOCK(&f);    int r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_TLS,                                STREAM_TOSERVER, sslbuf1, ssllen1);    FAIL_IF(r != 0);    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_TLS, STREAM_TOSERVER,                            sslbuf2, ssllen2);    FAIL_IF(r != 0);    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_TLS, STREAM_TOSERVER,                            sslbuf3, ssllen3);    FAIL_IF(r != 0);    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_TLS, STREAM_TOSERVER,                            sslbuf4, ssllen4);    FAIL_IF(r != 0);    FLOWLOCK_UNLOCK(&f);    SSLState *app_state = f.alstate;    FAIL_IF_NULL(app_state);    FAIL_IF(app_state->client_connp.content_type != 0x16);    FAIL_IF(app_state->client_connp.version != TLS_VERSION_10);    /* do detect */    SigMatchSignatures(&th_v, de_ctx, det_ctx, p);    FAIL_IF_NOT(PacketAlertCheck(p, 1));    AppLayerParserThreadCtxFree(alp_tctx);    DetectEngineThreadCtxDeinit(&th_v, (void *)det_ctx);    DetectEngineCtxFree(de_ctx);    StreamTcpFreeConfig(TRUE);    FLOW_DESTROY(&f);    UTHFreePackets(&p, 1);    PASS;}

//.........这里部分代码省略.........    p2 = UTHBuildPacketReal(server_hello, sizeof(server_hello), IPPROTO_TCP,                            "192.168.1.1", "192.168.1.5", 443, 51251);    p3 = UTHBuildPacketReal(certificate, sizeof(certificate), IPPROTO_TCP,                            "192.168.1.1", "192.168.1.5", 443, 51251);    FLOW_INITIALIZE(&f);    f.flags |= FLOW_IPV4;    f.proto = IPPROTO_TCP;    f.protomap = FlowGetProtoMapping(f.proto);    f.alproto = ALPROTO_TLS;    p1->flow = &f;    p1->flags |= PKT_HAS_FLOW | PKT_STREAM_EST;    p1->flowflags |= FLOW_PKT_TOSERVER;    p1->flowflags |= FLOW_PKT_ESTABLISHED;    p1->pcap_cnt = 1;    p2->flow = &f;    p2->flags |= PKT_HAS_FLOW | PKT_STREAM_EST;    p2->flowflags |= FLOW_PKT_TOCLIENT;    p2->flowflags |= FLOW_PKT_ESTABLISHED;    p2->pcap_cnt = 2;    p3->flow = &f;    p3->flags |= PKT_HAS_FLOW | PKT_STREAM_EST;    p3->flowflags |= FLOW_PKT_TOCLIENT;    p3->flowflags |= FLOW_PKT_ESTABLISHED;    p3->pcap_cnt = 3;    StreamTcpInitConfig(TRUE);    DetectEngineCtx *de_ctx = DetectEngineCtxInit();    FAIL_IF_NULL(de_ctx);    de_ctx->mpm_matcher = DEFAULT_MPM;    de_ctx->flags |= DE_QUIET;    s = DetectEngineAppendSig(de_ctx, "alert tls any any -> any any "                              "(msg:/"Test tls_cert_issuer/"; "                              "tls_cert_issuer; content:/"google/"; nocase; "                              "sid:1;)");    FAIL_IF_NULL(s);    SigGroupBuild(de_ctx);    DetectEngineThreadCtxInit(&tv, (void *)de_ctx, (void *)&det_ctx);    FLOWLOCK_WRLOCK(&f);    int r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_TLS,                                STREAM_TOSERVER, client_hello,                                sizeof(client_hello));    FLOWLOCK_UNLOCK(&f);    FAIL_IF(r != 0);    ssl_state = f.alstate;    FAIL_IF_NULL(ssl_state);    SigMatchSignatures(&tv, de_ctx, det_ctx, p1);    FAIL_IF(PacketAlertCheck(p1, 1));    FLOWLOCK_WRLOCK(&f);    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_TLS, STREAM_TOCLIENT,                            server_hello, sizeof(server_hello));    FLOWLOCK_UNLOCK(&f);    FAIL_IF(r != 0);    SigMatchSignatures(&tv, de_ctx, det_ctx, p2);    FAIL_IF(PacketAlertCheck(p2, 1));    FLOWLOCK_WRLOCK(&f);    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_TLS, STREAM_TOCLIENT,                            certificate, sizeof(certificate));    FLOWLOCK_UNLOCK(&f);    FAIL_IF(r != 0);    SigMatchSignatures(&tv, de_ctx, det_ctx, p3);    FAIL_IF_NOT(PacketAlertCheck(p3, 1));    if (alp_tctx != NULL)        AppLayerParserThreadCtxFree(alp_tctx);    if (det_ctx != NULL)        DetectEngineThreadCtxDeinit(&tv, det_ctx);    if (de_ctx != NULL)        SigGroupCleanup(de_ctx);    if (de_ctx != NULL)        DetectEngineCtxFree(de_ctx);    StreamTcpFreeConfig(TRUE);    FLOW_DESTROY(&f);    UTHFreePacket(p1);    UTHFreePacket(p2);    UTHFreePacket(p3);    PASS;}

static TmEcode LogFilestoreLogWrap(ThreadVars *tv, Packet *p, void *data, PacketQueue *pq, PacketQueue *postpq, int ipver){    SCEnter();    LogFilestoreLogThread *aft = (LogFilestoreLogThread *)data;    uint8_t flags = 0;    /* no flow, no htp state */    if (p->flow == NULL) {        SCReturnInt(TM_ECODE_OK);    }    if (p->flowflags & FLOW_PKT_TOCLIENT)        flags |= STREAM_TOCLIENT;    else        flags |= STREAM_TOSERVER;    int file_close = (p->flags & PKT_PSEUDO_STREAM_END) ? 1 : 0;    int file_trunc = 0;    FLOWLOCK_WRLOCK(p->flow);    file_trunc = StreamTcpReassembleDepthReached(p);    FileContainer *ffc = AppLayerGetFilesFromFlow(p->flow, flags);    SCLogDebug("ffc %p", ffc);    if (ffc != NULL) {        File *ff;        for (ff = ffc->head; ff != NULL; ff = ff->next) {            int file_fd = -1;            if (FileForceMagic() && ff->magic == NULL) {                FilemagicGlobalLookup(ff);            }            SCLogDebug("ff %p", ff);            if (ff->flags & FILE_STORED) {                SCLogDebug("stored flag set");                continue;            }            if (!(ff->flags & FILE_STORE)) {                SCLogDebug("ff FILE_STORE not set");                continue;            }            FileData *ffd;            for (ffd = ff->chunks_head; ffd != NULL; ffd = ffd->next) {                SCLogDebug("ffd %p", ffd);                if (ffd->stored == 1) {                    if (file_close == 1 && ffd->next == NULL) {                        LogFilestoreLogCloseMetaFile(ff);                        ff->flags |= FILE_STORED;                    }                    continue;                }                /* store */                SCLogDebug("trying to open file");                char filename[PATH_MAX] = "";                if (ff->file_id == 0) {                    ff->file_id = SC_ATOMIC_ADD(file_id, 1);                    snprintf(filename, sizeof(filename), "%s/file.%u",                            g_logfile_base_dir, ff->file_id);                    file_fd = open(filename, O_CREAT | O_TRUNC | O_NOFOLLOW | O_WRONLY, 0644);                    if (file_fd == -1) {                        SCLogDebug("failed to open file");                        continue;                    }                    /* create a .meta file that contains time, src/dst/sp/dp/proto */                    LogFilestoreLogCreateMetaFile(p, ff, filename, ipver);                    aft->file_cnt++;                } else {                    snprintf(filename, sizeof(filename), "%s/file.%u",                            g_logfile_base_dir, ff->file_id);                    file_fd = open(filename, O_APPEND | O_NOFOLLOW | O_WRONLY);                    if (file_fd == -1) {                        SCLogDebug("failed to open file %s: %s", filename, strerror(errno));                        continue;                    }                }                ssize_t r = write(file_fd, (const void *)ffd->data, (size_t)ffd->len);                if (r == -1) {                    SCLogDebug("write failed: %s", strerror(errno));                    close(file_fd);                    continue;                }                close(file_fd);                if (file_trunc && ff->state < FILE_STATE_CLOSED)                    ff->state = FILE_STATE_TRUNCATED;                if (ff->state == FILE_STATE_CLOSED ||//.........这里部分代码省略.........