diff --git a/cmd/broker/main.go b/cmd/broker/main.go index f713fd36..7b1a2e94 100644 --- a/cmd/broker/main.go +++ b/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: diff --git a/cmd/proxy/lfs.go b/cmd/proxy/lfs.go index 7252095d..a205679d 100644 --- a/cmd/proxy/lfs.go +++ b/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" diff --git a/cmd/proxy/lfs_http.go b/cmd/proxy/lfs_http.go index 8cc231c5..fe16500e 100644 --- a/cmd/proxy/lfs_http.go +++ b/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) diff --git a/cmd/proxy/logging.go b/cmd/proxy/logging.go new file mode 100644 index 00000000..79d049e4 --- /dev/null +++ b/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", "") +} diff --git a/cmd/proxy/logging_test.go b/cmd/proxy/logging_test.go new file mode 100644 index 00000000..24b74ddb --- /dev/null +++ b/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) + } +} diff --git a/cmd/proxy/main.go b/cmd/proxy/main.go index ed316321..865743a6 100644 --- a/cmd/proxy/main.go +++ b/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 {