detect/alert: add infra for new alert queue

Initial work to bring part of the alert queue processing to
DetectEngineThreadCtx.

Task #4943

(cherry picked from commit 88805f03eec038fd67a18599d7d9f037db083941)

diff --git a/src/detect-engine-alert.c b/src/detect-engine-alert.c
index a8a540d481a7e99192dc07ad3f677697cf35f214..da905ea6744a6e72fa1cd232d8046260e50d7547 100644 (file)
--- a/src/detect-engine-alert.c
+++ b/src/detect-engine-alert.c
@@ -272,6 +272,86 @@ static void PacketApplySignatureActions(Packet *p, const Signature *s, const uin
     }
 }
 
+void AlertQueueInit(DetectEngineThreadCtx *det_ctx)
+{
+    det_ctx->alert_queue_size = 0;
+    det_ctx->alert_queue = SCCalloc(packet_alert_max, sizeof(PacketAlert));
+    if (det_ctx->alert_queue == NULL) {
+        FatalError(SC_ERR_MEM_ALLOC, "failed to allocate %" PRIu64 " bytes for the alert queue",
+                (uint64_t)(packet_alert_max * sizeof(PacketAlert)));
+    }
+    det_ctx->alert_queue_capacity = packet_alert_max;
+    SCLogDebug("alert queue initialized to %u elements (%" PRIu64 " bytes)", packet_alert_max,
+            (uint64_t)(packet_alert_max * sizeof(PacketAlert)));
+}
+
+void AlertQueueFree(DetectEngineThreadCtx *det_ctx)
+{
+    SCFree(det_ctx->alert_queue);
+    det_ctx->alert_queue_capacity = 0;
+}
+
+/** \internal
+ * \retval the new capacity
+ */
+static uint16_t AlertQueueExpand(DetectEngineThreadCtx *det_ctx)
+{
+    uint16_t new_cap = det_ctx->alert_queue_capacity * 2;
+    void *tmp_queue = SCRealloc(det_ctx->alert_queue, (size_t)(sizeof(PacketAlert) * new_cap));
+    if (unlikely(tmp_queue == NULL)) {
+        /* queue capacity didn't change */
+        return det_ctx->alert_queue_capacity;
+    }
+    det_ctx->alert_queue = tmp_queue;
+    det_ctx->alert_queue_capacity = new_cap;
+    SCLogDebug("Alert queue size doubled: %u elements, bytes: %" PRIuMAX "",
+            det_ctx->alert_queue_capacity,
+            (uintmax_t)(sizeof(PacketAlert) * det_ctx->alert_queue_capacity));
+    return new_cap;
+}
+
+/** \internal
+ */
+static inline PacketAlert PacketAlertSet(
+        DetectEngineThreadCtx *det_ctx, const Signature *s, uint64_t tx_id, uint8_t alert_flags)
+{
+    PacketAlert pa;
+    pa.num = s->num;
+    pa.action = s->action;
+    pa.s = (Signature *)s;
+    pa.flags = alert_flags;
+    pa.tx_id = (tx_id == UINT64_MAX) ? 0 : tx_id;
+    return pa;
+}
+
+/**
+ * \brief Append signature to local packet alert queue for later preprocessing
+ */
+void AlertQueueAppend(DetectEngineThreadCtx *det_ctx, const Signature *s, Packet *p, uint64_t tx_id,
+        uint8_t alert_flags)
+{
+    /* first time we see a drop action signature, set that in the packet */
+    /* we do that even before inserting into the queue, so we save it even if appending fails */
+    if (p->alerts.drop.action == 0 && s->action & ACTION_DROP) {
+        p->alerts.drop = PacketAlertSet(det_ctx, s, tx_id, alert_flags);
+        SCLogDebug("Set PacketAlert drop action. s->num %" PRIu32 "", s->num);
+    }
+
+    uint16_t pos = det_ctx->alert_queue_size;
+    if (pos == det_ctx->alert_queue_capacity) {
+        /* we must grow the alert queue */
+        if (pos == AlertQueueExpand(det_ctx)) {
+            /* this means we failed to expand the queue */
+            return;
+        }
+    }
+    det_ctx->alert_queue[pos] = PacketAlertSet(det_ctx, s, tx_id, alert_flags);
+
+    SCLogDebug("Appending sid %" PRIu32 ", s->num %" PRIu32 " to alert queue", s->id, s->num);
+    det_ctx->alert_queue_size++;
+    return;
+}
+
 /**
  * \brief Check the threshold of the sigs that match, set actions, break on pass action
  *        This function iterate the packet alerts array, removing those that didn't match

diff --git a/src/detect-engine-alert.h b/src/detect-engine-alert.h
index dc6687885cc068e2ad52089678b191146a4cf8d7..81126b069157937418b6b011489ce2b121179405 100644 (file)
--- a/src/detect-engine-alert.h
+++ b/src/detect-engine-alert.h
@@ -28,6 +28,10 @@
 #include "decode.h"
 #include "detect.h"
 
+void AlertQueueInit(DetectEngineThreadCtx *det_ctx);
+void AlertQueueFree(DetectEngineThreadCtx *det_ctx);
+void AlertQueueAppend(DetectEngineThreadCtx *det_ctx, const Signature *s,
+                      Packet *p, uint64_t tx_id, uint8_t alert_flags);
 void PacketAlertFinalize(DetectEngineCtx *, DetectEngineThreadCtx *, Packet *);
 int PacketAlertAppend(DetectEngineThreadCtx *, const Signature *,
         Packet *, uint64_t tx_id, uint8_t);

diff --git a/src/detect-engine.c b/src/detect-engine.c
index a75afed0f662adc55cbbbcc7158fa4b9ac9b8c25..0d48d2a440402c06a8a17f3971fa54a1e8607ead 100644 (file)
--- a/src/detect-engine.c
+++ b/src/detect-engine.c
@@ -2803,6 +2803,9 @@ static TmEcode ThreadCtxDoInit (DetectEngineCtx *de_ctx, DetectEngineThreadCtx *
         RuleMatchCandidateTxArrayInit(det_ctx, de_ctx->sig_array_len);
     }
 
+    /* Alert processing queue */
+    AlertQueueInit(det_ctx);
+
     /* byte_extract storage */
     det_ctx->bj_values = SCMalloc(sizeof(*det_ctx->bj_values) *
                                   (de_ctx->byte_extract_max_local_id + 1));
