aiden
/
suricata
mirror of https://github.com/OISF/suricata
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

pgsql: use message type for transaction actions

Browse Source 
parse_request and parse_response were using the parser state for
transaction bookkeeping.
Instead, use the message types themselves.

This change means that message types that ended up not being added to
the responses array, before (due to not advancing the parser state), now
are pushed and thus can be logged.

** Noticeable effect: AuthenticationSSPI, Notice and Notification
messages, previously missed due to falling into the None state progress
branch, could now get logged.
** To prevent affecting existing setups that don't expect these, since
they weren't logged before, we refrain from pushing them into the
responses queue, for now.

Leave state progress for state machine keeping.

Bug: #8886
pull/16213/head
Juliana Fajardini 2 months ago committed by Victor Julien
parent c18f5e6f96
commit 42098ede68
1 changed files with 74 additions and 61 deletions
Show Stats Download Patch File Download Diff File

135
rust/src/pgsql/pgsql.rs
Unescape Escape View File

@ -403,6 +403,7 @@ impl PgsqlState {
if let Some(state) = new_state { if let Some(state) = new_state {
self.state_progress = state; self.state_progress = state;
}; };
let curr_state = self.state_progress;
// PostreSQL progress states can be represented as a finite state machine // PostreSQL progress states can be represented as a finite state machine
// After the connection phase, the backend/ server will be mostly waiting in a state of `ReadyForQuery`, unless // After the connection phase, the backend/ server will be mostly waiting in a state of `ReadyForQuery`, unless
// it's processing some request. // it's processing some request.
@ -414,30 +415,30 @@ impl PgsqlState {
// https://samadhiweb.com/blog/2013.04.28.graphviz.postgresv3.html // https://samadhiweb.com/blog/2013.04.28.graphviz.postgresv3.html
if let Some(tx) = self.find_or_create_tx() { if let Some(tx) = self.find_or_create_tx() {
tx.tx_data.0.updated_ts = true; tx.tx_data.0.updated_ts = true;
if let Some(state) = new_state { if curr_state == PgsqlStateProgress::FirstCopyDataInReceived
if state == PgsqlStateProgress::FirstCopyDataInReceived || curr_state == PgsqlStateProgress::ConsolidatingCopyDataIn
|| state == PgsqlStateProgress::ConsolidatingCopyDataIn {
{ // here we're actually only counting how many messages were received.
// here we're actually only counting how many messages were received. // frontends are not forced to send one row per message
// frontends are not forced to send one row per message if let PgsqlFEMessage::ConsolidatedCopyDataIn(ref msg) = request {
if let PgsqlFEMessage::ConsolidatedCopyDataIn(ref msg) = request { tx.sum_data_size(msg.data_size);
tx.sum_data_size(msg.data_size); tx.incr_row_cnt();
tx.incr_row_cnt();
}
} else if (state == PgsqlStateProgress::CopyDoneReceived
|| state == PgsqlStateProgress::CopyFailReceived)
&& tx.get_row_cnt() > 0
{
let consolidated_copy_data = PgsqlFEMessage::ConsolidatedCopyDataIn(
ConsolidatedDataRowPacket {
identifier: b'd',
row_cnt: tx.get_row_cnt(),
data_size: tx.data_size, // total byte count of all copy_data messages combined
},
);
tx.requests.push(consolidated_copy_data);
} }
} else if (matches!(
request,
PgsqlFEMessage::CopyDone(_) | PgsqlFEMessage::CopyFail(_)
)) && tx.get_row_cnt() > 0
{
let consolidated_copy_data =
PgsqlFEMessage::ConsolidatedCopyDataIn(ConsolidatedDataRowPacket {
identifier: b'd',
row_cnt: tx.get_row_cnt(),
data_size: tx.data_size, // total byte count of all copy_data messages combined
});
tx.requests.push(consolidated_copy_data);
}
if let Some(state) = new_state {
if Self::request_is_complete(state) { if Self::request_is_complete(state) {
tx.requests.push(request); tx.requests.push(request);
// The request is complete at this point // The request is complete at this point
@ -574,6 +575,7 @@ impl PgsqlState {
_ => { _ => {
// We don't always have to change current state when we see a response... // We don't always have to change current state when we see a response...
// NotificationResponse and NoticeResponse fall here // NotificationResponse and NoticeResponse fall here
// AuthenticationSSPI also falls here
None None
} }
} }
@ -637,46 +639,57 @@ impl PgsqlState {
if tx.tx_res_state == PgsqlTxProgress::Init { if tx.tx_res_state == PgsqlTxProgress::Init {
tx.tx_res_state = PgsqlTxProgress::Received; tx.tx_res_state = PgsqlTxProgress::Received;
} }
if let Some(state) = new_state { if matches!(response, PgsqlBEMessage::ConsolidatedDataRow(_)) {
if state == PgsqlStateProgress::DataRowReceived { tx.incr_row_cnt();
tx.incr_row_cnt(); } else if matches!(response, PgsqlBEMessage::CommandComplete(_))
} else if state == PgsqlStateProgress::CommandCompletedReceived && tx.get_row_cnt() > 0
&& tx.get_row_cnt() > 0 {
{ // let's summarize the info from the data_rows in one response
// let's summarize the info from the data_rows in one response let consolidated_data_row =
let consolidated_data_row = PgsqlBEMessage::ConsolidatedDataRow( PgsqlBEMessage::ConsolidatedDataRow(ConsolidatedDataRowPacket {
ConsolidatedDataRowPacket { identifier: b'D',
identifier: b'D', row_cnt: tx.get_row_cnt(),
row_cnt: tx.get_row_cnt(), data_size: tx.data_size, // total byte count of all data_row messages combined
data_size: tx.data_size, // total byte count of all data_row messages combined });
}, tx.responses.push(consolidated_data_row);
); tx.responses.push(response);
tx.responses.push(consolidated_data_row); // reset values
tx.responses.push(response); tx.data_row_cnt = 0;
// reset values tx.data_size = 0;
tx.data_row_cnt = 0; } else if matches!(response, PgsqlBEMessage::ConsolidatedCopyDataOut(_)) {
tx.data_size = 0; tx.incr_row_cnt();
} else if state == PgsqlStateProgress::CopyDataOutReceived { } else if matches!(response, PgsqlBEMessage::CopyDone(_))
tx.incr_row_cnt(); && tx.get_row_cnt() > 0
} else if state == PgsqlStateProgress::CopyDoneReceived {
&& tx.get_row_cnt() > 0 // let's summarize the info from the data_rows in one response
{ let consolidated_copy_data = PgsqlBEMessage::ConsolidatedCopyDataOut(
// let's summarize the info from the data_rows in one response ConsolidatedDataRowPacket {
let consolidated_copy_data = identifier: b'd',
PgsqlBEMessage::ConsolidatedCopyDataOut( row_cnt: tx.get_row_cnt(),
ConsolidatedDataRowPacket { data_size: tx.data_size, // total byte count of all data_row messages combined
identifier: b'd', },
row_cnt: tx.get_row_cnt(), );
data_size: tx.data_size, // total byte count of all data_row messages combined tx.responses.push(consolidated_copy_data);
}, tx.responses.push(response);
); // reset values
tx.responses.push(consolidated_copy_data); tx.data_row_cnt = 0;
tx.responses.push(response); tx.data_size = 0;
// reset values } else {
tx.data_row_cnt = 0; if !matches!(
tx.data_size = 0; response,
} else { PgsqlBEMessage::UnknownMessageType(_)
| PgsqlBEMessage::NoticeResponse(_)
| PgsqlBEMessage::NotificationResponse(_)
| PgsqlBEMessage::AuthenticationSSPI(_)
) {
// since we are not logging UnknownMessages right now, let's not push them into the responses
// vector, either
// Don't log NoticeResponse, NotificationResponse, nor AuthenticationSSPI messages,
// either (temporarily)
tx.responses.push(response); tx.responses.push(response);
}
if let Some(state) = new_state {
if Self::response_is_complete(state) { if Self::response_is_complete(state) {
tx.tx_req_state = PgsqlTxProgress::Done; tx.tx_req_state = PgsqlTxProgress::Done;
tx.tx_res_state = PgsqlTxProgress::Done; tx.tx_res_state = PgsqlTxProgress::Done;

Write Preview
Loading…
Cancel
Save