fix(mysqlx): four minor security findings from issue #5676

Addresses the four Minor / Nit findings deferred when the Important
re-auth bug was fixed in commit 09c15d6d5.

## (1) Reject TLS upgrade post-auth or on encrypted channel

`handler_capabilities_set` previously honored `tls=true` regardless
of session state, so a logged-in client could ship CapabilitiesSet
with tls=true and the server would dutifully start an SSL handshake
on an already-authenticated session — or, worse, a second handshake
on an already-encrypted channel that left the state machine in
"expecting fresh handshake" while the client kept its existing TLS
session. Both cases are spec violations (the X Protocol forbids TLS
negotiation post-AuthenticateOk) and produce a confused-client kill.

Fix: reject with code 5052 (FATAL) when client_ds_.is_encrypted() is
true OR status_ has progressed past the pre-auth states. The TLS
branch only runs from CONNECTING_CLIENT / X_CAPABILITIES_GET /
X_CAPABILITIES_SET now.

## (2) Cap leading-NOTICE drain in backend auth

`read_auth_frame` loops while the backend ships NOTICE frames, on
the assumption that any legitimate backend emits at most a couple
during auth. A compromised or malicious backend could pin the worker
thread by streaming NOTICEs continuously. Add a cap of 64 leading
notices (MAX_LEADING_NOTICES); on the 65th, treat as auth failure
(BACKEND_AUTH_ERROR) and let the chassis tear the session down.

## (3) Per-session pre-auth CapabilitiesGet/Set replay cap

A client can reset back to CONNECTING_CLIENT after each
CapabilitiesGet/Set without ever authenticating, and each iteration
allocates a fresh protobuf and runs the full handler. Without a cap
this is a free CPU/memory amplifier (bounded only by the OS recv
buffer). Add a per-session counter (pre_auth_cap_msgs_) bumped on
every CapabilitiesGet/Set seen while status_ != WAITING_CLIENT_XMSG.
Threshold MAX_PRE_AUTH_CAP_MSGS = 64 (legitimate clients send 1-3);
on overflow, send 5008 + healthy=false. Counter resets on session
reset() and is bypassed once auth completes (post-auth introspection
of capabilities is legitimate and unbounded).

## (4) MysqlxDataStream::enqueue_frame: surface oversize as parse error

`if (body_len + 1 > X_MAX_PAYLOAD_SIZE) return;` previously dropped
oversize bodies silently, leaving the client expecting a frame that
never arrives. Set parse_error_=true so the session loop tears the
connection down cleanly. Reaching this branch is an internal bug
(server-side senders are responsible for never producing a >16MiB
body), but a clean teardown is materially better than a half-working
session.

## Verified

After this commit (PROXYSQLGENAI=1 NOJEMALLOC=1 WITHASAN=1):
- mysqlx_session_unit-t        60 / 0 (was 60 / 0)
- mysqlx_robustness_unit-t     74 / 0 (was 74 / 0)
- mysqlx_compression_unit-t    64 / 0 (was 64 / 0)
- mysqlx_protocol_unit-t       42 / 0 (was 42 / 0)
- mysqlx_data_stream_unit-t    18 / 0 (was 18 / 0)
- ASAN: 0 errors

These are defense-in-depth changes; none addresses an exploitable
crash. All five tests pass with no regressions.

plugins/mysqlx/include/mysqlx_session.h

@@ -279,6 +279,16 @@ private:
 	std::vector<uint8_t> compress_batch_framed_;
 	uint32_t compress_batch_count_;
 
+	// Pre-auth capability-message replay counter. A hostile or buggy
+	// client can replay CapabilitiesGet / CapabilitiesSet arbitrarily
+	// many times before authenticating; each one runs through protobuf
+	// parsing and a small allocation. Bound the count per session so
+	// the path is not a free CPU/memory amplifier. Reset to zero on
+	// successful auth; bumps on every CapabilitiesGet/Set seen while
+	// status_ is still pre-auth (CONNECTING_CLIENT / X_CAPABILITIES_*).
+	uint32_t pre_auth_cap_msgs_;
+	static constexpr uint32_t MAX_PRE_AUTH_CAP_MSGS = 64;
+
 public:
 	// Test-only accessors for compression negotiation outcome.
 	MysqlxCompressionAlgo compression_algo_for_test() const { return compression_algo_; }

plugins/mysqlx/src/mysqlx_connection.cpp

