build(deps): secure AMQP frames and select merged Console

Signed-off-by: Feng Ruohang <rh@vonng.com>

go.mod

@@ -4,7 +4,7 @@ go 1.27.1
 
 // Console and MC retain their historical module paths for best-effort upstream
 // compatibility. Pin the maintained PGSTY implementations used by SILO.
-replace github.com/minio/console => github.com/pgsty/silo-console v0.0.0-20260911081814-153f48fb5f80
+replace github.com/minio/console => github.com/pgsty/silo-console v0.0.0-20260911085047-638eefd7aece
 
 replace github.com/minio/mc => github.com/pgsty/mc v0.0.0-20260910093317-e6a60edf0952
 
@@ -92,7 +92,7 @@ require (
 	github.com/prometheus/common v0.71.0
 	github.com/prometheus/procfs v0.22.0
 	github.com/puzpuzpuz/xsync/v3 v3.5.1
-	github.com/rabbitmq/amqp091-go v1.10.0
+	github.com/rabbitmq/amqp091-go v1.14.0
 	github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9
 	github.com/rs/cors v1.11.1
 	github.com/secure-io/sio-go v0.3.1

go.sum

@@ -547,8 +547,8 @@ github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30/go.mod h1:85jBQOZwp
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/pgsty/mc v0.0.0-20260910093317-e6a60edf0952 h1:b6jpBuZWUiZhBox64eF1cmNL4HWGwdSk/LVXNc3DBQ0=
 github.com/pgsty/mc v0.0.0-20260910093317-e6a60edf0952/go.mod h1:VsGNEmditljwBgmYKy7X+//dXLN8gEfSPR2m4lcMPmo=
-github.com/pgsty/silo-console v0.0.0-20260911081814-153f48fb5f80 h1:bEpCn6B+aEOLriH9EjJwsGY9tarR5P/9Npv54S7DMJQ=
-github.com/pgsty/silo-console v0.0.0-20260911081814-153f48fb5f80/go.mod h1:bCyOnZactQajLOqlgCn0lHwQWMNylvP6MUOqqLako00=
+github.com/pgsty/silo-console v0.0.0-20260911085047-638eefd7aece h1:1QGCeVCRV4TzDLww9IFfRT70YOI8r/X88pRo+KILvZ4=
+github.com/pgsty/silo-console v0.0.0-20260911085047-638eefd7aece/go.mod h1:bCyOnZactQajLOqlgCn0lHwQWMNylvP6MUOqqLako00=
 github.com/pgsty/silo-pkg/v3 v3.13.4-0.20260910091716-2d8fd3cbbf07 h1:IKm2AyPsvuL4NyniK3ixqqQz1CIargmhyCkL2eZvlhA=
 github.com/pgsty/silo-pkg/v3 v3.13.4-0.20260910091716-2d8fd3cbbf07/go.mod h1:1JdcUcM+TRXObb9QrC8FWhav0RKk6/u8RT2RgimfxRQ=
 github.com/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM=
@@ -590,8 +590,8 @@ github.com/prometheus/prometheus v0.314.0 h1:YjsimqsIi6/mOtzZcrPEYUALO6zpfaht9O5
 github.com/prometheus/prometheus v0.314.0/go.mod h1:zjg3pMTAkY0/JG8jy/h8/YgSQUVB+aCXMhUqN6l64jg=
 github.com/puzpuzpuz/xsync/v3 v3.5.1 h1:GJYJZwO6IdxN/IKbneznS6yPkVC+c3zyY/j19c++5Fg=
 github.com/puzpuzpuz/xsync/v3 v3.5.1/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA=
-github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzukfVhBw=
-github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
+github.com/rabbitmq/amqp091-go v1.14.0 h1:RSaT7aOKt/OrkVUyswPDW29lnRz9psuGmfZFBmLqLek=
+github.com/rabbitmq/amqp091-go v1.14.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
 github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 h1:bsUq1dX0N8AOIL7EB/X911+m4EHsnWEHeJ0c+3TTBrg=
 github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/rivo/uniseg v0.1.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=

internal/event/target/amqp_test.go

@@ -0,0 +1,77 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package target
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/rabbitmq/amqp091-go"
+)
+
+// A broker must not make the notification client allocate or read an oversized
+// frame, including before connection.tune negotiates the frame size limit.
+func TestAMQPRejectsOversizedHandshakeFrame(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { listener.Close() })
+
+	uri, err := amqp091.ParseURI("amqp://" + listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	target, err := NewAMQPTarget("oversized-frame", AMQPArgs{Enable: true, URL: uri},
+		func(context.Context, error, string, ...any) {})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { target.Close() })
+
+	brokerDone := make(chan error, 1)
+	go func() {
+		brokerDone <- func() error {
+			conn, err := listener.Accept()
+			if err != nil {
+				return err
+			}
+			defer conn.Close()
+			if err := conn.SetDeadline(time.Now().Add(5 * time.Second)); err != nil {
+				return err
+			}
+			var protocol [8]byte
+			if _, err := io.ReadFull(conn, protocol[:]); err != nil {
+				return err
+			}
+			if protocol != [8]byte{'A', 'M', 'Q', 'P', 0, 0, 9, 1} {
+				return fmt.Errorf("unexpected AMQP protocol header: %x", protocol)
+			}
+			// A method frame declaring an 8 KiB payload exceeds the 4 KiB
+			// pre-negotiation limit. Send only its header: rejection must
+			// happen without allocating or waiting for the declared body.
+			if _, err := conn.Write([]byte{1, 0, 0, 0, 0, 0x20, 0}); err != nil {
+				return err
+			}
+			var response [1]byte
+			if n, err := conn.Read(response[:]); n != 0 || !errors.Is(err, io.EOF) {
+				return fmt.Errorf("client did not close after oversized frame header: read %d bytes, error %v", n, err)
+			}
+			return nil
+		}()
+	}()
+
+	active, err := target.IsActive()
+	if active || err == nil {
+		t.Errorf("oversized handshake frame accepted: active=%v, error=%v", active, err)
+	}
+	if err := <-brokerDone; err != nil {
+		t.Fatal(err)
+	}
+}