ssh: finish parked banners at kex

A banner line that completes while parked in banner_wait_eol now
takes the direction to kex: the banner data was already parsed at
entry, the completion only advances the state. Before, the
direction skipped the kex state entirely and jumped to session at
the newkeys record.

The tx is marked updated at parse entry, as the base parser did,
before the dispatch: every delivery - including the stash and
header-fragment cases that parse_record handles and publish
nothing - is decided by the state rules and the firewall
default-policy sweep (default: drop) rather than left to the
firewall default-accept, as an un-updated tx is not evaluated
mid-flow. The parse entry does not pre-check those cases:
parse_record returns the same AppLayerResult for them, and the
record_left arithmetic must stay in sync in one place only.

Ticket: #9046.

rust/src/ssh/ssh.rs

@@ -192,34 +192,33 @@ impl SSHState {
             (&mut self.transaction.srv_hdr, &self.transaction.cli_hdr)
         };
         let il = input.len();
-        //first skip record left bytes
+        // first skip record left bytes; a buffer shorter than the
+        // pending record is absorbed into the stash. The parse entry
+        // does not pre-check this case (nor the leading header
+        // fragment, handled by the Incomplete arm below): both return
+        // the same AppLayerResult as the entry short-circuit would,
+        // and only this site must stay in sync with the record_left
+        // arithmetic
         if hdr.record_left > 0 {
-            //should we check for overflow ?
             let ilen = input.len() as u32;
-            if hdr.record_left > ilen {
-                hdr.record_left -= ilen;
+            if stash_record_bytes(&mut hdr.record_left, ilen) {
                 return AppLayerResult::ok();
-            } else {
-                let start = hdr.record_left as usize;
-                match hdr.record_left_msg {
-                    // parse reassembled tcp segments
-                    parser::MessageCode::Kexinit if hassh_is_enabled() => {
-                        if let Ok((_rem, key_exchange)) =
-                            parser::ssh_parse_key_exchange(&input[..start])
-                        {
-                            key_exchange.generate_hassh(
-                                &mut hdr.hassh_string,
-                                &mut hdr.hassh,
-                                &resp,
-                            );
-                        }
-                        hdr.record_left_msg = parser::MessageCode::Undefined(0);
+            }
+            let start = hdr.record_left as usize;
+            match hdr.record_left_msg {
+                // parse reassembled tcp segments
+                parser::MessageCode::Kexinit if hassh_is_enabled() => {
+                    if let Ok((_rem, key_exchange)) =
+                        parser::ssh_parse_key_exchange(&input[..start])
+                    {
+                        key_exchange.generate_hassh(&mut hdr.hassh_string, &mut hdr.hassh, &resp);
                     }
-                    _ => {}
+                    hdr.record_left_msg = parser::MessageCode::Undefined(0);
                 }
-                input = &input[start..];
-                hdr.record_left = 0;
+                _ => {}
             }
+            input = &input[start..];
+            hdr.record_left = 0;
         }
         //parse records out of input
         while !input.is_empty() {
@@ -339,9 +338,16 @@ impl SSHState {
                             return AppLayerResult::ok();
                         }
                         Err(Err::Incomplete(_)) => {
-                            //we may have consumed data from previous records
+                            // the buffer ran out between records in
+                            // this call; do not trust nom's
+                            // incomplete value. The header parser only
+                            // reports Incomplete when fewer than
+                            // SSH_RECORD_HEADER_LEN bytes remain, so
+                            // this asserts the arm's branch
+                            // precondition (a later parse change that
+                            // makes it reachable with a full header
+                            // aborts debug builds)
                             debug_validate_bug_on!(input.len() >= SSH_RECORD_HEADER_LEN);
-                            //do not trust nom incomplete value
                             return AppLayerResult::incomplete(
                                 (il - input.len()) as u32,
                                 SSH_RECORD_HEADER_LEN as u32,
@@ -378,6 +384,9 @@ impl SSHState {
         if hdr.state == SSHConnectionState::SshStateBannerWaitEol {
             match parser::ssh_parse_line(input) {
                 Ok((rem, _)) => {
+                    // line complete: the banner data was parsed at
+                    // entry, this only takes the direction to kex
+                    hdr.state = SSHConnectionState::SshStateKex;
                     let mut r = self.parse_record(rem, resp, pstate, flow, stream_slice);
                     if r.is_incomplete() {
                         //adds bytes consumed by banner to incomplete result
@@ -497,6 +506,17 @@ extern "C" fn ssh_state_tx_free(_state: *mut std::os::raw::c_void, _tx_id: u64)
     //do nothing
 }
 
+/// Absorb `ilen` bytes into the pending-record stash. True if the
+/// whole chunk was stashed (nothing was parsed).
+fn stash_record_bytes(record_left: &mut u32, ilen: u32) -> bool {
+    if *record_left > ilen {
+        *record_left -= ilen;
+        true
+    } else {
+        false
+    }
+}
+
 unsafe extern "C" fn ssh_parse_request(
     flow: *mut Flow, state: *mut std::os::raw::c_void, pstate: *mut AppLayerParserState,
     stream_slice: StreamSlice, _data: *mut std::os::raw::c_void,
@@ -512,12 +532,19 @@ unsafe extern "C" fn ssh_parse_request(
     if hdr.error.is_some() {
         return AppLayerResult::ok();
     }
+    // Mark the tx updated at parse entry, as the base parser did:
+    // every delivery - including the short-circuits below that
+    // publish nothing - is decided by the state rules and the
+    // firewall default-policy sweep rather than the firewall
+    // default-accept.
     state.transaction.tx_data.0.updated_ts = true;
-    if hdr.state < SSHConnectionState::SshStateKex {
-        return state.parse_banner(buf, false, pstate, flow, &stream_slice);
+
+    let r = if hdr.state < SSHConnectionState::SshStateKex {
+        state.parse_banner(buf, false, pstate, flow, &stream_slice)
     } else {
-        return state.parse_record(buf, false, pstate, flow, &stream_slice);
-    }
+        state.parse_record(buf, false, pstate, flow, &stream_slice)
+    };
+    return r;
 }
 
 unsafe extern "C" fn ssh_parse_response(
@@ -528,16 +555,19 @@ unsafe extern "C" fn ssh_parse_response(
     let buf = stream_slice.as_slice();
     let hdr = &mut state.transaction.srv_hdr;
 
-    // See ssh_parse_request for the failure freeze.
+    // See ssh_parse_request for the failure freeze and the updated
+    // flag.
     if hdr.error.is_some() {
         return AppLayerResult::ok();
     }
     state.transaction.tx_data.0.updated_tc = true;
-    if hdr.state < SSHConnectionState::SshStateKex {
-        return state.parse_banner(buf, true, pstate, flow, &stream_slice);
+
+    let r = if hdr.state < SSHConnectionState::SshStateKex {
+        state.parse_banner(buf, true, pstate, flow, &stream_slice)
     } else {
-        return state.parse_record(buf, true, pstate, flow, &stream_slice);
-    }
+        state.parse_record(buf, true, pstate, flow, &stream_slice)
+    };
+    return r;
 }
 
 #[no_mangle]

src/app-layer-ssh.c

@@ -1700,6 +1700,11 @@ static int SSHParserTest28(void)
     FAIL_IF(SCSshTxGetAlStateProgress(tx, STREAM_TOSERVER) != SshStateBanner);
     FAIL_IF(SCSshTxGetAlStateProgress(tx, STREAM_TOCLIENT) != SshStateBanner);
 
+    /* the failure event is published data: the tx is marked updated */
+    struct AppLayerTxData *txdata = AppLayerParserGetTxData(IPPROTO_TCP, ALPROTO_SSH, tx);
+    FAIL_IF_NULL(txdata);
+    FAIL_IF(!txdata->updated_ts);
+
     /* the failed direction is frozen: a valid banner does not advance it */
     r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOSERVER, banner, bannerlen);
     FAIL_IF(r != 0);
@@ -1727,6 +1732,11 @@ static int SSHParserTest28(void)
     tx = SCSshStateGetTx(ssh_state, 0);
     FAIL_IF(SCSshTxGetAlStateProgress(tx, STREAM_TOSERVER) != SshStateKex);
 
+    /* the failure event is published data: the tx is marked updated */
+    txdata = AppLayerParserGetTxData(IPPROTO_TCP, ALPROTO_SSH, tx);
+    FAIL_IF_NULL(txdata);
+    FAIL_IF(!txdata->updated_ts);
+
     /* the failed direction is frozen: a valid newkeys record does not
      * advance it to session */
     uint8_t newkeys[] = { 0x00, 0x00, 0x00, 0x03, 0x01, 21, 0x00 };
@@ -1742,6 +1752,152 @@ static int SSHParserTest28(void)
     PASS;
 }
 
+/** \test A record reassembly stash and a header fragment mark the tx
+ *  \test updated at parse entry, like a record completion: the flag
+ *  \test is set at the parse entry, as the base parser did, so the
+ *  \test deliveries are swept by the firewall default policy. A
+ *  \test pkt_len=29 record (27 payload bytes after the 6B header,
+ *  \test msg 50) is fed in 3 calls: 16B (header + 10), 4B (pure
+ *  \test stash), 13B (completion); then a 3B header fragment.
+ */
+static int SSHParserTest29(void)
+{
+    Flow f;
+    TcpSession ssn;
+    AppLayerParserThreadCtx *alp_tctx = NULL;
+
+    uint8_t banner[] = "SSH-2.0-Client-1.0\r\n";
+    uint8_t seg2[] = { 0x00, 0x00, 0x00, 29, 0x00, 50, 0x61, 0x62, 0x63, 0x64, 0x65, 0x66, 0x67,
+        0x68, 0x69, 0x6a };
+    uint8_t seg3[4] = { 0x6b, 0x6c, 0x6d, 0x6e };
+    uint8_t seg4[13] = { 0x6f, 0x70, 0x71, 0x72, 0x73, 0x74, 0x75, 0x76, 0x77, 0x78, 0x79, 0x7a,
+        0x7b };
+    uint8_t seg5[3] = { 0x7c, 0x7d, 0x7e };
+
+    void *tx = NULL;
+    struct AppLayerTxData *txdata = NULL;
+    int r = 0;
+
+    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);
+    alp_tctx = AppLayerParserThreadCtxAlloc();
+    FAIL_IF_NULL(alp_tctx);
+
+    r = AppLayerParserParse(
+            NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOSERVER, banner, sizeof(banner) - 1);
+    FAIL_IF(r != 0);
+    FAIL_IF_NULL(f.alstate);
+    tx = SCSshStateGetTx(f.alstate, 0);
+    FAIL_IF(SCSshTxGetAlStateProgress(tx, STREAM_TOSERVER) != SshStateKex);
+    txdata = AppLayerParserGetTxData(IPPROTO_TCP, ALPROTO_SSH, tx);
+    FAIL_IF_NULL(txdata);
+    FAIL_IF(!txdata->updated_ts);
+
+    // record header + 10 payload bytes: frames are published
+    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOSERVER, seg2, sizeof(seg2));
+    FAIL_IF(r != 0);
+    FAIL_IF(!txdata->updated_ts);
+    txdata->updated_ts = false;
+
+    // the 4B chunk is fully held by the reassembly stash: nothing is
+    // published, but the tx is marked updated at parse entry
+    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOSERVER, seg3, sizeof(seg3));
+    FAIL_IF(r != 0);
+    FAIL_IF(!txdata->updated_ts);
+
+    // the final 13B chunk completes the record
+    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOSERVER, seg4, sizeof(seg4));
+    FAIL_IF(r != 0);
+    FAIL_IF(!txdata->updated_ts);
+    txdata->updated_ts = false;
+
+    // 3B header fragment: held for the next delivery, the tx is
+    // marked updated at parse entry
+    r = AppLayerParserParse(NULL, alp_tctx, &f, ALPROTO_SSH, STREAM_TOSERVER, seg5, sizeof(seg5));
+    FAIL_IF(r != 1);
+    FAIL_IF(!txdata->updated_ts);
+    FAIL_IF(SCSshTxGetAlStateProgress(tx, STREAM_TOSERVER) != SshStateKex);
+
+    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
+ *  \test the direction to kex. */
+static int SSHParserTest32(void)
+{
+    TcpReassemblyThreadCtx *ra_ctx = NULL;
+    ThreadVars tv;
+    TcpSession ssn;
+    Flow *f = NULL;
+    Packet *p = NULL;
+
+    uint8_t part1[] = "SSH-2.0-TestClient-1.0";
+    uint8_t part2[] = "\r\n";
+    uint32_t part1len = sizeof(part1) - 1;
+    uint32_t part2len = sizeof(part2) - 1;
+
+    memset(&tv, 0x00, sizeof(tv));
+
+    StreamTcpUTInit(&ra_ctx);
+    StreamTcpUTInitInline();
+    StreamTcpUTSetupSession(&ssn);
+    StreamTcpUTSetupStream(&ssn.server, 1);
+    StreamTcpUTSetupStream(&ssn.client, 1);
+
+    f = UTHBuildFlow(AF_INET, "1.1.1.1", "2.2.2.2", 1234, 2222);
+    FAIL_IF_NULL(f);
+    f->protoctx = &ssn;
+    f->proto = IPPROTO_TCP;
+    f->alproto = ALPROTO_SSH;
+
+    p = PacketGetFromAlloc();
+    FAIL_IF_NULL(p);
+    p->proto = IPPROTO_TCP;
+    p->flow = f;
+
+    uint32_t seqcli = 2;
+    // segment 1: the open banner line; the parse consumes nothing and
+    // publishes nothing, but the tx is marked updated at parse entry
+    FAIL_IF(StreamTcpUTAddSegmentWithPayload(&tv, ra_ctx, &ssn.client, seqcli, part1, part1len) ==
+            -1);
+    seqcli += part1len;
+    FAIL_IF(StreamTcpReassembleAppLayer(&tv, ra_ctx, &ssn, &ssn.client, p, UPDATE_DIR_PACKET) < 0);
+
+    void *ssh_state = f->alstate;
+    FAIL_IF_NULL(ssh_state);
+    void *tx = SCSshStateGetTx(ssh_state, 0);
+    struct AppLayerTxData *txdata = AppLayerParserGetTxData(IPPROTO_TCP, ALPROTO_SSH, tx);
+    FAIL_IF_NULL(txdata);
+    FAIL_IF(!txdata->updated_ts);
+
+    // segment 2: the end-of-line completes the line; the banner
+    // publishes and the direction advances to kex
+    FAIL_IF(StreamTcpUTAddSegmentWithPayload(&tv, ra_ctx, &ssn.client, seqcli, part2, part2len) ==
+            -1);
+    seqcli += part2len;
+    FAIL_IF(StreamTcpReassembleAppLayer(&tv, ra_ctx, &ssn, &ssn.client, p, UPDATE_DIR_PACKET) < 0);
+
+    FAIL_IF(SCSshTxGetAlStateProgress(tx, STREAM_TOSERVER) != SshStateKex);
+    FAIL_IF(!txdata->updated_ts);
+
+    UTHFreePacket(p);
+    UTHFreeFlow(f);
+    StreamTcpUTClearSession(&ssn);
+    StreamTcpUTDeinit(ra_ctx);
+    PASS;
+}
+
 /** \test NewKeys while the peer direction failed: the peer never
  *  \test actually switched keys, so it is not terminal for the
  *  \test no-inspection decision (encryption bypass stays armed-able
@@ -2016,6 +2172,10 @@ void SSHParserRegisterTests(void)
     UtRegisterTest("SSHParserTest27 - Per-direction session state", SSHParserTest27);
     UtRegisterTest("SSHParserTest28 - failure freezes the failing direction at its state",
             SSHParserTest28);
+    UtRegisterTest(
+            "SSHParserTest29 - stash and fragment deliveries mark the tx updated", SSHParserTest29);
+    UtRegisterTest("SSHParserTest32 - banner continuation marks the tx updated at parse entry",
+            SSHParserTest32);
     UtRegisterTest("SSHParserTest33 - failed peer does not arm no-inspection", SSHParserTest33);
     UtRegisterTest(
             "SSHParserTest35 - split NewKeys arms the no-inspection/bypass flags", SSHParserTest35);