ssh: expose the parse failure reason and state in the eve ssh log

A flow that hit an unrecoverable ssh parse error (invalid banner
or invalid record) was never visible in the eve ssh output: the
log condition only admitted a successful handshake (both
directions at kex or session), and the tx logger is one-shot, so
the successful-handshake log fired before a later failure could
be reported and the failure was then silent - the reason was
only visible through the app-layer events.

The log condition is now failure-only: the failure latches, so
the object is emitted exactly once - at the failure, carrying
the error field (invalid_banner / invalid_record) and the state
field (the state the failure occurred in - banner, kex, or
session if a later record fails after NewKeys) - and successful
flows are logged at the flow-end flush, where the engine logs
unconditionally. Logging on the success condition mid-flow would
consume the one-shot tx log and drop a later failure, so it is
not done. The state field is emitted only alongside the error.
The object may be error + state only, when no banner was
parsed. A unit test pins the condition for a failed, a
banner-only and a both-at-kex flow (the failure admits the log
on the failing delivery); the user guide and eve reference doc
commits that follow carry the field docs.

That flow-end flush is skipped for flows that end without a
further app-layer update, though: in IPS/firewall mode a drop
disables the app layer (StreamTcpCheckFlowDrops ->
StreamTcpDisableAppLayer), after which no app-layer update
reaches the tx loggers, and a bypassed flow gets no flow-end
packet either. A successful ssh record - e.g. one that only saw
the banners - was therefore never emitted for those flows. The
hook is the app-layer end: on the transition to app-layer EOF,
StreamTcpDisableAppLayer raises FLOW_APP_LAYER_FLUSH_PENDING,
and FlowUpdateState raises it on the transition to a bypassed
state; OutputTxLog runs the tx loggers once for the next packet,
also when that packet itself has no app-layer update and is not
marked established (the disabling packet returns from the
stream engine before that), and flow-worker clears the flag when
it frees the app layer of a bypassed flow. The flush is
deliberately not a flow-end flush: when the flush is the only
reason to run the loggers, eof is not forced, so each logger's
condition decides and no other parser starts logging records it
suppressed; a disabling packet that does carry an app-layer
update keeps the existing EOF behaviour. SSHTxLogCondition
admits an IPS drop or a bypass, so the pending record is emitted
there; an IDS drop does not end the app layer, so the condition
stays failure-only - a later failure still wins, and the success
record is emitted by the flow-end flush, which the flow timeout
path now also runs for IDS-dropped flows. The unrelated
schema-sort ordering of the new state keys is fixed up as well.

Ticket: #9046.

etc/schema.json

@@ -6311,6 +6311,10 @@
                     "type": "object",
                     "additionalProperties": false,
                     "properties": {
+                        "error": {
+                            "type": "string",
+                            "description": "Unrecoverable parse failure in this direction (invalid_banner or invalid_record)"
+                        },
                         "hassh": {
                             "type": "object",
                             "additionalProperties": false,
@@ -6328,6 +6332,10 @@
                         },
                         "software_version": {
                             "type": "string"
+                        },
+                        "state": {
+                            "type": "string",
+                            "description": "State the parse failure occurred in (banner, kex or session); present only alongside the error field"
                         }
                     }
                 },
@@ -6335,6 +6343,10 @@
                     "type": "object",
                     "additionalProperties": false,
                     "properties": {
+                        "error": {
+                            "type": "string",
+                            "description": "Unrecoverable parse failure in this direction (invalid_banner or invalid_record)"
+                        },
                         "hassh": {
                             "type": "object",
                             "additionalProperties": false,
@@ -6352,6 +6364,10 @@
                         },
                         "software_version": {
                             "type": "string"
+                        },
+                        "state": {
+                            "type": "string",
+                            "description": "State the parse failure occurred in (banner, kex or session); present only alongside the error field"
                         }
                     }
                 }

rust/src/ssh/logger.rs

@@ -15,65 +15,69 @@
  * 02110-1301, USA.
  */
 
-use super::ssh::{SSHTransaction, SSH_MAX_BANNER_LEN};
+use super::ssh::{SSHConnectionState, SSHError, SSHTransaction, SshHeader, SSH_MAX_BANNER_LEN};
 use crate::jsonbuilder::{JsonBuilder, JsonError};
 
