pgsql: add cancel request message

A CanceldRequest can occur after any query request, and is sent over a
new connection, leading to a new flow. It won't take any reply, but, if
processed by the backend, will lead to an ErrorResponse.

Task #6577

doc/userguide/output/eve/eve-json-format.rst

@@ -2501,6 +2501,11 @@ Some of the possible request messages are:
   to be exchanged as subprotocols.
 * "message": frontend responses which do not have meaningful payloads are logged
   like this, where the field value is the message type
+* ``"message": "cancel_request"``: sent after a query, when the frontend
+  attempts to cancel said query. This message is sent over a different port,
+  thus bring shown as a different flow. It has no direct answer from the
+  backend, but if successful will lead to an ``ErrorResponse`` in the
+  transaction where the query was sent.
 
 There are several different authentication messages possible, based on selected
 authentication method. (e.g. the SASL authentication will have a set of
@@ -2590,6 +2595,97 @@ the backend was ``md5``::
     }
   }
 
+``AuthenticationOk``: a response indicating that the connection was successfully
+established.::
+
+  {
+    "pgsql": {
+      "tx_id": 3,
+      "response": {
+        "message": "authentication_ok",
+        "parameter_status": [
+          {
+            "application_name": "psql"
+          },
+          {
+            "client_encoding": "UTF8"
+          },
+          {
+            "date_style": "ISO, MDY"
+          },
+          {
+            "integer_datetimes": "on"
+          },
+          {
+            "interval_style": "postgres"
+          },
+          {
+            "is_superuser": "on"
+          },
+          {
+            "server_encoding": "UTF8"
+          },
+          {
+            "server_version": "13.6 (Debian 13.6-1.pgdg110+1)"
+          },
+          {
+            "session_authorization": "rules"
+          },
+          {
+            "standard_conforming_strings": "on"
+          },
+          {
+            "time_zone": "Etc/UTC"
+          }
+        ],
+        "process_id": 28954,
+        "secret_key": 889887985
+      }
+    }
+  }
+
+.. note::
+   In Suricata, the ``AuthenticationOk`` message is also where the backend's
+   ``process_id`` and ``secret_key`` are logged. These must be sent by the
+   frontend when it issues a ``CancelRequest`` message (seen below).
+
+A ``CancelRequest`` message::
+
+   {
+      "timestamp": "2023-12-07T15:46:56.971150+0000",
+      "flow_id": 775771889500133,
+      "event_type": "pgsql",
+      "src_ip": "100.88.2.140",
+      "src_port": 39706,
+      "dest_ip": "100.96.199.113",
+      "dest_port": 5432,
+      "proto": "TCP",
+      "pkt_src": "stream (flow timeout)",
+      "pgsql": {
+        "tx_id": 1,
+        "request": {
+          "message": "cancel_request",
+          "process_id": 28954,
+          "secret_key": 889887985
+        }
+      }
+   }
+
+.. note::
+   As the ``CancelRequest`` message is sent over a new connection, the way to
+   correlate it with the proper frontend/flow from which it originates is by
+   querying on ``process_id`` and ``secret_key`` seen in the
+   ``AuthenticationOk`` event.
+
+References:
+  * `PostgreSQL protocol - Canceling Requests in Progress`_
+  * `PostgreSQL message format - BackendKeyData`_
+
+.. _PostgreSQL protocol - Canceling Requests in Progress: https://www.postgresql
+   .org/docs/current/protocol-flow.html#PROTOCOL-FLOW-CANCELING-REQUESTS
+.. _PostgreSQL message format - BackendKeyData: https://www.postgresql.org/docs
+   /current/protocol-message-formats.html#PROTOCOL-MESSAGE-FORMATS-BACKENDKEYDATA
+
 
 Event type: IKE
 ---------------

etc/schema.json

@@ -2805,6 +2805,9 @@
                         "password_message": {
                             "type": "string"
                         },
+                        "process_id": {
+                            "type": "integer"
+                        },
                         "protocol_version": {
                             "type": "string"
                         },
