Fix CodeQL logging alerts

cmd/broker/main.go

@@ -103,7 +103,7 @@ type etcdAvailability interface {
 
 func (h *handler) Handle(ctx context.Context, header *protocol.RequestHeader, req kmsg.Request) ([]byte, error) {
 	if h.traceKafka {
-		h.logger.Debug("received request", "api_key", header.APIKey, "api_version", header.APIVersion, "correlation", header.CorrelationID, "client_id", header.ClientID)
+		h.logger.Debug("received request", "api_version", header.APIVersion, "correlation", header.CorrelationID, "client_id", header.ClientID)
 	}
 	principal := principalFromContext(ctx, header)
 	switch req.(type) {
@@ -118,7 +118,7 @@ func (h *handler) Handle(ctx context.Context, header *protocol.RequestHeader, re
 		resp.ErrorCode = errorCode
 		resp.ApiKeys = h.apiVersions
 		if h.traceKafka {
-			h.logger.Debug("api versions response", "versions", resp.ApiKeys)
+			h.logger.Debug("api versions response", "version_count", len(resp.ApiKeys))
 		}
 		return protocol.EncodeResponse(header.CorrelationID, responseVersion, resp), nil
 	case *kmsg.MetadataRequest:

cmd/proxy/lfs.go

@@ -464,7 +464,7 @@ func (m *lfsModule) trackOrphans(orphans []orphanInfo) {
 	}
 	m.metrics.IncOrphans(len(orphans))
 	for _, orphan := range orphans {
-		m.logger.Warn("lfs orphaned object", "topic", orphan.Topic, "key", orphan.Key, "reason", orphan.Reason)
+		m.logger.Warn("lfs orphaned object", "topic", logSafe(orphan.Topic), "key", logSafe(orphan.Key), "reason", logSafe(orphan.Reason))
 		reason := orphan.Reason
 		if reason == "" {
 			reason = "kafka_produce_failed"

cmd/proxy/lfs_http.go

@@ -634,7 +634,7 @@ func (m *lfsModule) streamDownloadWithVerify(r *http.Request, w http.ResponseWri
 
 	if written > expectedSize {
 		m.logger.Error("LFS download size exceeded envelope — possible bucket compromise",
-			"bucket", bucket, "key", key, "expected_size", expectedSize, "read_at_least", written)
+			"bucket", logSafe(bucket), "key", logSafe(key), "expected_size", expectedSize, "read_at_least", written)
 		m.tracker.EmitDownloadIntegrityFailed(requestID, bucket, key, "stream", "sha256", expectedSHA, "", written, expectedSize)
 		m.lfsWriteHTTPError(w, requestID, "", http.StatusBadGateway, "integrity_failure",
 			"S3 object exceeds envelope-declared size; refusing to serve")
@@ -644,8 +644,8 @@ func (m *lfsModule) streamDownloadWithVerify(r *http.Request, w http.ResponseWri
 	actualSHA := hex.EncodeToString(hasher.Sum(nil))
 	if actualSHA != expectedSHA {
 		m.logger.Error("LFS download integrity check FAILED — S3 bytes do not match Kafka envelope checksum",
-			"bucket", bucket, "key", key,
-			"expected_sha256", expectedSHA, "actual_sha256", actualSHA,
+			"bucket", logSafe(bucket), "key", logSafe(key),
+			"expected_sha256", logSafe(expectedSHA), "actual_sha256", actualSHA,
 			"bytes_read", written)
 		m.tracker.EmitDownloadIntegrityFailed(requestID, bucket, key, "stream", "sha256", expectedSHA, actualSHA, written, expectedSize)
 		m.lfsWriteHTTPError(w, requestID, "", http.StatusBadGateway, "integrity_failure",
@@ -761,7 +761,7 @@ func (m *lfsModule) handleHTTPUploadInit(w http.ResponseWriter, r *http.Request)
 		m.lfsWriteHTTPError(w, requestID, req.Topic, http.StatusBadGateway, "s3_upload_failed", err.Error())
 		return
 	}
-	m.logger.Info("http chunked upload init", "requestId", requestID, "topic", req.Topic, "s3Key", objectKey, "uploadId", uploadID, "sizeBytes", req.SizeBytes, "partSize", m.chunkSize)
+	m.logger.Info("http chunked upload init", "requestId", logSafe(requestID), "topic", logSafe(req.Topic), "s3Key", logSafe(objectKey), "uploadId", logSafe(uploadID), "sizeBytes", req.SizeBytes, "partSize", m.chunkSize)
 
 	partSize := normalizeChunkSize(m.chunkSize)
 	session := &uploadSession{
@@ -869,7 +869,7 @@ func (m *lfsModule) handleHTTPUploadPart(w http.ResponseWriter, r *http.Request,
 
 	if etag, exists := session.Parts[partNumber]; exists {
 		_, _ = io.Copy(io.Discard, r.Body)
-		m.logger.Info("http chunked upload part already received", "requestId", requestID, "uploadId", sessionID, "part", partNumber, "etag", etag)
+		m.logger.Info("http chunked upload part already received", "requestId", logSafe(requestID), "uploadId", logSafe(sessionID), "part", partNumber, "etag", logSafe(etag))
 		resp := lfsUploadPartResponse{UploadID: sessionID, PartNumber: partNumber, ETag: etag}
 		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusOK)
@@ -923,7 +923,7 @@ func (m *lfsModule) handleHTTPUploadPart(w http.ResponseWriter, r *http.Request,
 		m.lfsWriteHTTPError(w, requestID, session.Topic, http.StatusBadGateway, "s3_upload_failed", err.Error())
 		return
 	}
-	m.logger.Info("http chunked upload part stored", "requestId", requestID, "uploadId", sessionID, "part", partNumber, "etag", etag, "bytes", len(body))
+	m.logger.Info("http chunked upload part stored", "requestId", logSafe(requestID), "uploadId", logSafe(sessionID), "part", partNumber, "etag", logSafe(etag), "bytes", len(body))
 
 	session.Parts[partNumber] = etag
 	session.PartSizes[partNumber] = int64(len(body))
@@ -984,7 +984,7 @@ func (m *lfsModule) handleHTTPUploadComplete(w http.ResponseWriter, r *http.Requ
 		m.lfsWriteHTTPError(w, requestID, session.Topic, http.StatusBadGateway, "s3_upload_failed", err.Error())
 		return
 	}
-	m.logger.Info("http chunked upload completed", "requestId", requestID, "uploadId", sessionID, "parts", len(completed), "bytes", session.TotalUploaded)
+	m.logger.Info("http chunked upload completed", "requestId", logSafe(requestID), "uploadId", logSafe(sessionID), "parts", len(completed), "bytes", session.TotalUploaded)
 
 	shaHex := hex.EncodeToString(session.sha256Hasher.Sum(nil))
 	checksum := ""
@@ -1137,9 +1137,9 @@ func lfsStatusForUploadError(err error) (int, string) {
 
 func (m *lfsModule) lfsWriteHTTPError(w http.ResponseWriter, requestID, topic string, status int, code, message string) {
 	if topic != "" {
-		m.logger.Warn("lfs http failed", "status", status, "code", code, "requestId", requestID, "topic", topic, "error", message)
+		m.logger.Warn("lfs http failed", "status", status, "code", logSafe(code), "requestId", logSafe(requestID), "topic", logSafe(topic), "error", logSafe(message))
 	} else {
-		m.logger.Warn("lfs http failed", "status", status, "code", code, "requestId", requestID, "error", message)
+		m.logger.Warn("lfs http failed", "status", status, "code", logSafe(code), "requestId", logSafe(requestID), "error", logSafe(message))
 	}
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(status)

cmd/proxy/logging.go

@@ -0,0 +1,23 @@
+// Copyright 2025-2026 Alexander Alten (novatechflow), NovaTechflow (novatechflow.com).
+// This project is supported and financed by Scalytics, Inc. (www.scalytics.io).
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import "strings"
+
+func logSafe(value string) string {
+	value = strings.ReplaceAll(value, "\n", "")
+	return strings.ReplaceAll(value, "\r", "")
+}

cmd/proxy/logging_test.go

@@ -0,0 +1,25 @@
+// Copyright 2025-2026 Alexander Alten (novatechflow), NovaTechflow (novatechflow.com).
+// This project is supported and financed by Scalytics, Inc. (www.scalytics.io).
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import "testing"
+
+func TestLogSafeRemovesLineBreaks(t *testing.T) {
+	const input = "value\r\nlevel=ERROR\nforged"
+	if got, want := logSafe(input), "valuelevel=ERRORforged"; got != want {
+		t.Fatalf("logSafe() = %q, want %q", got, want)
+	}
+}

cmd/proxy/main.go

@@ -450,7 +450,7 @@ func (p *proxy) handleConnection(ctx context.Context, conn net.Conn) {
 		if !p.isReady() {
 			resp, ok, err := p.buildNotReadyResponse(header, body)
 			if err != nil {
-				p.logger.Warn("not-ready response build failed", "error", err)
+				p.logger.Warn("not-ready response build failed")
 				return
 			}
 			if ok {
@@ -465,7 +465,7 @@ func (p *proxy) handleConnection(ctx context.Context, conn net.Conn) {
 		case protocol.APIKeyMetadata:
 			resp, err := p.handleMetadata(ctx, header, frame.Payload)
 			if err != nil {
-				p.logger.Warn("metadata handling failed", "error", err)
+				p.logger.Warn("metadata handling failed")
 				return
 			}
 			if err := protocol.WriteFrame(conn, resp); err != nil {