/**
 * \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));