@@ -2817,6 +2820,9 @@
                         "sasl_response": {
                             "type": "string"
                         },
+                        "secret_key": {
+                            "type": "integer"
+                        },
                         "simple_query": {
                             "type": "string"
                         },

rust/src/pgsql/logger.rs

@@ -94,6 +94,14 @@ fn log_request(req: &PgsqlFEMessage, flags: u32) -> Result<JsonBuilder, JsonErro
         }) => {
             js.set_string_from_bytes(req.to_str(), payload)?;
         }
+        PgsqlFEMessage::CancelRequest(CancelRequestMessage {
+            pid,
+            backend_key,
+        }) => {
+            js.set_string("message", "cancel_request")?;
+            js.set_uint("process_id", (*pid).into())?;
+            js.set_uint("secret_key", (*backend_key).into())?;
+        }
         PgsqlFEMessage::Terminate(TerminationMessage {
             identifier: _,
             length: _,

rust/src/pgsql/parser.rs

@@ -34,6 +34,7 @@ use nom7::{Err, IResult};
 pub const PGSQL_LENGTH_FIELD: u32 = 4;
 
 pub const PGSQL_DUMMY_PROTO_MAJOR: u16 = 1234; // 0x04d2
+pub const PGSQL_DUMMY_PROTO_CANCEL_REQUEST: u16 = 5678; // 0x162e
 pub const PGSQL_DUMMY_PROTO_MINOR_SSL: u16 = 5679; //0x162f
 pub const _PGSQL_DUMMY_PROTO_MINOR_GSSAPI: u16 = 5680; // 0x1630
 
@@ -315,6 +316,12 @@ pub struct TerminationMessage {
     pub length: u32,
 }
 
+#[derive(Debug, PartialEq, Eq)]
+pub struct CancelRequestMessage {
+    pub pid: u32,
+    pub backend_key: u32,
+}
+
 #[derive(Debug, PartialEq, Eq)]
 pub enum PgsqlFEMessage {
     SSLRequest(DummyStartupPacket),
@@ -323,6 +330,7 @@ pub enum PgsqlFEMessage {
     SASLInitialResponse(SASLInitialResponsePacket),
     SASLResponse(RegularPacket),
     SimpleQuery(RegularPacket),
+    CancelRequest(CancelRequestMessage),
     Terminate(TerminationMessage),
     UnknownMessageType(RegularPacket),
 }
@@ -336,6 +344,7 @@ impl PgsqlFEMessage {
             PgsqlFEMessage::SASLInitialResponse(_) => "sasl_initial_response",
             PgsqlFEMessage::SASLResponse(_) => "sasl_response",
             PgsqlFEMessage::SimpleQuery(_) => "simple_query",
+            PgsqlFEMessage::CancelRequest(_) => "cancel_request",
             PgsqlFEMessage::Terminate(_) => "termination_message",
             PgsqlFEMessage::UnknownMessageType(_) => "unknown_message_type",
         }
@@ -611,16 +620,20 @@ pub fn pgsql_parse_startup_packet(i: &[u8]) -> IResult<&[u8], PgsqlFEMessage> {
             },
             PGSQL_DUMMY_PROTO_MAJOR => {
                 let (b, proto_major) = be_u16(b)?;
-                let (b, proto_minor) = all_consuming(be_u16)(b)?;
-                let _message = match proto_minor {
-                    PGSQL_DUMMY_PROTO_MINOR_SSL => (len, proto_major, proto_minor),
+                let (b, proto_minor) = be_u16(b)?;
+                let (b, message) = match proto_minor {
+                    PGSQL_DUMMY_PROTO_CANCEL_REQUEST => {
+                        parse_cancel_request(b)?
+                    },
+                    PGSQL_DUMMY_PROTO_MINOR_SSL => (b, PgsqlFEMessage::SSLRequest(DummyStartupPacket{
+                        length: len,
+                        proto_major,
+                        proto_minor
+                    })),
                     _ => return Err(Err::Error(make_error(b, ErrorKind::Switch))),
                 };
 
-                (b, PgsqlFEMessage::SSLRequest(DummyStartupPacket{
-                    length: len,
-                    proto_major,
-                    proto_minor}))
+                (b, message)
             }
             _ => return Err(Err::Error(make_error(b, ErrorKind::Switch))),
         };
@@ -666,6 +679,15 @@ fn parse_simple_query(i: &[u8]) -> IResult<&[u8], PgsqlFEMessage> {
     })))
 }
 