@@ -140,10 +140,23 @@ std::optional<MysqlxFrame> MysqlxConnection::read_auth_frame() {
 	// emit a session-state-change notice before AuthenticateContinue / Ok;
 	// returning nullopt on a NOTICE without re-trying caused the auth state
 	// machine to spin until the 10s handshake timeout.
+	//
+	// Cap the NOTICE drain so a misbehaving or hostile backend can't pin
+	// this worker thread by streaming NOTICEs forever. 64 is more than any
+	// legitimate backend would emit during auth — typical is 1-2.
+	constexpr int MAX_LEADING_NOTICES = 64;
+	int notice_count = 0;
 	while (true) {
 		auto frame = backend_ds_.try_read_one_frame();
 		if (!frame) return std::nullopt;
 		if (frame->size() >= 5 && (*frame)[4] == Mysqlx::ServerMessages_Type_NOTICE) {
+			if (++notice_count > MAX_LEADING_NOTICES) {
+				// Treat as auth failure; the caller will mark the session
+				// unhealthy and the chassis will return the connection to
+				// the pool / drop it.
+				auth_state_ = BACKEND_AUTH_ERROR;
+				return std::nullopt;
+			}
 			continue;
 		}
 		return frame;

plugins/mysqlx/src/mysqlx_data_stream.cpp

@@ -135,7 +135,19 @@ void MysqlxDataStream::pop_frame() {
 }
 
 void MysqlxDataStream::enqueue_frame(uint8_t msg_type, const uint8_t* body, size_t body_len) {
-	if (body_len + 1 > X_MAX_PAYLOAD_SIZE) return;
+	if (body_len + 1 > X_MAX_PAYLOAD_SIZE) {
+		// Server attempted to enqueue a frame larger than the X-Protocol
+		// 16 MiB cap. Silently dropping the body would leave the client
+		// expecting a frame that never arrives — usually a protocol
+		// stall. Mark the parser as broken so the session loop tears
+		// down cleanly instead of half-working. The caller (always a
+		// server-side path: send_error, send_capabilities, compression
+		// emit) is responsible for never feeding bodies anywhere near
+		// this size; reaching here means an internal bug, not a
+		// protocol-level event.
+		parse_error_ = true;
+		return;
+	}
 	uint32_t payload_size = static_cast<uint32_t>(body_len) + 1;
 	write_buf_.push_back(static_cast<uint8_t>(payload_size & 0xFF));
 	write_buf_.push_back(static_cast<uint8_t>((payload_size >> 8) & 0xFF));

plugins/mysqlx/src/mysqlx_session.cpp

@@ -104,7 +104,8 @@ MysqlxSession::MysqlxSession()
 	, compression_max_combine_messages_(0)
 	, zstd_dctx_(nullptr)
 	, zstd_cctx_(nullptr)
-	, compress_batch_count_(0) {
+	, compress_batch_count_(0)
+	, pre_auth_cap_msgs_(0) {
 }
 
 MysqlxSession::~MysqlxSession() {
@@ -167,6 +168,7 @@ void MysqlxSession::reset() {
 	compression_combine_mixed_messages_ = false;
 	compression_max_combine_messages_ = 0;
 	reset_compression_state();
+	pre_auth_cap_msgs_ = 0;
 }
 
 int MysqlxSession::handler() {
@@ -265,6 +267,22 @@ void MysqlxSession::handler_connecting_client() {
 void MysqlxSession::handler_capabilities_get() {
 	if (!client_ds_.has_complete_frame()) return;
 
+	// Bound how many pre-auth capability messages a client can ship per
+	// session. Each one runs through frame parsing + send_capabilities()
+	// allocations; without a cap, an idle hostile client can pin a
+	// worker on the cap-replay path forever. The counter applies only
+	// while still pre-auth; post-auth callers (dispatch_client_message
+	// → CON_CAPABILITIES_GET) routinely query capabilities and should
+	// not trip the bound. status_ == WAITING_CLIENT_XMSG indicates auth
+	// completed.
+	if (status_ != WAITING_CLIENT_XMSG &&
+	    ++pre_auth_cap_msgs_ > MAX_PRE_AUTH_CAP_MSGS) {
+		client_ds_.pop_frame();
+		send_error(5008, "Too many pre-auth capability messages", true);
+		healthy = false;
+		return;
+	}
+
 	client_ds_.pop_frame();
 	send_capabilities();
 	status_ = CONNECTING_CLIENT;
@@ -352,6 +370,16 @@ static bool parse_compression_capability(
 void MysqlxSession::handler_capabilities_set() {
 	if (!client_ds_.has_complete_frame()) return;
 
+	// Same per-session bound as handler_capabilities_get(); see comment
+	// there. Counter only applies pre-auth.
+	if (status_ != WAITING_CLIENT_XMSG &&
+	    ++pre_auth_cap_msgs_ > MAX_PRE_AUTH_CAP_MSGS) {
+		client_ds_.pop_frame();
+		send_error(5008, "Too many pre-auth capability messages", true);
+		healthy = false;
+		return;
+	}
+
 	const auto& frame = client_ds_.front_frame();
 	if (frame.size() > 5) {
 		Mysqlx::Connection::CapabilitiesSet cap_set;
@@ -398,6 +426,26 @@ void MysqlxSession::handler_capabilities_set() {
 		for (const auto& cap : cap_set.capabilities().capabilities()) {
 			if (cap.name() == "tls") {
 				client_ds_.pop_frame();
+				// Reject TLS upgrade post-auth or on an already-encrypted
+				// channel. The X Protocol forbids TLS negotiation after
+				// AuthenticateOk, and a second tls=true on an already-TLS
+				// channel would desync the state machine (the server
+				// expects a fresh handshake; the client expects to keep
+				// the existing TLS session). Either case is a hostile or
+				// confused client; respond with 5052 and drop the
+				// session before driving SSL_do_handshake.
+				if (client_ds_.is_encrypted()) {
+					send_error(5052, "TLS already negotiated on this session", true);
+					healthy = false;
+					return;
+				}
+				if (status_ != CONNECTING_CLIENT &&
+				    status_ != X_CAPABILITIES_GET &&
+				    status_ != X_CAPABILITIES_SET) {
+					send_error(5052, "TLS negotiation not allowed after authentication", true);
+					healthy = false;
+					return;
+				}
 				SSL_CTX* ctx = thread_ptr_ ? thread_ptr_->get_ssl_ctx() : nullptr;
 				if (!ctx) {
 					send_error(3150, "TLS is not configured on server");