+fn log_ssh_direction(js: &mut JsonBuilder, name: &str, hdr: &SshHeader) -> Result<(), JsonError> {
+    // the direction object exists when it has data or a failure: a
+    // failed direction may have parsed no banner at all
+    if hdr.protover.is_empty() && hdr.error.is_none() {
+        return Ok(());
+    }
+    js.open_object(name)?;
+    if !hdr.protover.is_empty() {
+        js.set_string_from_bytes_limited("proto_version", &hdr.protover, SSH_MAX_BANNER_LEN)?;
+    }
+    if !hdr.swver.is_empty() {
+        js.set_string_from_bytes_limited("software_version", &hdr.swver, SSH_MAX_BANNER_LEN)?;
+    }
+    if let Some(error) = hdr.error {
+        let err_name = match error {
+            SSHError::InvalidBanner => "invalid_banner",
+            SSHError::InvalidRecord => "invalid_record",
+        };
+        js.set_string("error", err_name)?;
+        // the scale must stay total: a direction can be frozen in
+        // any of these, and the disrupted-flow end state (done)
+        // must not be mislabelled
+        let st_name = match hdr.state {
+            SSHConnectionState::SshStateBanner => "banner",
+            SSHConnectionState::SshStateBannerWaitEol => "banner_wait_eol",
+            SSHConnectionState::SshStateKex => "kex",
+            // a direction that reached session can still fail on a
+            // later record (the encryption bypass has not kicked in
+            // yet)
+            SSHConnectionState::SshStateSession => "session",
+            SSHConnectionState::SshStateDone => "done",
+        };
+        js.set_string("state", st_name)?;
+    }
+    if !hdr.hassh.is_empty() || !hdr.hassh_string.is_empty() {
+        js.open_object("hassh")?;
+        if !hdr.hassh.is_empty() {
+            js.set_string_from_bytes("hash", &hdr.hassh)?;
+        }
+        if !hdr.hassh_string.is_empty() {
+            js.set_string_from_bytes("string", &hdr.hassh_string)?;
+        }
+        js.close()?;
+    }
+    js.close()?;
+    Ok(())
+}
+
 fn log_ssh(tx: &SSHTransaction, js: &mut JsonBuilder) -> Result<bool, JsonError> {
     js.open_object("ssh")?;
-    if tx.cli_hdr.protover.is_empty() && tx.srv_hdr.protover.is_empty() {
+    let cli = !tx.cli_hdr.protover.is_empty() || tx.cli_hdr.error.is_some();
+    let srv = !tx.srv_hdr.protover.is_empty() || tx.srv_hdr.error.is_some();
+    if !cli && !srv {
         return Ok(false);
     }
-    if !tx.cli_hdr.protover.is_empty() {
-        js.open_object("client")?;
-        js.set_string_from_bytes_limited(
-            "proto_version",
-            &tx.cli_hdr.protover,
-            SSH_MAX_BANNER_LEN,
-        )?;
-        if !tx.cli_hdr.swver.is_empty() {
-            js.set_string_from_bytes_limited(
-                "software_version",
-                &tx.cli_hdr.swver,
-                SSH_MAX_BANNER_LEN,
-            )?;
-        }
-        if !tx.cli_hdr.hassh.is_empty() || !tx.cli_hdr.hassh_string.is_empty() {
-            js.open_object("hassh")?;
-            if !tx.cli_hdr.hassh.is_empty() {
-                js.set_string_from_bytes("hash", &tx.cli_hdr.hassh)?;
-            }
-            if !tx.cli_hdr.hassh_string.is_empty() {
-                js.set_string_from_bytes("string", &tx.cli_hdr.hassh_string)?;
-            }
-            js.close()?;
-        }
-        js.close()?;
+    if cli {
+        log_ssh_direction(js, "client", &tx.cli_hdr)?;
     }
-    if !tx.srv_hdr.protover.is_empty() {
-        js.open_object("server")?;
-        js.set_string_from_bytes_limited(
-            "proto_version",
-            &tx.srv_hdr.protover,
-            SSH_MAX_BANNER_LEN,
-        )?;
-        if !tx.srv_hdr.swver.is_empty() {
-            js.set_string_from_bytes_limited(
-                "software_version",
-                &tx.srv_hdr.swver,
-                SSH_MAX_BANNER_LEN,
-            )?;
-        }
-        if !tx.srv_hdr.hassh.is_empty() || !tx.srv_hdr.hassh_string.is_empty() {
-            js.open_object("hassh")?;
-            if !tx.srv_hdr.hassh.is_empty() {
-                js.set_string_from_bytes("hash", &tx.srv_hdr.hassh)?;
-            }
-            if !tx.srv_hdr.hassh_string.is_empty() {
-                js.set_string_from_bytes("string", &tx.srv_hdr.hassh_string)?;
-            }
-            js.close()?;
-        }
-        js.close()?;
+    if srv {
+        log_ssh_direction(js, "server", &tx.srv_hdr)?;
     }
     js.close()?;
     return Ok(true);

rust/src/ssh/ssh.rs

@@ -127,7 +127,7 @@ pub struct SshHeader {
     record_left: u32,
     record_left_msg: parser::MessageCode,
 
-    state: SSHConnectionState,
+    pub state: SSHConnectionState,
     pub protover: Vec<u8>,
     pub swver: Vec<u8>,
     pub error: Option<SSHError>,
@@ -763,21 +763,13 @@ pub extern "C" fn SCSshEnableBypass(mode: EncryptionHandling) {
 pub unsafe extern "C" fn SCSshTxGetLogCondition(tx: *mut std::os::raw::c_void) -> bool {
     let tx = cast_pointer!(tx, SSHTransaction);
 
-    // >= rather than ==: a direction at session has necessarily
-    // passed kex, and a single delivery can take a direction
-    // banner -> kex -> session, so a final {session, kex} pair must
-    // still admit the log (done is never assigned, so >= changes
-    // only the kex branch)
-    if SCSshHasshIsEnabled() {
-        if tx.cli_hdr.state >= SSHConnectionState::SshStateSession
-            && tx.srv_hdr.state >= SSHConnectionState::SshStateSession
-        {
-            return true;
-        }
-    } else if tx.cli_hdr.state >= SSHConnectionState::SshStateKex
-        && tx.srv_hdr.state >= SSHConnectionState::SshStateKex
-    {
-        return true;
-    }
-    return false;
+    // Failure-only: the tx logger is one-shot (the engine's logged
+    // bit is never reset), and a successful handshake reaches kex
+    // long before a later failure can occur, so a mid-flow success
+    // condition would consume the log before the failure could be
+    // reported. The failure latches, so the object is emitted
+    // exactly once - at the failure, carrying the error and the
+    // frozen state - and successful flows are logged at the next
+    // EOF-flush delivery, where the engine logs unconditionally.
+    tx.cli_hdr.error.is_some() || tx.srv_hdr.error.is_some()
 }

src/app-layer-ssh.c

@@ -26,6 +26,7 @@
  */
 
 #include "suricata-common.h"
+#include "suricata.h"
 #include "decode.h"
 
 #include "app-layer-detect-proto.h"
@@ -63,6 +64,21 @@ static int SSHRegisterPatternsForProtocolDetection(void)
 
 bool SSHTxLogCondition(ThreadVars *tv, const Packet *p, void *state, void *tx, uint64_t tx_id)
 {
+    /* A pending success record must be emitted when an IPS drop or a
+     * bypass ends the flow: the flow-end flush that would log it is
+     * skipped for those flows, and the app layer is disabled or freed
+     * right after, so no further update reaches this condition. An IDS
+     * drop keeps parsing, so the condition stays failure-only there: a
+     * later failure still wins, and the success record is emitted by
+     * the flow-end flush. */
+    if (p->flow != NULL) {
+        if ((p->flow->flags & FLOW_ACTION_DROP) != 0 && EngineModeIsIPS()) {
+            return true;
+        }
+        if (FlowIsBypassed(p->flow)) {
+            return true;
+        }
+    }
     return SCSshTxGetLogCondition(tx);
 }
 
@@ -1900,6 +1916,101 @@ static int SSHParserTest30(void)
     PASS;
 }
 
+/** \test The eve ssh log condition is failure-only: a direction
+ *  \test that fails at banner (invalid banner) satisfies the
+ *  \test condition (the error field makes the failure observable
+ *  \test in eve), while a flow without failure does not - neither
+ *  \test a banner-only flow nor one with both banners parsed
+ *  \test (both directions at kex): the one-shot tx log must not be
+ *  \test consumed mid-flow before a later failure could be
+ *  \test reported, so successful flows are logged at the flow-end
+ *  \test flush. The failure admits the log on the failing
+ *  \test delivery.
+ */
+static int SSHParserTest31(void)
+{
+    Flow f;
+    uint8_t badbanner[] = "SSH-bogus\r\n";
+    uint32_t badbannerlen = sizeof(badbanner) - 1;
+    uint8_t banner[] = "SSH-2.0-TestClient-1.0\r\n";
+    uint32_t bannerlen = sizeof(banner) - 1;
+    TcpSession ssn;
+    AppLayerParserThreadCtx *alp_tctx = AppLayerParserThreadCtxAlloc();
+    FAIL_IF_NULL(alp_tctx);
+
+    /* invalid banner: the failed direction admits the log condition */
+    memset(&f, 0, sizeof(f));
+    memset(&ssn, 0, sizeof(ssn));
+    FLOW_INITIALIZE(&f);
+    f.protoctx = (void *)&ssn;
+    f.proto = IPPROTO_TCP;
+    f.alproto = ALPROTO_SSH;
+
+    StreamTcpInitConfig(true);
+
+    int r = AppLayerParserParse(
+            NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOSERVER, badbanner, badbannerlen);
+    FAIL_IF(r != -1);
+    void *ssh_state = f.alstate;
+    FAIL_IF_NULL(ssh_state);
+    void *tx = SCSshStateGetTx(ssh_state, 0);
+    FAIL_IF(SCSshTxGetAlStateProgress(tx, STREAM_TOSERVER) != SshStateBanner);
+    FAIL_IF(SCSshTxGetLogCondition(tx) != true);
+
+    FLOW_DESTROY(&f);
+
+    /* no failure yet (client banner only): the condition stays
+     * false - the one-shot tx log is reserved for the failure
+     * or the next EOF-flush delivery */
+    memset(&f, 0, sizeof(f));
+    memset(&ssn, 0, sizeof(ssn));
+    FLOW_INITIALIZE(&f);
+    f.protoctx = (void *)&ssn;
+    f.proto = IPPROTO_TCP;
+    f.alproto = ALPROTO_SSH;
+
+    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOSERVER, banner, bannerlen);
+    FAIL_IF(r != 0);
+    ssh_state = f.alstate;
+    FAIL_IF_NULL(ssh_state);
+    tx = SCSshStateGetTx(ssh_state, 0);
+    FAIL_IF(SCSshTxGetLogCondition(tx) != false);
+
+    FLOW_DESTROY(&f);
+
+    /* both banners parsed (both directions at kex), still no
+     * failure: the condition must stay false - a mid-flow success
+     * log would consume the one-shot tx log before a later failure
+     * could be reported */
+    memset(&f, 0, sizeof(f));
+    memset(&ssn, 0, sizeof(ssn));
+    FLOW_INITIALIZE(&f);
+    f.protoctx = (void *)&ssn;
+    f.proto = IPPROTO_TCP;
+    f.alproto = ALPROTO_SSH;
+
+    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOSERVER, banner, bannerlen);
+    FAIL_IF(r != 0);
+    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOCLIENT, banner, bannerlen);
+    FAIL_IF(r != 0);
+    ssh_state = f.alstate;
+    FAIL_IF_NULL(ssh_state);
+    tx = SCSshStateGetTx(ssh_state, 0);
+    FAIL_IF(SCSshTxGetLogCondition(tx) != false);
+
+    /* the failure admits the log on the failing delivery */
+    uint8_t badrecord[] = { 0x00, 0x00, 0x00, 0x00, 0x00, 0x00 };
+    r = AppLayerParserParse(
+            NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOCLIENT, badrecord, sizeof(badrecord));
+    FAIL_IF(r != -1);
+    FAIL_IF(SCSshTxGetLogCondition(tx) != true);
+
+    FLOW_DESTROY(&f);
+    AppLayerParserThreadCtxFree(alp_tctx);
+    StreamTcpFreeConfig(true);
+    PASS;
+}
+
 /** \test A banner line split over two segments: the open-line segment
  *  \test publishes nothing but still marks the tx updated at parse
  *  \test entry; the line completion publishes the banner and takes
@@ -2091,13 +2202,13 @@ static int SSHParserTest34(void)
     // the failing direction is frozen at kex, still readable
     FAIL_IF(SCSshTxGetAlStateProgress(tx, STREAM_TOSERVER) != SshStateKex);
     // the failure event published: the tx is marked updated, so the
-    // engine evaluates it (not only at the flow-end flush)
+    // engine evaluates it (not only at the next EOF-flush delivery)
     struct AppLayerTxData *txdata = AppLayerParserGetTxData(IPPROTO_TCP, ALPROTO_SSH, tx);
     FAIL_IF_NULL(txdata);
     FAIL_IF(!txdata->updated_ts);
 
     // further data in either direction does not reach the parser -
-    // neither directly nor via the flow-end flush: the disable flag
+    // neither directly nor via the next EOF-flush delivery: the disable flag
     // is checked at the reassembly entry, so the buffered
     // pre-failure server banner is not delivered at EOF
     p->flags |= PKT_PSEUDO_STREAM_END;
@@ -2246,6 +2357,7 @@ void SSHParserRegisterTests(void)
     UtRegisterTest(
             "SSHParserTest29 - stash and fragment deliveries mark the tx updated", SSHParserTest29);
     UtRegisterTest("SSHParserTest30 - wait_eol failure freezes at kex", SSHParserTest30);
+    UtRegisterTest("SSHParserTest31 - failed flow admits the eve log condition", SSHParserTest31);
     UtRegisterTest("SSHParserTest32 - banner continuation marks the tx updated at parse entry",
             SSHParserTest32);
     UtRegisterTest("SSHParserTest33 - failed peer does not arm no-inspection", SSHParserTest33);

src/detect-engine-alert.c

@@ -181,9 +181,7 @@ static inline void RuleActionToFlow(const uint8_t action, Flow *f, const bool fw
          * firewall rules. */
         if (f->flags & FLOW_ACTION_ACCEPT) {
             f->flags &= ~(FLOW_ACTION_ACCEPT | FLOW_ACTION_BY_FIREWALL);
-            f->flags |= FLOW_ACTION_DROP;
-            if (fw_rule)
-                f->flags |= FLOW_ACTION_BY_FIREWALL;
+            FlowSetDropAction(f, fw_rule ? FLOW_ACTION_BY_FIREWALL : 0);
             SCLogDebug("replaced FLOW_ACTION_ACCEPT with FLOW_ACTION_DROP");
         }
         if (f->flags & (FLOW_ACTION_DROP | FLOW_ACTION_PASS)) {
@@ -192,7 +190,7 @@ static inline void RuleActionToFlow(const uint8_t action, Flow *f, const bool fw
                     (action & ACTION_PASS) ? "pass" : "drop",
                     (f->flags & FLOW_ACTION_DROP) ? "drop" : "pass");
         } else {
-            f->flags |= FLOW_ACTION_DROP;
+            FlowSetDropAction(f, 0);
             SCLogDebug("setting flow action drop");
         }
     }
@@ -575,7 +573,7 @@ static struct DetectFirewallPolicy HandleFirewallRule(
                 }
                 PacketDrop(p, pol.action, drop_reason);
                 if (p->flow && pol.action_scope == ACTION_SCOPE_FLOW) {
-                    p->flow->flags |= FLOW_ACTION_DROP | FLOW_ACTION_BY_FIREWALL;
+                    FlowSetDropAction(p->flow, FLOW_ACTION_BY_FIREWALL);
                     SCLogDebug("packet %" PRIu64 ": FLOW_ACTION_DROP set by firewall by sid %u",
                             PcapPacketCntGet(p), s->id);
                 }

src/detect.c

@@ -1773,8 +1773,7 @@ static struct DetectFirewallPolicy DetectFirewallApplyDefaultAppPolicy(
         PacketDrop(p, policy->action, PKT_DROP_REASON_FW_DEFAULT_APP_POLICY);
         if (policy->action_scope == ACTION_SCOPE_FLOW) {
             SCLogDebug("dropping flow");
-            p->flow->flags |= FLOW_ACTION_DROP;
-            p->flow->flags |= FLOW_ACTION_BY_FIREWALL;
+            FlowSetDropAction(p->flow, FLOW_ACTION_BY_FIREWALL);
         }
         if (policy->action & ACTION_ALERT) {
             DetectRunAppendDefaultAppPolicyAlert(det_ctx, p, true, tx, ap);
@@ -2251,8 +2250,7 @@ static void DetectRunTxFirewallRuleFullMatch(DetectEngineThreadCtx *det_ctx, con
         PacketDrop(p, s->action, PKT_DROP_REASON_FW_RULES);
         if (s->action_scope == ACTION_SCOPE_FLOW) {
             SCLogDebug("drop flow because of rule with drop action");
-            f->flags |= FLOW_ACTION_DROP;
-            f->flags |= FLOW_ACTION_BY_FIREWALL;
+            FlowSetDropAction(f, FLOW_ACTION_BY_FIREWALL);
         }
         SCLogDebug("append alert");
         AlertQueueAppendAppTx(det_ctx, s, p, tx->tx_id, tx->tx_type, 0);

src/flow-manager.c

@@ -23,6 +23,7 @@
  */
 
 #include "suricata-common.h"
+#include "suricata.h"
 #include "conf.h"
 #include "threadvars.h"
 #include "tm-threads.h"
@@ -303,8 +304,10 @@ static uint32_t ProcessAsideQueue(FlowManagerTimeoutThread *td, FlowTimeoutCount
     while ((f = FlowQueuePrivateGetFromTop(&td->aside_queue)) != NULL) {
         /* flow is still locked */
 
-        if (f->proto == IPPROTO_TCP &&
-                !(f->flags & (FLOW_TIMEOUT_REASSEMBLY_DONE | FLOW_ACTION_DROP)) &&
+        /* An IPS drop ends the flow; an IDS drop is an alert and the app
+         * layer keeps running, so the final flush still applies. */
+        const bool drop_ends = (f->flags & FLOW_ACTION_DROP) != 0 && EngineModeIsIPS();
+        if (f->proto == IPPROTO_TCP && !(f->flags & FLOW_TIMEOUT_REASSEMBLY_DONE) && !drop_ends &&
                 !FlowIsBypassed(f) && FlowNeedsReassembly(f)) {
             /* Send the flow to its thread */
             FlowSendToLocalThread(f);

src/flow-worker.c

@@ -163,8 +163,10 @@ static void CheckWorkQueue(ThreadVars *tv, FlowWorkerThreadData *fw, FlowTimeout
         f->flow_end_flags |= FLOW_END_FLAG_TIMEOUT; //TODO emerg
 
         if (f->proto == IPPROTO_TCP) {
-            if (!(f->flags & (FLOW_TIMEOUT_REASSEMBLY_DONE | FLOW_ACTION_DROP)) &&
-                    !FlowIsBypassed(f) && FlowNeedsReassembly(f) && f->ffr != 0) {
+            /* an IDS drop does not end the app layer, so keep the final flush */
+            const bool drop_ends = (f->flags & FLOW_ACTION_DROP) != 0 && EngineModeIsIPS();
+            if (!(f->flags & FLOW_TIMEOUT_REASSEMBLY_DONE) && !drop_ends && !FlowIsBypassed(f) &&
+                    FlowNeedsReassembly(f) && f->ffr != 0) {
                 /* read detect thread in case we're doing a reload */
                 void *detect_thread = SC_ATOMIC_GET(fw->detect_thread);
                 int cnt = FlowFinish(tv, f, fw, detect_thread);
@@ -673,6 +675,9 @@ pre_flow_drop:
         DEBUG_ASSERT_FLOW_LOCKED(p->flow);
 
         if (FlowIsBypassed(p->flow)) {
+            /* the tx loggers already ran for a pending flush (OutputLoggerLog
+             * above): clear the flag and free the app layer */
+            p->flow->flags &= ~FLOW_APP_LAYER_FLUSH_PENDING;
             FlowCleanupAppLayer(p->flow);
             if (p->proto == IPPROTO_TCP) {
                 StreamTcpSessionCleanup(p->flow->protoctx);

src/flow.c

@@ -1196,6 +1196,18 @@ uint8_t FlowGetDisruptionFlags(const Flow *f, uint8_t flags)
 void FlowUpdateState(Flow *f, const enum FlowState s)
 {
     if (s != f->flow_state) {
+        /* a bypass ends the app layer without a further update: bypassed
+         * flows get no flow-end packets, so the txs still pending must be
+         * logged once */
+        if ((s == FLOW_STATE_LOCAL_BYPASSED
+#ifdef CAPTURE_OFFLOAD
+                    || s == FLOW_STATE_CAPTURE_BYPASSED
+#endif
+                    ) &&
+                f->alparser != NULL) {
+            f->flags |= FLOW_APP_LAYER_FLUSH_PENDING;
+        }
+
         /* set the state */
         // Explicit cast from the enum type to the compact version
         f->flow_state = (FlowStateType)s;

src/flow.h

@@ -127,6 +127,11 @@ typedef struct AppLayerParserState_ AppLayerParserState;
 /** Flow action issued by exception policy */
 #define FLOW_ACTION_BY_EXCEPTION_POLICY BIT_U64(33)
 
+/** The app layer was disabled mid-flow: run the tx loggers once at the
+ *  next output stage, as no further app-layer update will reach them.
+ *  Consumed by the tx logger. */
+#define FLOW_APP_LAYER_FLUSH_PENDING BIT_U64(34)
+
 /* File flags */
 
 #define FLOWFILE_INIT                   0
@@ -593,6 +598,18 @@ static inline void *FlowGetAppState(const Flow *f)
  *
  * \param f Flow to set the flag in
  */
+/** \brief mark the flow dropped and schedule the pending tx loggers.
+ *
+ * A flow scope drop ends the app layer without a further update: the txs
+ * still pending (e.g. the ssh success record) must be logged once. The
+ * output stage of the dropping packet consumes the flag. */
+static inline void FlowSetDropAction(Flow *f, const uint64_t by)
+{
+    f->flags |= FLOW_ACTION_DROP | by;
+    if (f->alparser != NULL)
+        f->flags |= FLOW_APP_LAYER_FLUSH_PENDING;
+}
+
 static inline void FlowSetNoPayloadInspectionFlag(Flow *f)
 {
     SCEnter();

src/output-tx.c

@@ -367,17 +367,38 @@ static TmEcode OutputTxLog(ThreadVars *tv, Packet *p, void *thread_data)
     DEBUG_VALIDATE_BUG_ON(thread_data == NULL);
     if (p->flow == NULL)
         return TM_ECODE_OK;
-    if (!PKT_IS_PSEUDOPKT(p) && p->app_update_direction == 0 &&
+
+    /* A flow whose app layer gets disabled mid-flow (IPS/firewall drop,
+     * parser error, bypass) must still run the tx loggers once: no
+     * further app-layer update will reach them and flow end skips
+     * FLOW_ACTION_DROP flows. The disable path raises
+     * FLOW_APP_LAYER_FLUSH_PENDING; consume it here. */
+    const bool flush_pending = (p->flow->flags & FLOW_APP_LAYER_FLUSH_PENDING) != 0;
+
+    const bool no_app_update =
+            !PKT_IS_PSEUDOPKT(p) && p->app_update_direction == 0 &&
             ((PKT_IS_TOSERVER(p) && (p->flow->flags & FLOW_TS_APP_UPDATED) == 0) ||
-                    (PKT_IS_TOCLIENT(p) && (p->flow->flags & FLOW_TC_APP_UPDATED) == 0))) {
+                    (PKT_IS_TOCLIENT(p) && (p->flow->flags & FLOW_TC_APP_UPDATED) == 0));
+    if (no_app_update && !flush_pending) {
         SCLogDebug("not pseudo, no app update: skip");
         return TM_ECODE_OK;
     }
-    if ((p->flags & PKT_STREAM_EST) == 0 && p->proto == IPPROTO_TCP) {
+    /* The flush is the only reason to run the loggers for this packet:
+     * the packet itself carries no app-layer update, so it is not a
+     * flow end either and the log conditions decide instead of an
+     * unconditional log. */
+    const bool flush_only = flush_pending && no_app_update;
+
+    /* The packet that disables the app layer returns from the stream
+     * engine before it is marked established, so the flush request has
+     * to bypass this check as well. */
+    if ((p->flags & PKT_STREAM_EST) == 0 && p->proto == IPPROTO_TCP && !flush_only) {
         return TM_ECODE_OK;
     }
     SCLogDebug("pseudo, or app update: run output");
 
+    p->flow->flags &= ~FLOW_APP_LAYER_FLUSH_PENDING;
+
     OutputTxLoggerThreadData *op_thread_data = (OutputTxLoggerThreadData *)thread_data;
 
     Flow * const f = p->flow;
@@ -413,7 +434,7 @@ static TmEcode OutputTxLog(ThreadVars *tv, Packet *p, void *thread_data)
     const bool ts_eof = SCAppLayerParserStateIssetFlag(f->alparser, APP_LAYER_PARSER_EOF_TS) != 0;
     const bool tc_eof = SCAppLayerParserStateIssetFlag(f->alparser, APP_LAYER_PARSER_EOF_TC) != 0;
 
-    const bool eof = last_pseudo || (ts_eof && tc_eof);
+    const bool eof = !flush_only && (last_pseudo || (ts_eof && tc_eof));
     SCLogDebug("eof %d last_pseudo %d ts_eof %d tc_eof %d", eof, last_pseudo, ts_eof, tc_eof);
 
     const uint8_t ts_disrupt_flags = FlowGetDisruptionFlags(f, STREAM_TOSERVER);
@@ -499,8 +520,12 @@ static TmEcode OutputTxLog(ThreadVars *tv, Packet *p, void *thread_data)
             }
         }
         SCLogDebug("logger: expect %08x, have %08x", logger_expectation, txd->logged);
-        if (!txd->updated_tc && !txd->updated_ts && !(tx_progress_ts == complete_ts) &&
-                !(tx_progress_tc == complete_tc) && !ts_eof && !tc_eof) {
+        /* a flush runs the loggers for the txs still pending even when
+         * nothing marked them updated: the flow is ending without a further
+         * app-layer update */
+        if (!flush_only && !txd->updated_tc && !txd->updated_ts &&
+                !(tx_progress_ts == complete_ts) && !(tx_progress_tc == complete_tc) && !ts_eof &&
+                !tc_eof) {
             gap = true;
             goto next_tx;
         }

src/stream-tcp-reassemble.c

@@ -452,8 +452,18 @@ void StreamTcpDisableAppLayer(Flow *f)
     StreamTcpSetStreamFlagAppProtoDetectionCompleted(&ssn->server);
     StreamTcpDisableAppLayerReassembly(ssn);
     if (f->alparser) {
-        SCAppLayerParserStateSetFlag(
-                f->alparser, (APP_LAYER_PARSER_EOF_TS | APP_LAYER_PARSER_EOF_TC));
+        const uint16_t eof_flags = APP_LAYER_PARSER_EOF_TS | APP_LAYER_PARSER_EOF_TC;
+        const bool was_eof = SCAppLayerParserStateIssetFlag(f->alparser, eof_flags) == eof_flags;
+        SCAppLayerParserStateSetFlag(f->alparser, eof_flags);
+        /* The app layer is at EOF from here on: no further app-layer
+         * update will reach the tx loggers, while the flow may end up
+         * dropped (flow end skips FLOW_ACTION_DROP flows), so the txs
+         * still pending would never be logged. Ask the next output
+         * stage to run the loggers once. Only on the transition, so
+         * the flush does not repeat per packet. */
+        if (!was_eof) {
+            f->flags |= FLOW_APP_LAYER_FLUSH_PENDING;
+        }
     }
 }