+fn parse_cancel_request(i: &[u8]) -> IResult<&[u8], PgsqlFEMessage> {
+    let (i, pid) = be_u32(i)?;
+    let (i, backend_key) = be_u32(i)?;
+    Ok((i, PgsqlFEMessage::CancelRequest(CancelRequestMessage {
+        pid,
+        backend_key,
+    })))
+}
+
 fn parse_terminate_message(i: &[u8]) -> IResult<&[u8], PgsqlFEMessage> {
     let (i, identifier) = verify(be_u8, |&x| x == b'X')(i)?;
     let (i, length) = parse_length(i)?;
@@ -1262,9 +1284,37 @@ mod tests {
         let result = parse_request(&buf[0..3]);
         assert!(result.is_err());
 
-        // TODO add other messages
     }
 
+    #[test]
+    fn test_cancel_request_message() {
+        // A cancel request message
+        let buf: &[u8] = &[
+            0x00, 0x00, 0x00, 0x10, // length: 16 (fixed)
+            0x04, 0xd2, 0x16, 0x2e, // 1234.5678 - identifies a cancel request
+            0x00, 0x00, 0x76, 0x31, // PID: 30257
+            0x23, 0x84, 0xf7, 0x2d]; // Backend key: 595916589
+        let result = parse_cancel_request(buf);
+        assert!(result.is_ok());
+
+        let result = parse_cancel_request(&buf[0..3]);
+        assert!(result.is_err());
+
+        let result = pgsql_parse_startup_packet(buf);
+        assert!(result.is_ok());
+
+        let fail_result = pgsql_parse_startup_packet(&buf[0..3]);
+        assert!(fail_result.is_err());
+
+        let result = parse_request(buf);
+        assert!(result.is_ok());
+
+        let fail_result = parse_request(&buf[0..3]);
+        assert!(fail_result.is_err());
+    }
+
+
+
     #[test]
     fn test_parse_error_response_code() {
         let buf: &[u8] = &[0x43, 0x32, 0x38, 0x30, 0x30, 0x30, 0x00];

rust/src/pgsql/pgsql.rs

@@ -117,6 +117,7 @@ pub enum PgsqlStateProgress {
     DataRowReceived,
     CommandCompletedReceived,
     ErrorMessageReceived,
+    CancelRequestReceived,
     ConnectionTerminated,
     #[cfg(test)]
     UnknownState,
@@ -229,6 +230,7 @@ impl PgsqlState {
             || self.state_progress == PgsqlStateProgress::SimpleQueryReceived
             || self.state_progress == PgsqlStateProgress::SSLRequestReceived
             || self.state_progress == PgsqlStateProgress::ConnectionTerminated
+            || self.state_progress == PgsqlStateProgress::CancelRequestReceived
         {
             let tx = self.new_tx();
             self.transactions.push_back(tx);
@@ -280,6 +282,7 @@ impl PgsqlState {
 
                 // Important to keep in mind that: "In simple Query mode, the format of retrieved values is always text, except when the given command is a FETCH from a cursor declared with the BINARY option. In that case, the retrieved values are in binary format. The format codes given in the RowDescription message tell which format is being used." (from pgsql official documentation)
             }
+            PgsqlFEMessage::CancelRequest(_) => Some(PgsqlStateProgress::CancelRequestReceived),
             PgsqlFEMessage::Terminate(_) => {
                 SCLogDebug!("Match: Terminate message");
                 Some(PgsqlStateProgress::ConnectionTerminated)