Merge branch 'staging' into test/integration-test-kafka

Author: saksham-datazip

.trivyignore

@@ -2,4 +2,10 @@ CVE-2025-30065
 CVE-2018-1320
 CVE-2019-0205
 CVE-2020-13949
-CVE-2025-46762
+CVE-2025-46762
+# CVE-2026-34040: Moby AuthZ plugin bypass via oversized request bodies.
+# TODO: Upgrade testcontainers-go once this PR is merged and released:
+# https://github.com/testcontainers/testcontainers-go/pull/3617
+# This vulnerability only affects Docker AuthZ plugins (not used by Olake)
+# and is only present in test-time dependencies.
+CVE-2026-34040

constants/state_version.go

@@ -26,11 +26,15 @@ package constants
 //     * Earlier if the session timezone or global was set in offset format, it was not parsed correctly and used to fallback to UTC.
 //     * Now it parses the offset correctly and uses the timezone offset to set the timezone for the connection.
 //
-//   - Version 4: (Current Version) Unsigned int/integer/bigint map to Int64.
+//   - Version 4: Unsigned int/integer/bigint map to Int64.
 //     * Earlier unsigned int/integer/bigint were mapped to Int32 which caused integer overflows.
+//
+//   - Version 5: (Current Version) MongoDB nested DateTime values decoded as UTC time.Time.
+//     * BSON DateTime at any depth is now decoded directly to time.Time (UTC) via a custom client registry, preventing json.Marshal crashes for out-of-range years ([0,9999]).
+//     * Top-level DateTime fields that previously formatted with the local machine timezone (e.g. "+05:30") now always output UTC ("Z").
 
 const (
-	LatestStateVersion = 4
+	LatestStateVersion = 5
 )
 
 // Used as the current version of the state when the program is running

drivers/mongodb/internal/mon.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"math"
 	"net"
+	"reflect"
 	"strings"
 	"sync"
 	"time"
@@ -16,12 +17,65 @@ import (
 	"github.com/datazip-inc/olake/utils/logger"
 	"github.com/datazip-inc/olake/utils/typeutils"
 	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/bsoncodec"
+	"go.mongodb.org/mongo-driver/bson/bsonrw"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
 	"golang.org/x/crypto/ssh"
 )
 
+// safeDecodeRegistry is a BSON decode registry registered on the MongoDB client.
+// It intercepts two problem cases for interface{} slots (every value in bson.M):
+//
+//  1. BSON Double (float64) — NaN and ±Inf are not valid JSON; they crash
+//     encoding/json.  These are coerced to nil for all state versions.
+//
+//  2. BSON DateTime — primitive.DateTime.MarshalJSON calls time.Time.MarshalJSON
+//     which panics for years outside [0, 9999].  For state version > 4, DateTime
+//     is decoded as a clamped UTC time.Time so downstream json.Marshal is always
+//     safe.  For state version ≤ 4 the fallback (primitive.DateTime) is used to
+//     preserve the pre-existing local-timezone output format.
+var safeDecodeRegistry = func() *bsoncodec.Registry {
+	tEmpty := reflect.TypeOf((*interface{})(nil)).Elem()
+	reg := bson.NewRegistry()
+	// Capture the stock interface{} decoder before replacing it.
+	fallback, _ := reg.LookupDecoder(tEmpty)
+	reg.RegisterTypeDecoder(
+		tEmpty,
+		bsoncodec.ValueDecoderFunc(func(dc bsoncodec.DecodeContext, vr bsonrw.ValueReader, val reflect.Value) error {
+			switch vr.Type() {
+			case bson.TypeDouble:
+				f, err := vr.ReadDouble()
+				if err != nil {
+					return err
+				}
+				if math.IsNaN(f) || math.IsInf(f, 0) {
+					val.Set(reflect.Zero(val.Type()))
+				} else {
+					val.Set(reflect.ValueOf(f))
+				}
+				return nil
+			case bson.TypeDateTime:
+				if constants.LoadedStateVersion > 4 {
+					ms, err := vr.ReadDateTime()
+					if err != nil {
+						return err
+					}
+					t, err := typeutils.ReformatDate(primitive.DateTime(ms).Time().UTC(), true)
+					if err != nil {
+						return err
+					}
+					val.Set(reflect.ValueOf(t))
+					return nil
+				}
+			}
+			return fallback.DecodeValue(dc, vr, val)
+		}),
+	)
+	return reg
+}()
+
 const (
 	cdcCursorField = "_data"
 )
@@ -84,6 +138,7 @@ func (m *Mongo) Setup(ctx context.Context) error {
 
 	opts.ApplyURI(m.config.URI())
 	opts.SetCompressors([]string{"snappy"}) // using Snappy compression; read here https://en.wikipedia.org/wiki/Snappy_(compression)
+	opts.SetRegistry(safeDecodeRegistry)
 	if m.sshDialer != nil {
 		opts.SetDialer(m.sshDialer)
 	}
@@ -267,7 +322,6 @@ func filterMongoObject(doc bson.M) {
 			var err error
 			doc[key], err = typeutils.ReformatDate(t, true)
 			if err != nil {
-				logger.Warnf("failed to reformat date for key %s: %s", key, err)
 				doc[key] = time.Unix(0, 0).UTC()
 			}
 		case primitive.Null:
@@ -278,12 +332,6 @@ func filterMongoObject(doc bson.M) {
 			doc[key] = value.String()
 		case primitive.ObjectID:
 			doc[key] = value.Hex()
-		case float64:
-			if math.IsNaN(value) || math.IsInf(value, 0) {
-				doc[key] = nil
-			} else {
-				doc[key] = value
-			}
 		default:
 			doc[key] = value
 		}