@@ -3036,6 +3039,7 @@ static void DetectEngineThreadCtxFree(DetectEngineThreadCtx *det_ctx)
 
     if (det_ctx->bj_values != NULL)
         SCFree(det_ctx->bj_values);
+    AlertQueueFree(det_ctx);
 
     /* Decoded base64 data. */
     if (det_ctx->base64_decoded != NULL) {

diff --git a/src/detect.c b/src/detect.c
index 748b891c33b485f7f90c4a82aaac5d8a275ca44e..675f3e79b08a8b168197a2f2fd27e0697d5b184a 100644 (file)
--- a/src/detect.c
+++ b/src/detect.c
@@ -829,6 +829,9 @@ static DetectRunScratchpad DetectRunSetup(
     det_ctx->raw_stream_progress = 0;
     det_ctx->match_array_cnt = 0;
 
+    det_ctx->alert_queue_size = 0;
+    p->alerts.drop.action = 0;
+
 #ifdef DEBUG
     if (p->flags & PKT_STREAM_ADD) {
         det_ctx->pkt_stream_add_cnt++;

diff --git a/src/detect.h b/src/detect.h
index e81f335ba071a830e2cfb4a4a7c2c9d4ef4450d3..ea3c0b8ae5457ad0e801e691098cd0a796409b62 100644 (file)
--- a/src/detect.h
+++ b/src/detect.h
@@ -1080,6 +1080,10 @@ typedef struct DetectEngineThreadCtx_ {
     uint64_t tx_id;
     Packet *p;
 
+    uint16_t alert_queue_size;
+    uint16_t alert_queue_capacity;
+    PacketAlert *alert_queue;
+
     SC_ATOMIC_DECLARE(int, so_far_used_by_detect);
 
     /* holds the current recursion depth on content inspection */
@@ -1493,6 +1497,11 @@ void *DetectThreadCtxGetKeywordThreadCtx(DetectEngineThreadCtx *, int);
 void RuleMatchCandidateTxArrayInit(DetectEngineThreadCtx *det_ctx, uint32_t size);
 void RuleMatchCandidateTxArrayFree(DetectEngineThreadCtx *det_ctx);
 
+void AlertQueueInit(DetectEngineThreadCtx *det_ctx);
+void AlertQueueFree(DetectEngineThreadCtx *det_ctx);
+void AlertQueueAppend(DetectEngineThreadCtx *det_ctx, const Signature *s,
+                      Packet *p, uint64_t tx_id, uint8_t alert_flags);
+
 int DetectFlowbitsAnalyze(DetectEngineCtx *de_ctx);
 
 int DetectMetadataHashInit(DetectEngineCtx *de_ctx);