| 57 | } |
| 58 | |
| 59 | pub async fn parse_message(&self, cursor: &mut Cursor<&[u8]>) -> Result<FrontendMessage, ParseError> { |
| 60 | if cursor.position() >= cursor.get_ref().len() as u64 { |
| 61 | return Err(ParseError::BufferUnderrun); |
| 62 | } |
| 63 | |
| 64 | let message_type = cursor.read_u8()?; |
| 65 | |
| 66 | // Handle SSL request separately (no length header) |
| 67 | if message_type == 0x80 { |
| 68 | return self.parse_ssl_request(cursor).await; |
| 69 | } |
| 70 | |
| 71 | let length = cursor.read_u32::<BigEndian>()?; |
| 72 | |
| 73 | // Security check: length must be at least 4 (includes length field itself) |
| 74 | if length < 4 { |
| 75 | warn!("Message length too small: {} bytes", length); |
| 76 | events::protocol_violation(None, &format!("Message length too small: {} bytes", length)); |
| 77 | return Err(ParseError::ProtocolViolation("Message length too small".to_string())); |
| 78 | } |
| 79 | |
| 80 | // Security check: prevent DoS via huge messages |
| 81 | if length > self.max_message_size { |
| 82 | warn!("Message too large: {} bytes, max: {}", length, self.max_message_size); |
| 83 | events::protocol_violation(None, &format!("Message too large: {} bytes", length)); |
| 84 | return Err(ParseError::MessageTooLarge(length)); |
| 85 | } |
| 86 | |
| 87 | // Ensure we have enough data |
| 88 | let remaining = cursor.get_ref().len() as u64 - cursor.position(); |
| 89 | if remaining < (length - 4) as u64 { // -4 because length includes the length field |
| 90 | return Err(ParseError::BufferUnderrun); |
| 91 | } |
| 92 | |
| 93 | let payload_length = length - 4; |
| 94 | |
| 95 | match message_type { |
| 96 | b'Q' => self.parse_query(cursor, payload_length).await, |
| 97 | b'P' => self.parse_parse(cursor, payload_length).await, |
| 98 | b'B' => self.parse_bind(cursor, payload_length).await, |
| 99 | b'E' => self.parse_execute(cursor, payload_length).await, |
| 100 | b'S' => Ok(FrontendMessage::Sync), |
| 101 | b'X' => Ok(FrontendMessage::Terminate), |
| 102 | b'C' => self.parse_close(cursor, payload_length).await, |
| 103 | b'D' => self.parse_describe(cursor, payload_length).await, |
| 104 | b'H' => Ok(FrontendMessage::Flush), |
| 105 | _ => { |
| 106 | events::protocol_violation(None, &format!("Unknown message type: {}", message_type as char)); |
| 107 | Err(ParseError::InvalidMessageType(message_type)) |
| 108 | } |
| 109 | } |
| 110 | } |
| 111 | |
| 112 | pub async fn parse_startup_message(&self, cursor: &mut Cursor<&[u8]>) -> Result<StartupMessage, ParseError> { |
| 113 | let length = cursor.read_u32::<BigEndian>()?; |