diff --git a/pkg/hhfab/matrix.go b/pkg/hhfab/matrix.go
index 97df430ae..a35825a33 100644
--- a/pkg/hhfab/matrix.go
+++ b/pkg/hhfab/matrix.go
@@ -10,6 +10,7 @@ import (
 	"log/slog"
 	"net/netip"
 	"slices"
+	"strings"
 	"sync"
 	"time"
 
@@ -332,6 +333,50 @@ func (m *ConnectivityMatrix) Lookup(src, dst *Endpoint, pp ProtoPort) Connectivi
 	}
 }
 
+// ProtoPortEntries returns every expectation for (src, dst) whose ProtoPort is
+// non-zero, sorted by (Protocol, Port) for deterministic ordering and logs. The
+// default ProtoPort{} entry (if any) is excluded. Returns nil when the pair has
+// no protocol-scoped expectations.
+func (m *ConnectivityMatrix) ProtoPortEntries(src, dst *Endpoint) []ConnectivityExpectation {
+	byPP, ok := m.entries[EndpointPair{Source: src, Destination: dst}]
+	if !ok {
+		return nil
+	}
+	out := make([]ConnectivityExpectation, 0, len(byPP))
+	for pp, e := range byPP {
+		if pp == (ProtoPort{}) {
+			continue
+		}
+		out = append(out, e)
+	}
+	slices.SortFunc(out, func(a, b ConnectivityExpectation) int {
+		if a.ProtoPort.Protocol != b.ProtoPort.Protocol {
+			return strings.Compare(a.ProtoPort.Protocol, b.ProtoPort.Protocol)
+		}
+
+		return int(a.ProtoPort.Port) - int(b.ProtoPort.Port)
+	})
+
+	return out
+}
+
+// HasProtoPortEntries reports whether (src, dst) carries any non-zero ProtoPort
+// expectation. The runner uses this to route protocol-scoped pairs to
+// runMatrixProtoPortPhase and away from the default server-server phase.
+func (m *ConnectivityMatrix) HasProtoPortEntries(src, dst *Endpoint) bool {
+	byPP, ok := m.entries[EndpointPair{Source: src, Destination: dst}]
+	if !ok {
+		return false
+	}
+	for pp := range byPP {
+		if pp != (ProtoPort{}) {
+			return true
+		}
+	}
+
+	return false
+}
+
 // reachabilityFromExpectation projects a matrix expectation onto the
 // Reachability struct used by the ping/iperf helpers. The matrix's
 // Verdict, Reason, and Peering map directly.
@@ -416,6 +461,13 @@ func runMatrixServerServerPhase(ctx context.Context, opts TestConnectivityOpts,
 				continue
 			}
 
+			// Protocol/port-scoped pairs are owned entirely by
+			// runMatrixProtoPortPhase (including their ICMP), so skip them
+			// here to avoid double-probing the default port.
+			if matrix.HasProtoPortEntries(src, dst) {
+				continue
+			}
+
 			entry := matrix.Lookup(src, dst, ProtoPort{})
 			// Port-forward destinations (DestinationPort set) are L4-only
 			// and handled by runMatrixPortForwardPhase below.
@@ -597,6 +649,148 @@ func runMatrixPortForwardPhase(ctx context.Context, opts TestConnectivityOpts, m
 	}
 }
 
+// persistentIperf3Port is the port the always-on iperf3 -s daemon serves
+// (TCP+UDP). Proto-port entries on this port need no on-demand listener.
+const persistentIperf3Port = 5201
+
+// startMatrixProtoPortListeners starts one iperf3 server per distinct
+// (destination server, port) referenced by a non-zero ProtoPort entry, except
+// port 5201 which the always-on iperf3 -s already serves. A single iperf3
+// server handles both TCP and UDP for its port, so listeners are deduped by
+// (host, port) regardless of protocol. The returned teardown func stops every
+// listener it started (best-effort). On any start failure, already-started
+// listeners are torn down before returning the error.
+func startMatrixProtoPortListeners(ctx context.Context, matrix *ConnectivityMatrix, deps *matrixTestDeps) (func(), error) {
+	type hostPort struct {
+		host string
+		port uint16
+	}
+	wanted := map[hostPort]struct{}{}
+	for _, src := range matrix.AllEndpoints {
+		if src.Server == nil {
+			continue
+		}
+		for _, dst := range matrix.AllEndpoints {
+			if dst.Server == nil || src == dst {
+				continue
+			}
+			for _, e := range matrix.ProtoPortEntries(src, dst) {
+				pp := e.ProtoPort
+				if pp.Protocol != "tcp" && pp.Protocol != "udp" {
+					continue
+				}
+				if pp.Port == 0 || pp.Port == persistentIperf3Port {
+					continue
+				}
+				wanted[hostPort{host: dst.Server.Name, port: pp.Port}] = struct{}{}
+			}
+		}
+	}
+
+	started := make([]hostPort, 0, len(wanted))
+	teardown := func() {
+		// Teardown must run even when the caller's ctx has been canceled
+		tctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
+		defer cancel()
+		for _, hp := range started {
+			ssh := deps.sshByServer[hp.host]
+			if ssh == nil {
+				continue
+			}
+			cmd := fmt.Sprintf("sudo docker exec iperf3 pkill -f 'iperf3 -s -p %d'", hp.port)
+			if _, stderr, err := retrySSHCmd(tctx, ssh, cmd, hp.host); err != nil {
+				slog.Warn("Failed to stop proto-port iperf3 listener", "host", hp.host, "port", hp.port, "err", err, "stderr", stderr)
+			}
+		}
+	}
+
+	for hp := range wanted {
+		ssh := deps.sshByServer[hp.host]
+		if ssh == nil {
+			teardown()
+
+			return nil, fmt.Errorf("no ssh config for server %q needed as proto-port listener", hp.host) //nolint:goerr113
+		}
+		cmd := fmt.Sprintf("sudo docker exec -d iperf3 iperf3 -s -p %d", hp.port)
+		if _, stderr, err := retrySSHCmd(ctx, ssh, cmd, hp.host); err != nil {
+			teardown()
+
+			return nil, fmt.Errorf("starting proto-port iperf3 listener on %s:%d: %w: %s", hp.host, hp.port, err, stderr)
+		}
+		started = append(started, hp)
+		slog.Debug("Started proto-port iperf3 listener", "host", hp.host, "port", hp.port)
+	}
+
+	return teardown, nil
+}
+
+// runMatrixProtoPortPhase exercises every non-zero ProtoPort matrix entry with a
+// protocol-specific probe: "icmp" → ping, "tcp" → nc connect, "udp" → iperf3 -u
+// loss check. A pair may carry several protocol entries (e.g. TCP allow + UDP
+// deny); each is probed independently. These pairs are skipped by the default
+// server-server phase (see the HasProtoPortEntries gate there), so this phase
+// owns all of their probing including ICMP. A static DNAT (NAT.DestinationIP)
+// on the entry retargets the probe, matching the other phases.
+func runMatrixProtoPortPhase(ctx context.Context, opts TestConnectivityOpts, matrix *ConnectivityMatrix, deps *matrixTestDeps) {
+	for _, src := range matrix.AllEndpoints {
+		if src.Server == nil || !deps.inSources(src.Server.Name) {
+			continue
+		}
+		for _, dst := range matrix.AllEndpoints {
+			if dst.Server == nil || src == dst {
+				continue
+			}
+			if IsSameEndpointNode(src, dst) {
+				continue
+			}
+			if !deps.inDestinations(dst.Server.Name) {
+				continue
+			}
+			for _, entry := range matrix.ProtoPortEntries(src, dst) {
+				expected := reachabilityFromExpectation(entry).Reachable
+				pp := entry.ProtoPort
+				fromName := src.Server.Name
+				toName := dst.Server.Name
+				fromSSH := deps.sshByServer[fromName]
+				toIP := dst.Server.IP
+				if entry.NAT != nil && entry.NAT.DestinationIP.IsValid() {
+					toIP = entry.NAT.DestinationIP
+				}
+				if !toIP.IsValid() {
+					deps.errChan <- fmt.Errorf("matrix proto entry %s→%s (%s/%d) has no valid target IP", fromName, toName, pp.Protocol, pp.Port) //nolint:goerr113
+
+					continue
+				}
+
+				switch pp.Protocol {
+				case "icmp":
+					deps.wg.Go(func() {
+						if pe := checkPing(ctx, opts.PingsCount, deps.pings, fromName, toName, fromSSH, toIP, nil, expected); pe != nil {
+							deps.errChan <- pe
+						}
+					})
+				case "tcp":
+					port := pp.Port
+					deps.wg.Go(func() {
+						if ie := checkTCPPort(ctx, deps.iperfs, fromName, fromSSH, toIP, port, expected); ie != nil {
+							deps.errChan <- ie
+						}
+					})
+				case "udp":
+					port := pp.Port
+					deps.wg.Go(func() {
+						if ie := checkUDPPort(ctx, opts, deps.iperfs, fromName, fromSSH, toIP, port, expected); ie != nil {
+							deps.errChan <- ie
+						}
+					})
+				default:
+					deps.errChan <- fmt.Errorf("matrix proto entry %s→%s has unsupported protocol %q", fromName, toName, pp.Protocol) //nolint:goerr113
+				}
+			}
+		}
+	}
+}
+
 func (c *Config) TestConnectivityWithMatrix(ctx context.Context, vlab *VLAB, opts TestConnectivityOpts, matrix *ConnectivityMatrix) error {
 	if matrix == nil {
 		return fmt.Errorf("connectivity matrix must be non-nil") //nolint:goerr113
@@ -645,7 +839,16 @@ func (c *Config) TestConnectivityWithMatrix(ctx context.Context, vlab *VLAB, opt
 	}
 
 	n := len(matrix.AllEndpoints)
-	errChan := make(chan error, 2*n*n+n)
+	protoEntries := 0
+	for _, src := range matrix.AllEndpoints {
+		for _, dst := range matrix.AllEndpoints {
+			if src == dst {
+				continue
+			}
+			protoEntries += len(matrix.ProtoPortEntries(src, dst))
+		}
+	}
+	errChan := make(chan error, 2*n*n+n+protoEntries)
 	deps := &matrixTestDeps{
 		sshByServer: sshByServer,
 		pings:       semaphore.NewWeighted(opts.PingsParallel),
@@ -661,6 +864,19 @@ func (c *Config) TestConnectivityWithMatrix(ctx context.Context, vlab *VLAB, opt
 		errChan: errChan,
 	}
 
+	// Start on-demand iperf3 listeners for any non-5201 proto-port before
+	// spawning probe goroutines, so a listener failure returns cleanly. The
+	// deferred teardown runs at function return, i.e. after deps.wg.Wait().
+	teardownListeners := func() {}
+	if opts.PingsCount > 0 || opts.IPerfsSeconds > 0 {
+		td, err := startMatrixProtoPortListeners(ctx, matrix, deps)
+		if err != nil {
+			return err
+		}
+		teardownListeners = td
+	}
+	defer teardownListeners()
+
 	if opts.PingsCount > 0 || opts.IPerfsSeconds > 0 {
 		if err := runMatrixServerServerPhase(ctx, opts, matrix, deps); err != nil {
 			return err
@@ -672,6 +888,9 @@ func (c *Config) TestConnectivityWithMatrix(ctx context.Context, vlab *VLAB, opt
 	if opts.IPerfsSeconds > 0 {
 		runMatrixPortForwardPhase(ctx, opts, matrix, deps)
 	}
+	if opts.PingsCount > 0 || opts.IPerfsSeconds > 0 {
+		runMatrixProtoPortPhase(ctx, opts, matrix, deps)
+	}
 
 	deps.wg.Wait()
 	close(errChan)
diff --git a/pkg/hhfab/matrix_test.go b/pkg/hhfab/matrix_test.go
new file mode 100644
index 000000000..5852ff01b
--- /dev/null
+++ b/pkg/hhfab/matrix_test.go
@@ -0,0 +1,140 @@
+// Copyright 2026 Hedgehog
+// SPDX-License-Identifier: Apache-2.0
+
+package hhfab
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestProtoPortEntries_ExcludesDefaultAndSorts(t *testing.T) {
+	m := NewConnectivityMatrix()
+	src := serverEP("server-1", "vpc-1", "default", "10.0.1.1")
+	dst := serverEP("server-2", "vpc-2", "default", "10.0.2.2")
+	m.AllEndpoints = []*Endpoint{src, dst}
+
+	// A default (ProtoPort{}) entry must be excluded from the proto list.
+	m.Add(ConnectivityExpectation{
+		Pair:    EndpointPair{Source: src, Destination: dst},
+		Verdict: VerdictAllow,
+	})
+	// Add out of order to prove sorting by (Protocol, Port).
+	m.Add(ConnectivityExpectation{
+		Pair:      EndpointPair{Source: src, Destination: dst},
+		Verdict:   VerdictDeny,
+		ProtoPort: ProtoPort{Protocol: "udp", Port: 5201},
+	})
+	m.Add(ConnectivityExpectation{
+		Pair:      EndpointPair{Source: src, Destination: dst},
+		Verdict:   VerdictAllow,
+		ProtoPort: ProtoPort{Protocol: "tcp", Port: 6201},
+	})
+	m.Add(ConnectivityExpectation{
+		Pair:      EndpointPair{Source: src, Destination: dst},
+		Verdict:   VerdictAllow,
+		ProtoPort: ProtoPort{Protocol: "tcp", Port: 5201},
+	})
+
+	got := m.ProtoPortEntries(src, dst)
+	require.Len(t, got, 3, "default entry excluded")
+	require.Equal(t, ProtoPort{Protocol: "tcp", Port: 5201}, got[0].ProtoPort)
+	require.Equal(t, ProtoPort{Protocol: "tcp", Port: 6201}, got[1].ProtoPort)
+	require.Equal(t, ProtoPort{Protocol: "udp", Port: 5201}, got[2].ProtoPort)
+	require.Equal(t, VerdictDeny, got[2].Verdict)
+}
+
+func TestProtoPortEntries_NoneForUnknownPair(t *testing.T) {
+	m := NewConnectivityMatrix()
+	src := serverEP("server-1", "vpc-1", "default", "10.0.1.1")
+	dst := serverEP("server-2", "vpc-2", "default", "10.0.2.2")
+	m.AllEndpoints = []*Endpoint{src, dst}
+
+	require.Nil(t, m.ProtoPortEntries(src, dst))
+	require.False(t, m.HasProtoPortEntries(src, dst))
+}
+
+func TestHasProtoPortEntries_IgnoresDefaultOnly(t *testing.T) {
+	m := NewConnectivityMatrix()
+	src := serverEP("server-1", "vpc-1", "default", "10.0.1.1")
+	dst := serverEP("server-2", "vpc-2", "default", "10.0.2.2")
+	m.AllEndpoints = []*Endpoint{src, dst}
+
+	// A pair with only a default entry must NOT be treated as proto-scoped,
+	// so the legacy server-server phase keeps owning it.
+	m.Add(ConnectivityExpectation{
+		Pair:    EndpointPair{Source: src, Destination: dst},
+		Verdict: VerdictAllow,
+	})
+	require.False(t, m.HasProtoPortEntries(src, dst))
+
+	// Once a non-zero ProtoPort entry lands, the pair is proto-scoped and the
+	// legacy phase gate (which keys on this) routes it to the proto-port phase.
+	m.Add(ConnectivityExpectation{
+		Pair:      EndpointPair{Source: src, Destination: dst},
+		Verdict:   VerdictAllow,
+		ProtoPort: ProtoPort{Protocol: "tcp", Port: 5201},
+	})
+	require.True(t, m.HasProtoPortEntries(src, dst))
+}
+
+func TestParseNCReturnCode(t *testing.T) {
+	cases := []struct {
+		name   string
+		stdout string
+		wantRC int
+		wantOk bool
+	}{
+		{"connect ok", "NCRC=0\n", 0, true},
+		{"refused/timeout", "nc: connect failed\nNCRC=1\n", 1, true},
+		{"not found", "bash: nc: command not found\nNCRC=127\n", 127, true},
+		{"marker with spaces", "  NCRC=1  \n", 1, true},
+		{"no marker (probe did not complete)", "some ssh noise\n", 0, false},
+		{"empty", "", 0, false},
+		{"non-numeric marker", "NCRC=oops\n", 0, false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			rc, ok := parseNCReturnCode(tc.stdout)
+			require.Equal(t, tc.wantOk, ok)
+			if tc.wantOk {
+				require.Equal(t, tc.wantRC, rc)
+			}
+		})
+	}
+}
+
+func TestSetVPCToVPCProtoVerdict_AccumulatesScopesAndPreservesPeering(t *testing.T) {
+	m := NewConnectivityMatrix()
+	a1 := serverEP("server-1", "vpc-1", "default", "10.0.1.1")
+	a2 := serverEP("server-2", "vpc-1", "default", "10.0.1.2")
+	b1 := serverEP("server-3", "vpc-2", "default", "10.0.2.1")
+	c1 := serverEP("server-4", "vpc-3", "default", "10.0.3.1")
+	m.AllEndpoints = []*Endpoint{a1, a2, b1, c1}
+
+	// Seed a default entry so we can prove Peering is preserved onto the
+	// proto entries.
+	m.Add(ConnectivityExpectation{
+		Pair:    EndpointPair{Source: a1, Destination: b1},
+		Verdict: VerdictAllow,
+		Peering: "vpc-1--vpc-2",
+	})
+
+	setVPCToVPCProtoVerdict(m, "vpc-1", "vpc-2", ProtoPort{Protocol: "tcp", Port: 5201}, VerdictAllow)
+	setVPCToVPCProtoVerdict(m, "vpc-1", "vpc-2", ProtoPort{Protocol: "udp", Port: 5201}, VerdictDeny)
+
+	// Both protocols coexist on the same pair.
+	tcp := m.Lookup(a1, b1, ProtoPort{Protocol: "tcp", Port: 5201})
+	udp := m.Lookup(a1, b1, ProtoPort{Protocol: "udp", Port: 5201})
+	require.Equal(t, VerdictAllow, tcp.Verdict)
+	require.Equal(t, VerdictDeny, udp.Verdict)
+	require.Equal(t, "vpc-1--vpc-2", tcp.Peering, "existing peering preserved")
+	require.Equal(t, ReachabilityReasonGatewayPeering, tcp.Reason)
+
+	// Applied to every source in vpc-1 (a2 too), not just a1.
+	require.Equal(t, VerdictAllow, m.Lookup(a2, b1, ProtoPort{Protocol: "tcp", Port: 5201}).Verdict)
+
+	// Not applied to a server outside the destination VPC.
+	require.False(t, m.HasProtoPortEntries(a1, c1), "vpc-3 destination untouched")
+}
diff --git a/pkg/hhfab/rt_acl_tests.go b/pkg/hhfab/rt_acl_tests.go
new file mode 100644
index 000000000..d1724967d
--- /dev/null
+++ b/pkg/hhfab/rt_acl_tests.go
@@ -0,0 +1,489 @@
+// Copyright 2026 Hedgehog
+// SPDX-License-Identifier: Apache-2.0
+
+package hhfab
+
+import (
+	"context"
+	"fmt"
+
+	gwapi "go.githedgehog.com/fabric/api/gateway/v1alpha1"
+	vpcapi "go.githedgehog.com/fabric/api/vpc/v1beta1"
+)
+
+// Gateway peering ACL tests. Each test establishes a single peering between the
+// first two VPCs with a specific PeeringACL and asserts the resulting
+// per-protocol/per-port connectivity, reusing the natTestSpec/runNATTest driver.
+//
+// After Repopulate, populateConnectivityMatrix marks the peered pair Allow at
+// the default ProtoPort{} purely from subnet presence — it does not read the
+// ACL. So every ACL test's Overlay must set EXPLICIT protocol/port-scoped
+// verdicts via setVPCToVPCProtoVerdict; runMatrixProtoPortPhase then owns all
+// probing for those pairs (the legacy phase skips them), including ICMP.
+//
+// Scope semantics (per the gateway design):
+//   - flow (default): stateful matching that relies on the dataplane keeping
+//     connection state. It is ONLY accepted on a peering that has stateful NAT
+//     (masquerade, possibly port-forward) — that is where the flow/conntrack
+//     information exists. A flow-scoped rule on a NAT-free peering is invalid.
+//   - packet: stateless, per-direction matching; valid on any peering, no NAT
+//     required.
+//
+// Because most ACL cases here run on NAT-free peerings, their rules use explicit
+// Scope:packet (the API default is flow, which those peerings could not accept).
+// The flow scope is exercised by a dedicated masquerade-backed case.
+//
+// Return-path reality: a probe (ping, TCP handshake, iperf3) only succeeds when
+// BOTH directions of the flow are permitted. With packet (stateless) scope the
+// reply is matched by its own rule, and because a reply has the ports swapped, a
+// destination-port allow needs a matching source-port allow for the reverse
+// direction. A single one-way packet rule therefore yields NO connectivity (the
+// reply is dropped). Cases that need a working allowed flow either permit both
+// directions with return-compatible matches (protocol-only, subnet-only, or a
+// dst-port + src-port pair) or use flow scope + masquerade, where conntrack
+// permits the return automatically.
+//
+// NOTE: these tests encode the intended behavior ahead of the dataplane
+// implementation, so they are expected to fail until ACL enforcement lands.
+// Rule precedence is assumed first-match-wins and ICMP is assumed to fall to the
+// default action (it matches neither tcp nor udp rules); the precedence and
+// protocol tests will reveal these if the dataplane differs.
+
+const (
+	// aclProbePort is served by the always-on iperf3 daemon (TCP+UDP), so
+	// tests probing it need no on-demand listener.
+	aclProbePort uint16 = 5201
+	// aclAltPort exercises the on-demand listener path for arbitrary-port ACLs.
+	aclAltPort uint16 = 6201
+	// aclAltPortRange is a port range containing aclAltPort (6201) but not
+	// aclProbePort (5201); used to exercise range matching in the ACL rules.
+	aclAltPortRange = "6000-6500"
+	// aclUnprobedPort is a port no probe ever targets. The dataplane rejects
+	// rule-less ACLs, so default-action tests carry a single narrow rule on
+	// this port; because nothing probes it, the default action still governs
+	// every path the matrix actually exercises.
+	aclUnprobedPort uint16 = 9999
+)
+
+// setACLDirVerdicts sets the icmp, tcp/aclProbePort and udp/aclProbePort
+// expectations for a single (srcVPC → dstVPC) direction. It is the common
+// three-probe shape most ACL cases assert.
+func setACLDirVerdicts(m *ConnectivityMatrix, srcVPC, dstVPC string, icmp, tcp, udp ConnectivityVerdict) {
+	setVPCToVPCProtoVerdict(m, srcVPC, dstVPC, ProtoPort{Protocol: "icmp"}, icmp)
+	setVPCToVPCProtoVerdict(m, srcVPC, dstVPC, ProtoPort{Protocol: "tcp", Port: aclProbePort}, tcp)
+	setVPCToVPCProtoVerdict(m, srcVPC, dstVPC, ProtoPort{Protocol: "udp", Port: aclProbePort}, udp)
+}
+
+// gatewayACLDefaultDenyTest: default=deny blocks all traffic despite the subnets
+// being exposed. The ACL carries one narrow allow rule on an unprobed port
+// (rule-less ACLs are rejected by the dataplane); every probed path still falls
+// to the default deny.
+func gatewayACLDefaultDenyTest(ctx context.Context, testCtx *VPCPeeringTestCtx, matrix *ConnectivityMatrix) (bool, []RevertFunc, error) {
+	return testCtx.runNATTest(ctx, matrix, natTestSpec{
+		Name: "gateway ACL default deny",
+		BuildSpec: func(vpc1, vpc2 *vpcapi.VPC) (peeringSpecs, error) {
+			specs := emptyPeeringSpecs()
+			acl := &gwapi.PeeringACL{
+				Default: gwapi.ACLDefaultDeny,
+				Rules: []gwapi.PeeringACLRule{{
+					Name: "allow-unprobed", From: vpc1.Name, To: vpc2.Name,
+					Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket,
+					Match: gwapi.PeeringACLMatch{
+						Protocol:    gwapi.ACLMatchProtocolTCP,
+						Destination: []gwapi.PeeringACLMatchEndpoint{{Ports: []string{fmt.Sprintf("%d", aclUnprobedPort)}}},
+					},
+				}},
+			}
+			err := appendGwPeeringSpec(specs.Gateway, vpc1, vpc2, &GwPeeringOptions{ACL: acl})
+
+			return specs, err
+		},
+		Overlay: func(vpc1, vpc2 *vpcapi.VPC, matrix *ConnectivityMatrix) error {
+			setACLDirVerdicts(matrix, vpc1.Name, vpc2.Name, VerdictDeny, VerdictDeny, VerdictDeny)
+			setACLDirVerdicts(matrix, vpc2.Name, vpc1.Name, VerdictDeny, VerdictDeny, VerdictDeny)
+
+			return nil
+		},
+	})
+}
+
+// gatewayACLDenyUnlessExposedTest: default=deny-unless-exposed keeps exposed
+// subnets reachable, while an explicit deny rule carves out a slice. Here the
+// carve-out denies UDP in both directions, so TCP and ICMP fall to the default
+// (exposed ⇒ allowed) and UDP is blocked. This is both the permissive-default
+// positive control AND coverage of a Protocol:udp deny rule. UDP denial is
+// verifiable because TCP stays allowed, so the iperf3 -u TCP control channel
+// still establishes before the datagrams are dropped.
+func gatewayACLDenyUnlessExposedTest(ctx context.Context, testCtx *VPCPeeringTestCtx, matrix *ConnectivityMatrix) (bool, []RevertFunc, error) {
+	return testCtx.runNATTest(ctx, matrix, natTestSpec{
+		Name: "gateway ACL deny-unless-exposed with udp carve-out",
+		BuildSpec: func(vpc1, vpc2 *vpcapi.VPC) (peeringSpecs, error) {
+			specs := emptyPeeringSpecs()
+			acl := &gwapi.PeeringACL{
+				Default: gwapi.ACLDefaultDenyUnlessExposed,
+				Rules: []gwapi.PeeringACLRule{
+					{Name: "deny-udp-fwd", From: vpc1.Name, To: vpc2.Name, Action: gwapi.ACLActionDeny, Scope: gwapi.ACLScopePacket, Match: gwapi.PeeringACLMatch{Protocol: gwapi.ACLMatchProtocolUDP}},
+					{Name: "deny-udp-rev", From: vpc2.Name, To: vpc1.Name, Action: gwapi.ACLActionDeny, Scope: gwapi.ACLScopePacket, Match: gwapi.PeeringACLMatch{Protocol: gwapi.ACLMatchProtocolUDP}},
+				},
+			}
+			err := appendGwPeeringSpec(specs.Gateway, vpc1, vpc2, &GwPeeringOptions{ACL: acl})
+
+			return specs, err
+		},
+		Overlay: func(vpc1, vpc2 *vpcapi.VPC, matrix *ConnectivityMatrix) error {
+			// TCP/ICMP: exposed ⇒ allowed both ways. UDP: denied both ways.
+			setACLDirVerdicts(matrix, vpc1.Name, vpc2.Name, VerdictAllow, VerdictAllow, VerdictDeny)
+			setACLDirVerdicts(matrix, vpc2.Name, vpc1.Name, VerdictAllow, VerdictAllow, VerdictDeny)
+
+			return nil
+		},
+	})
+}
+
+// gatewayACLExplicitAllowTest: default=deny plus explicit allow rules for both
+// directions restore full connectivity (contrasts the default-deny case).
+func gatewayACLExplicitAllowTest(ctx context.Context, testCtx *VPCPeeringTestCtx, matrix *ConnectivityMatrix) (bool, []RevertFunc, error) {
+	return testCtx.runNATTest(ctx, matrix, natTestSpec{
+		Name: "gateway ACL explicit allow rule",
+		BuildSpec: func(vpc1, vpc2 *vpcapi.VPC) (peeringSpecs, error) {
+			specs := emptyPeeringSpecs()
+			acl := &gwapi.PeeringACL{
+				Default: gwapi.ACLDefaultDeny,
+				Rules: []gwapi.PeeringACLRule{
+					{Name: "allow-fwd", From: vpc1.Name, To: vpc2.Name, Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket},
+					{Name: "allow-rev", From: vpc2.Name, To: vpc1.Name, Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket},
+				},
+			}
+			err := appendGwPeeringSpec(specs.Gateway, vpc1, vpc2, &GwPeeringOptions{ACL: acl})
+
+			return specs, err
+		},
+		Overlay: func(vpc1, vpc2 *vpcapi.VPC, matrix *ConnectivityMatrix) error {
+			setACLDirVerdicts(matrix, vpc1.Name, vpc2.Name, VerdictAllow, VerdictAllow, VerdictAllow)
+			setACLDirVerdicts(matrix, vpc2.Name, vpc1.Name, VerdictAllow, VerdictAllow, VerdictAllow)
+
+			return nil
+		},
+	})
+}
+
+// gatewayACLProtocolScopingTest: allow TCP in both directions; UDP and ICMP fall
+// to the default deny.
+func gatewayACLProtocolScopingTest(ctx context.Context, testCtx *VPCPeeringTestCtx, matrix *ConnectivityMatrix) (bool, []RevertFunc, error) {
+	return testCtx.runNATTest(ctx, matrix, natTestSpec{
+		Name: "gateway ACL protocol scoping",
+		BuildSpec: func(vpc1, vpc2 *vpcapi.VPC) (peeringSpecs, error) {
+			specs := emptyPeeringSpecs()
+			acl := &gwapi.PeeringACL{
+				Default: gwapi.ACLDefaultDeny,
+				Rules: []gwapi.PeeringACLRule{
+					{Name: "allow-tcp-fwd", From: vpc1.Name, To: vpc2.Name, Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket, Match: gwapi.PeeringACLMatch{Protocol: gwapi.ACLMatchProtocolTCP}},
+					{Name: "allow-tcp-rev", From: vpc2.Name, To: vpc1.Name, Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket, Match: gwapi.PeeringACLMatch{Protocol: gwapi.ACLMatchProtocolTCP}},
+				},
+			}
+			err := appendGwPeeringSpec(specs.Gateway, vpc1, vpc2, &GwPeeringOptions{ACL: acl})
+
+			return specs, err
+		},
+		Overlay: func(vpc1, vpc2 *vpcapi.VPC, matrix *ConnectivityMatrix) error {
+			setACLDirVerdicts(matrix, vpc1.Name, vpc2.Name, VerdictDeny, VerdictAllow, VerdictDeny)
+			setACLDirVerdicts(matrix, vpc2.Name, vpc1.Name, VerdictDeny, VerdictAllow, VerdictDeny)
+
+			return nil
+		},
+	})
+}
+
+// gatewayACLPacketOneWayTest: a single packet-scoped From:vpc1,To:vpc2 allow rule
+// permits only the forward packets; the reply (vpc2→vpc1) has no matching rule
+// and hits the default deny, so no probe can complete a handshake or get an ICMP
+// reply. This locks in the stateless return-path requirement: a one-way packet
+// rule yields NO connectivity in either direction. (Working directional allow is
+// covered by the flow+masquerade case, which is stateful.)
+func gatewayACLPacketOneWayTest(ctx context.Context, testCtx *VPCPeeringTestCtx, matrix *ConnectivityMatrix) (bool, []RevertFunc, error) {
+	return testCtx.runNATTest(ctx, matrix, natTestSpec{
+		Name: "gateway ACL packet one-way (no return)",
+		BuildSpec: func(vpc1, vpc2 *vpcapi.VPC) (peeringSpecs, error) {
+			specs := emptyPeeringSpecs()
+			acl := &gwapi.PeeringACL{
+				Default: gwapi.ACLDefaultDeny,
+				Rules: []gwapi.PeeringACLRule{
+					{Name: "allow-fwd", From: vpc1.Name, To: vpc2.Name, Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket},
+				},
+			}
+			err := appendGwPeeringSpec(specs.Gateway, vpc1, vpc2, &GwPeeringOptions{ACL: acl})
+
+			return specs, err
+		},
+		Overlay: func(vpc1, vpc2 *vpcapi.VPC, matrix *ConnectivityMatrix) error {
+			// Forward flow can't complete without its return, so both
+			// directions are unreachable.
+			setACLDirVerdicts(matrix, vpc1.Name, vpc2.Name, VerdictDeny, VerdictDeny, VerdictDeny)
+			setACLDirVerdicts(matrix, vpc2.Name, vpc1.Name, VerdictDeny, VerdictDeny, VerdictDeny)
+
+			return nil
+		},
+	})
+}
+
+// gatewayACLFlowScopeMasqueradeTest: a flow-scoped rule is only valid on a
+// peering that has stateful NAT (where the dataplane keeps flow/conntrack
+// state), so this case pairs a masquerade NAT on VPC1 with a flow-scoped
+// From:vpc1,To:vpc2 allow rule. Masquerade SNAT lets VPC1 reach VPC2's real
+// IPs and the flow rule permits that stateful flow (and its return traffic);
+// VPC2 cannot initiate (masquerade blocks unsolicited inbound and no reverse
+// rule exists). Verifies flow scope is accepted and enforced with masquerade.
+func gatewayACLFlowScopeMasqueradeTest(ctx context.Context, testCtx *VPCPeeringTestCtx, matrix *ConnectivityMatrix) (bool, []RevertFunc, error) {
+	const vpc1NATCIDR = "192.168.81.0/24"
+
+	return testCtx.runNATTest(ctx, matrix, natTestSpec{
+		Name: "gateway ACL flow scope with masquerade",
+		BuildSpec: func(vpc1, vpc2 *vpcapi.VPC) (peeringSpecs, error) {
+			specs := emptyPeeringSpecs()
+			acl := &gwapi.PeeringACL{
+				Default: gwapi.ACLDefaultDeny,
+				Rules: []gwapi.PeeringACLRule{
+					{Name: "allow-flow", From: vpc1.Name, To: vpc2.Name, Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopeFlow},
+				},
+			}
+			err := appendGwPeeringSpec(specs.Gateway, vpc1, vpc2, &GwPeeringOptions{
+				VPC1NATCIDR: []string{vpc1NATCIDR},
+				VPC1NATMode: NATModeMasquerade,
+				ACL:         acl,
+			})
+
+			return specs, err
+		},
+		Overlay: func(vpc1, vpc2 *vpcapi.VPC, matrix *ConnectivityMatrix) error {
+			// vpc1→vpc2 rides masquerade SNAT against vpc2's real IPs and is
+			// permitted by the flow rule. vpc2→vpc1 is blocked both by
+			// masquerade (stateful, no unsolicited inbound) and by the ACL
+			// default deny.
+			setACLDirVerdicts(matrix, vpc1.Name, vpc2.Name, VerdictAllow, VerdictAllow, VerdictAllow)
+			setACLDirVerdicts(matrix, vpc2.Name, vpc1.Name, VerdictDeny, VerdictDeny, VerdictDeny)
+
+			return nil
+		},
+	})
+}
+
+// gatewayACLSubnetScopingTest: packet-scoped rules matching source and
+// destination in BOTH directions so the reply is permitted too. The forward rule
+// selects by VPCSubnet name and the reverse rule by CIDR, so a single test
+// exercises both endpoint selectors. With one subnet per VPC each match covers
+// the whole VPC, so this validates the subnet/CIDR match plumbing (with a working
+// flow) rather than discriminating between subnets.
+func gatewayACLSubnetScopingTest(ctx context.Context, testCtx *VPCPeeringTestCtx, matrix *ConnectivityMatrix) (bool, []RevertFunc, error) {
+	return testCtx.runNATTest(ctx, matrix, natTestSpec{
+		Name: "gateway ACL subnet/CIDR scoping",
+		BuildSpec: func(vpc1, vpc2 *vpcapi.VPC) (peeringSpecs, error) {
+			specs := emptyPeeringSpecs()
+			vpc1CIDR, err := vpcFirstSubnetCIDR(vpc1)
+			if err != nil {
+				return specs, err
+			}
+			vpc2CIDR, err := vpcFirstSubnetCIDR(vpc2)
+			if err != nil {
+				return specs, err
+			}
+			acl := &gwapi.PeeringACL{
+				Default: gwapi.ACLDefaultDeny,
+				Rules: []gwapi.PeeringACLRule{
+					// forward: select the subnets by VPC subnet name
+					{
+						Name: "allow-subnet-fwd", From: vpc1.Name, To: vpc2.Name,
+						Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket,
+						Match: gwapi.PeeringACLMatch{
+							Source:      []gwapi.PeeringACLMatchEndpoint{{VPCSubnet: "subnet-01"}},
+							Destination: []gwapi.PeeringACLMatchEndpoint{{VPCSubnet: "subnet-01"}},
+						},
+					},
+					// reverse: select the same subnets by CIDR (return path)
+					{
+						Name: "allow-cidr-rev", From: vpc2.Name, To: vpc1.Name,
+						Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket,
+						Match: gwapi.PeeringACLMatch{
+							Source:      []gwapi.PeeringACLMatchEndpoint{{CIDR: vpc2CIDR}},
+							Destination: []gwapi.PeeringACLMatchEndpoint{{CIDR: vpc1CIDR}},
+						},
+					},
+				},
+			}
+			err = appendGwPeeringSpec(specs.Gateway, vpc1, vpc2, &GwPeeringOptions{ACL: acl})
+
+			return specs, err
+		},
+		Overlay: func(vpc1, vpc2 *vpcapi.VPC, matrix *ConnectivityMatrix) error {
+			// Neither match carries a port/protocol constraint, so returns
+			// match the reverse rule and every protocol flows both ways.
+			setACLDirVerdicts(matrix, vpc1.Name, vpc2.Name, VerdictAllow, VerdictAllow, VerdictAllow)
+			setACLDirVerdicts(matrix, vpc2.Name, vpc1.Name, VerdictAllow, VerdictAllow, VerdictAllow)
+
+			return nil
+		},
+	})
+}
+
+// gatewayACLPortScopingTest: allow a vpc1→vpc2 TCP flow whose port falls in
+// aclAltPortRange. The forward rule matches a destination-port RANGE; the reverse
+// rule matches the same SOURCE-port range so the server's replies (whose source
+// port is the listener port) are permitted and the handshake completes. The probe
+// hits aclAltPort (in range) → reachable forward (exercising the on-demand
+// listener); TCP/aclProbePort (out of range), UDP, ICMP, and any vpc2-initiated
+// flow fall to the default deny. Covers both port-range matching and src/dst-port
+// selectors in one case.
+func gatewayACLPortScopingTest(ctx context.Context, testCtx *VPCPeeringTestCtx, matrix *ConnectivityMatrix) (bool, []RevertFunc, error) {
+	return testCtx.runNATTest(ctx, matrix, natTestSpec{
+		Name: "gateway ACL port range scoping",
+		BuildSpec: func(vpc1, vpc2 *vpcapi.VPC) (peeringSpecs, error) {
+			specs := emptyPeeringSpecs()
+			acl := &gwapi.PeeringACL{
+				Default: gwapi.ACLDefaultDeny,
+				Rules: []gwapi.PeeringACLRule{
+					{
+						Name: "allow-alt-fwd", From: vpc1.Name, To: vpc2.Name,
+						Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket,
+						Match: gwapi.PeeringACLMatch{
+							Protocol:    gwapi.ACLMatchProtocolTCP,
+							Destination: []gwapi.PeeringACLMatchEndpoint{{Ports: []string{aclAltPortRange}}},
+						},
+					},
+					{
+						Name: "allow-alt-ret", From: vpc2.Name, To: vpc1.Name,
+						Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket,
+						Match: gwapi.PeeringACLMatch{
+							Protocol: gwapi.ACLMatchProtocolTCP,
+							Source:   []gwapi.PeeringACLMatchEndpoint{{Ports: []string{aclAltPortRange}}},
+						},
+					},
+				},
+			}
+			err := appendGwPeeringSpec(specs.Gateway, vpc1, vpc2, &GwPeeringOptions{ACL: acl})
+
+			return specs, err
+		},
+		Overlay: func(vpc1, vpc2 *vpcapi.VPC, matrix *ConnectivityMatrix) error {
+			// Forward: only TCP/aclAltPort completes (reply rides the src-port
+			// return rule). Reverse: a vpc2-initiated connect to aclAltPort has
+			// dst=aclAltPort/src=ephemeral, matching neither rule → default deny.
+			setVPCToVPCProtoVerdict(matrix, vpc1.Name, vpc2.Name, ProtoPort{Protocol: "tcp", Port: aclAltPort}, VerdictAllow)
+			setVPCToVPCProtoVerdict(matrix, vpc1.Name, vpc2.Name, ProtoPort{Protocol: "tcp", Port: aclProbePort}, VerdictDeny)
+			setVPCToVPCProtoVerdict(matrix, vpc1.Name, vpc2.Name, ProtoPort{Protocol: "udp", Port: aclProbePort}, VerdictDeny)
+			setVPCToVPCProtoVerdict(matrix, vpc1.Name, vpc2.Name, ProtoPort{Protocol: "icmp"}, VerdictDeny)
+			setACLDirVerdicts(matrix, vpc2.Name, vpc1.Name, VerdictDeny, VerdictDeny, VerdictDeny)
+			setVPCToVPCProtoVerdict(matrix, vpc2.Name, vpc1.Name, ProtoPort{Protocol: "tcp", Port: aclAltPort}, VerdictDeny)
+
+			return nil
+		},
+	})
+}
+
+// gatewayACLPrecedenceAllowThenDenyTest: an allow for TCP/aclProbePort ahead of a
+// broad deny (packet-scoped, vpc1→vpc2). Under first-match-wins TCP/aclProbePort
+// is allowed forward while everything else is denied. A reverse source-port rule
+// lets the server's replies through so the allowed flow completes; the reverse
+// direction otherwise follows the default deny.
+func gatewayACLPrecedenceAllowThenDenyTest(ctx context.Context, testCtx *VPCPeeringTestCtx, matrix *ConnectivityMatrix) (bool, []RevertFunc, error) {
+	probePortStr := fmt.Sprintf("%d", aclProbePort)
+
+	return testCtx.runNATTest(ctx, matrix, natTestSpec{
+		Name: "gateway ACL precedence allow-then-deny",
+		BuildSpec: func(vpc1, vpc2 *vpcapi.VPC) (peeringSpecs, error) {
+			specs := emptyPeeringSpecs()
+			acl := &gwapi.PeeringACL{
+				Default: gwapi.ACLDefaultDeny,
+				Rules: []gwapi.PeeringACLRule{
+					{
+						Name: "allow-tcp", From: vpc1.Name, To: vpc2.Name,
+						Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket,
+						Match: gwapi.PeeringACLMatch{
+							Protocol:    gwapi.ACLMatchProtocolTCP,
+							Destination: []gwapi.PeeringACLMatchEndpoint{{Ports: []string{probePortStr}}},
+						},
+					},
+					{Name: "deny-all", From: vpc1.Name, To: vpc2.Name, Action: gwapi.ACLActionDeny, Scope: gwapi.ACLScopePacket},
+					{
+						Name: "allow-tcp-ret", From: vpc2.Name, To: vpc1.Name,
+						Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket,
+						Match: gwapi.PeeringACLMatch{
+							Protocol: gwapi.ACLMatchProtocolTCP,
+							Source:   []gwapi.PeeringACLMatchEndpoint{{Ports: []string{probePortStr}}},
+						},
+					},
+				},
+			}
+			err := appendGwPeeringSpec(specs.Gateway, vpc1, vpc2, &GwPeeringOptions{ACL: acl})
+
+			return specs, err
+		},
+		Overlay: func(vpc1, vpc2 *vpcapi.VPC, matrix *ConnectivityMatrix) error {
+			setACLDirVerdicts(matrix, vpc1.Name, vpc2.Name, VerdictDeny, VerdictAllow, VerdictDeny)
+			setACLDirVerdicts(matrix, vpc2.Name, vpc1.Name, VerdictDeny, VerdictDeny, VerdictDeny)
+
+			return nil
+		},
+	})
+}
+
+// gatewayACLPrecedenceDenyThenAllowTest: the reverse rule order of the previous
+// test (same rule set, including the source-port return rule — only the order of
+// the forward allow/deny is swapped). Under first-match-wins the broad deny
+// matches first, so even TCP/aclProbePort is denied and nothing gets through.
+func gatewayACLPrecedenceDenyThenAllowTest(ctx context.Context, testCtx *VPCPeeringTestCtx, matrix *ConnectivityMatrix) (bool, []RevertFunc, error) {
+	probePortStr := fmt.Sprintf("%d", aclProbePort)
+
+	return testCtx.runNATTest(ctx, matrix, natTestSpec{
+		Name: "gateway ACL precedence deny-then-allow",
+		BuildSpec: func(vpc1, vpc2 *vpcapi.VPC) (peeringSpecs, error) {
+			specs := emptyPeeringSpecs()
+			acl := &gwapi.PeeringACL{
+				Default: gwapi.ACLDefaultDeny,
+				Rules: []gwapi.PeeringACLRule{
+					{Name: "deny-all", From: vpc1.Name, To: vpc2.Name, Action: gwapi.ACLActionDeny, Scope: gwapi.ACLScopePacket},
+					{
+						Name: "allow-tcp", From: vpc1.Name, To: vpc2.Name,
+						Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket,
+						Match: gwapi.PeeringACLMatch{
+							Protocol:    gwapi.ACLMatchProtocolTCP,
+							Destination: []gwapi.PeeringACLMatchEndpoint{{Ports: []string{probePortStr}}},
+						},
+					},
+					{
+						Name: "allow-tcp-ret", From: vpc2.Name, To: vpc1.Name,
+						Action: gwapi.ACLActionAllow, Scope: gwapi.ACLScopePacket,
+						Match: gwapi.PeeringACLMatch{
+							Protocol: gwapi.ACLMatchProtocolTCP,
+							Source:   []gwapi.PeeringACLMatchEndpoint{{Ports: []string{probePortStr}}},
+						},
+					},
+				},
+			}
+			err := appendGwPeeringSpec(specs.Gateway, vpc1, vpc2, &GwPeeringOptions{ACL: acl})
+
+			return specs, err
+		},
+		Overlay: func(vpc1, vpc2 *vpcapi.VPC, matrix *ConnectivityMatrix) error {
+			setACLDirVerdicts(matrix, vpc1.Name, vpc2.Name, VerdictDeny, VerdictDeny, VerdictDeny)
+			setACLDirVerdicts(matrix, vpc2.Name, vpc1.Name, VerdictDeny, VerdictDeny, VerdictDeny)
+
+			return nil
+		},
+	})
+}
+
+// getACLTestCases returns the gateway peering ACL test cases added to the
+// multi-VPC single-subnet suite.
+func getACLTestCases() []JUnitTestCase {
+	return []JUnitTestCase{
+		{Name: "Gateway Peering ACL Default Deny", F: gatewayACLDefaultDenyTest, SkipFlags: SkipFlags{NoGateway: true, NoServers: true}},
+		{Name: "Gateway Peering ACL Deny-Unless-Exposed UDP Carve-Out", F: gatewayACLDenyUnlessExposedTest, SkipFlags: SkipFlags{NoGateway: true, NoServers: true}},
+		{Name: "Gateway Peering ACL Explicit Allow", F: gatewayACLExplicitAllowTest, SkipFlags: SkipFlags{NoGateway: true, NoServers: true}},
+		{Name: "Gateway Peering ACL Protocol Scoping", F: gatewayACLProtocolScopingTest, SkipFlags: SkipFlags{NoGateway: true, NoServers: true}},
+		{Name: "Gateway Peering ACL Packet One-Way", F: gatewayACLPacketOneWayTest, SkipFlags: SkipFlags{NoGateway: true, NoServers: true}},
+		{Name: "Gateway Peering ACL Flow Scope Masquerade", F: gatewayACLFlowScopeMasqueradeTest, SkipFlags: SkipFlags{NoGateway: true, NoServers: true}},
+		{Name: "Gateway Peering ACL Subnet/CIDR Scoping", F: gatewayACLSubnetScopingTest, SkipFlags: SkipFlags{NoGateway: true, NoServers: true}},
+		{Name: "Gateway Peering ACL Port Range Scoping", F: gatewayACLPortScopingTest, SkipFlags: SkipFlags{NoGateway: true, NoServers: true}},
+		{Name: "Gateway Peering ACL Precedence Allow-Then-Deny", F: gatewayACLPrecedenceAllowThenDenyTest, SkipFlags: SkipFlags{NoGateway: true, NoServers: true}},
+		{Name: "Gateway Peering ACL Precedence Deny-Then-Allow", F: gatewayACLPrecedenceDenyThenAllowTest, SkipFlags: SkipFlags{NoGateway: true, NoServers: true}},
+	}
+}
diff --git a/pkg/hhfab/rt_acl_tests.md b/pkg/hhfab/rt_acl_tests.md
new file mode 100644
index 000000000..7ecce90a9
--- /dev/null
+++ b/pkg/hhfab/rt_acl_tests.md
@@ -0,0 +1,86 @@
+# Gateway Peering ACL release tests
+
+Documents the ACL cases in `rt_acl_tests.go`: the probe infrastructure they run
+on, what each case asserts, and the coverage they add. Intended for reviewers
+assessing whether the coverage is correct and sufficient.
+
+## Probe infrastructure
+
+ACL enforcement is protocol/port-specific, so the tests drive the
+`ConnectivityMatrix` (`matrix.go`) keyed by `ProtoPort{Protocol, Port}` rather
+than the legacy default check.
+
+- **Expectations** are stamped on the matrix via `setVPCToVPCProtoVerdict(m,
+  srcVPC, dstVPC, ProtoPort, verdict)` (and the `setACLDirVerdicts` helper, which
+  sets the common icmp + tcp/5201 + udp/5201 triple for one direction).
+- A pair carrying any non-zero `ProtoPort` entry is owned entirely by
+  `runMatrixProtoPortPhase` and skipped by the legacy server-server phase
+  (`HasProtoPortEntries` gate), so every expected verdict — including ICMP — is an
+  explicit entry.
+- Each case is a `natTestSpec` run through `runNATTest`: `BuildSpec` attaches the
+  `PeeringACL` to the gateway peering (via `GwPeeringOptions.ACL`), then
+  `DoSetupPeerings → WaitReady → matrix.Repopulate → Overlay →
+  DoVLABTestConnectivityWithMatrix`.
+
+### Per-protocol probes
+
+| Protocol | Helper | Wire probe | Verdict signal |
+|---|---|---|---|
+| `icmp` | `checkPing` | `ping -c N` | all replies received = allow; zero = deny |
+| `tcp`  | `checkTCPPort` | `nc -zw2 <ip> <port>; echo NCRC=$?` | rc 0 = allow, rc 1 = deny; anything else / no marker / SSH error = surfaced as infra failure |
+| `udp`  | `checkUDPPort` | `iperf3 -u -J -c <ip> -p <port>` | datagrams delivered (loss < 90%) = allow; control-channel error / 0 packets / loss ≥ 99% = deny; unparseable output = surfaced |
+
+Port 5201 is served by the always-on `iperf3 -s` daemon (TCP+UDP). Any other port
+(e.g. the 6xxx range) gets an on-demand `iperf3 -s -p <port>` listener started by
+`startMatrixProtoPortListeners` and torn down at the end of the run.
+
+### Two facts that shape every case
+
+1. **Return path** — a working probe (ping reply, TCP handshake, iperf3 control
+   channel) needs *both* directions permitted. With `packet` (stateless) scope
+   the reply must be matched by its own rule, and because the reply has ports
+   swapped, a destination-port allow needs a matching **source-port** allow for
+   the reverse direction. A single one-way packet rule yields **no** connectivity.
+2. **`flow` scope requires stateful NAT** — the dataplane only keeps
+   flow/conntrack state where masquerade (or port-forward) NAT is present, so a
+   `flow`-scoped rule is only valid on such a peering; there conntrack permits the
+   return automatically. NAT-free cases therefore use explicit `packet` scope.
+
+## Test cases
+
+All cases peer the first two VPCs (`vpc1`, `vpc2`), default action `deny` unless
+noted, and run with `SkipFlags{NoGateway, NoServers}`.
+
+| Test | ACL under test | Expected (fwd = vpc1→vpc2, rev = vpc2→vpc1) |
+|---|---|---|
+| **Default Deny** | `deny` default + one allow rule on an unprobed port | all deny both ways (probed traffic hits the default) |
+| **Deny-Unless-Exposed UDP Carve-Out** | `deny-unless-exposed` default + `deny udp` both dirs | tcp+icmp allow both ways (exposed); udp deny both ways |
+| **Explicit Allow** | `allow` rules both dirs (any proto) | all allow both ways |
+| **Protocol Scoping** | `allow tcp` both dirs | tcp allow; udp+icmp deny (fall to default) |
+| **Packet One-Way** | single `allow` rule, one direction only, `packet` | all deny both ways — reply is dropped, so nothing completes |
+| **Flow Scope Masquerade** | masquerade NAT on vpc1 + `flow allow` vpc1→vpc2 | fwd allow; rev deny (masquerade blocks inbound + default deny) |
+| **Subnet/CIDR Scoping** | allow both dirs; fwd matched by `VPCSubnet`, rev by `CIDR` | all allow both ways |
+| **Port Range Scoping** | allow tcp; fwd dst-port range `6000-6500`, rev src-port range | fwd tcp/6201 allow; tcp/5201 + udp + icmp + all rev = deny |
+| **Precedence Allow-Then-Deny** | `[allow tcp/5201, deny-all]` fwd + src-port return rule | fwd tcp/5201 allow; everything else deny |
+| **Precedence Deny-Then-Allow** | same rules, `deny-all` first | all deny (first match wins) |
+
+## Coverage
+
+| Dimension | Covered by |
+|---|---|
+| Default action `deny` / `deny-unless-exposed` | Default Deny / Deny-Unless-Exposed |
+| Rule action `allow` / `deny` | all allow cases / carve-out + precedence |
+| Protocol `tcp` / `udp` / any | Protocol Scoping, Port Range / UDP Carve-Out / Explicit Allow, Subnet |
+| Selector `VPCSubnet` / `CIDR` | Subnet/CIDR Scoping (both, one per direction) |
+| Ports single / range, dst / src side | Precedence (single dst+src) / Port Range (range dst+src) |
+| Scope `packet` / `flow` | all packet cases / Flow Scope Masquerade |
+| Rule precedence (first-match) | Precedence Allow-Then-Deny + Deny-Then-Allow |
+| Stateless return-path requirement | Packet One-Way (negative) |
+
+### Known limitations / assumptions
+
+- **UDP is verifiable only as "denied while TCP allowed."** The `iperf3 -u` probe
+  opens a TCP control channel on the same port, so "deny TCP + allow UDP on one
+  port" cannot be distinguished from a UDP block — no case relies on it.
+- **ICMP falls to the default action** (it matches neither `tcp` nor `udp`
+  rules). Numeric-protocol matching (e.g. proto `1` for ICMP) is **not** covered.
diff --git a/pkg/hhfab/rt_multi_vpc_single_subnet_suite.go b/pkg/hhfab/rt_multi_vpc_single_subnet_suite.go
index 156c4bb4c..1ca972223 100644
--- a/pkg/hhfab/rt_multi_vpc_single_subnet_suite.go
+++ b/pkg/hhfab/rt_multi_vpc_single_subnet_suite.go
@@ -121,6 +121,8 @@ func makeMultiVPCSingleSubnetSuite() *JUnitTestSuite {
 	suite.TestCases = append(suite.TestCases, getNATTestCases()...)
 	// Add external NAT test cases
 	suite.TestCases = append(suite.TestCases, getExternalNATTestCases()...)
+	// Add gateway peering ACL test cases
+	suite.TestCases = append(suite.TestCases, getACLTestCases()...)
 	suite.Tests = len(suite.TestCases)
 
 	return suite
diff --git a/pkg/hhfab/rt_nat_tests.go b/pkg/hhfab/rt_nat_tests.go
index f6eccdcef..cdc172238 100644
--- a/pkg/hhfab/rt_nat_tests.go
+++ b/pkg/hhfab/rt_nat_tests.go
@@ -217,6 +217,38 @@ func overrideVPCToVPCVerdict(matrix *ConnectivityMatrix, srcVPCName, dstVPCName
 	}
 }
 
+// setVPCToVPCProtoVerdict adds a protocol/port-scoped expectation on every
+// (server-in-srcVPCName → server-in-dstVPCName) pair. Unlike
+// overrideVPCToVPCVerdict (which writes the default ProtoPort{} entry), this
+// keys the entry by pp, so the matrix-driven runner exercises it with a
+// protocol-specific probe (icmp/tcp/udp) in runMatrixProtoPortPhase. Multiple
+// calls with different pp accumulate on the same pair, letting a single peering
+// express e.g. TCP allow + UDP deny. pp must be non-zero. The pair's existing
+// default-entry Peering is preserved for diagnostics.
+func setVPCToVPCProtoVerdict(matrix *ConnectivityMatrix, srcVPCName, dstVPCName string, pp ProtoPort, verdict ConnectivityVerdict) {
+	srcPred := ServerInVPC(srcVPCName)
+	dstPred := ServerInVPC(dstVPCName)
+	for _, src := range matrix.AllEndpoints {
+		if !srcPred(src) {
+			continue
+		}
+		for _, dst := range matrix.AllEndpoints {
+			if !dstPred(dst) {
+				continue
+			}
+			existing := matrix.Lookup(src, dst, ProtoPort{})
+			matrix.Add(ConnectivityExpectation{
+				Pair:      EndpointPair{Source: src, Destination: dst},
+				Verdict:   verdict,
+				Reason:    ReachabilityReasonGatewayPeering,
+				Peering:   existing.Peering,
+				NAT:       existing.NAT,
+				ProtoPort: pp,
+			})
+		}
+	}
+}
+
 // rebindMatrixServerEndpoint refreshes the matrix's endpoint(s) for
 // serverName to reflect the current cluster state. Used after a runtime
 // attachment change (e.g., the overlap NAT test moves one server to a
diff --git a/pkg/hhfab/rt_utils.go b/pkg/hhfab/rt_utils.go
index 717e7dcec..1ca0038c8 100644
--- a/pkg/hhfab/rt_utils.go
+++ b/pkg/hhfab/rt_utils.go
@@ -769,6 +769,8 @@ type GwPeeringOptions struct {
 	VPC1PortForwardRules []gwapi.PeeringNATPortForwardEntry
 	// VPC2PortForwardRules specifies port-forwarding rules for VPC2
 	VPC2PortForwardRules []gwapi.PeeringNATPortForwardEntry
+	// ACL is an optional peering-scoped ACL applied to the gateway peering.
+	ACL *gwapi.PeeringACL
 }
 
 // GwExtPeeringOptions contains optional parameters for gateway external peering configuration
@@ -914,6 +916,7 @@ func appendGwPeeringSpec(gwPeerings map[string]*gwapi.PeeringSpec, vpc1, vpc2 *v
 				Expose: vpc2Exposes,
 			},
 		},
+		ACL: opts.ACL,
 	}
 
 	return nil
diff --git a/pkg/hhfab/testing.go b/pkg/hhfab/testing.go
index c386e290a..60f34c48c 100644
--- a/pkg/hhfab/testing.go
+++ b/pkg/hhfab/testing.go
@@ -3359,6 +3359,198 @@ func isReportableIperfError(ie *IperfError) bool {
 	return ie.ClientMsg != "" || ie.ServerMsg != ""
 }
 
+// ncProbeMarker prefixes the exit code the TCP probe echoes after nc, so the
+// connection result is read from the marker rather than from the SSH command's
+// own exit status (which conflates transport failures with a refused connect).
+const ncProbeMarker = "NCRC="
+
+// parseNCReturnCode extracts the nc exit code from the "NCRC=<n>" marker the TCP
+// probe appends. ok is false when no valid marker is present (i.e. the probe did
+// not run to completion).
+func parseNCReturnCode(stdout string) (int, bool) {
+	for _, line := range strings.Split(stdout, "\n") {
+		rest, found := strings.CutPrefix(strings.TrimSpace(line), ncProbeMarker)
+		if !found {
+			continue
+		}
+		rc, err := strconv.Atoi(strings.TrimSpace(rest))
+		if err != nil {
+			return 0, false
+		}
+
+		return rc, true
+	}
+
+	return 0, false
+}
+
+// checkTCPPort asserts that a TCP connection from `from` to toIP:port matches
+// expected. It runs `nc -zw2 <ip> <port>` on the source: a completed handshake
+// means the path is open (allow), a refused/timed-out connect (nc exit 1) means
+// it is blocked (deny). Unlike checkIPerf there is no throughput floor — this is
+// a pure reachability check for protocol/port-scoped (ProtoPort) matrix entries.
+//
+// The nc exit code is captured via a shell marker so a genuine connection result
+// is distinguished from an infrastructure failure: an SSH/transport error (no
+// marker, or the command could not run) or any nc exit code other than 0/1 is
+// surfaced as an error for BOTH allow and deny expectations, rather than being
+// silently treated as a successful "deny". Reported as *IperfError so it routes
+// like the other L4 probes.
+func checkTCPPort(ctx context.Context, sem *semaphore.Weighted, from string, fromSSH *sshutil.Config, toIP netip.Addr, port uint16, expected bool) *IperfError {
+	target := fmt.Sprintf("%s:%d", toIP.String(), port)
+	ie := &IperfError{Source: from, Destination: target}
+
+	if sem != nil {
+		if err := sem.Acquire(ctx, 1); err != nil {
+			ie.ClientMsg = fmt.Sprintf("acquiring iperf3 semaphore: %s", err)
+
+			return ie
+		}
+		defer sem.Release(1)
+	}
+
+	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+
+	// Append the marker so the shell exits 0 whenever it ran; a non-nil err
+	// then unambiguously means the probe itself could not execute.
+	cmd := fmt.Sprintf("nc -zw2 %s %d; echo %s$?", toIP.String(), port, ncProbeMarker)
+	stdout, stderr, err := retrySSHCmd(ctx, fromSSH, cmd, from)
+	if err != nil {
+		ie.ClientMsg = fmt.Sprintf("TCP probe could not run: %s: %s", err, strings.TrimSpace(stderr))
+
+		return ie
+	}
+
+	rc, ok := parseNCReturnCode(stdout)
+	if !ok {
+		ie.ClientMsg = fmt.Sprintf("TCP probe produced no result marker (stdout %q, stderr %q)", strings.TrimSpace(stdout), strings.TrimSpace(stderr))
+
+		return ie
+	}
+
+	var connectOk bool
+	switch rc {
+	case 0:
+		connectOk = true
+	case 1:
+		// nc -z exits 1 on connection refused or -w timeout: a genuine block.
+		connectOk = false
+	default:
+		// e.g. 127 (nc not found) — a probe/environment failure, not a verdict.
+		ie.ClientMsg = fmt.Sprintf("TCP probe returned unexpected exit code %d (stdout %q, stderr %q)", rc, strings.TrimSpace(stdout), strings.TrimSpace(stderr))
+
+		return ie
+	}
+
+	slog.Debug("TCP port probe result", "from", from, "to", target, "expected", expected, "ok", connectOk, "rc", rc, "stderr", stderr)
+
+	if expected && !connectOk {
+		ie.ClientMsg = "should be reachable but TCP connect was refused/timed out"
+
+		return ie
+	}
+	if !expected && connectOk {
+		ie.ClientMsg = "should not be reachable but TCP connect succeeded"
+
+		return ie
+	}
+
+	return nil
+}
+
+// udpDenyLossThreshold and udpAllowLossThreshold bound the datagram-loss
+// interpretation of a UDP probe. iperf3 -u runs even when every datagram is
+// dropped, so allow/deny is inferred from loss: near-total loss (or a control-
+// channel error / zero datagrams) means the path is blocked, low-enough loss
+// means it is open. The allow bound is deliberately generous because VS /
+// CumulusVX links drop real packets.
+const (
+	udpDenyLossThreshold  = 99.0
+	udpAllowLossThreshold = 90.0
+)
+
+// checkUDPPort asserts that UDP datagrams from `from` to toIP:port match
+// expected. It runs iperf3 -u inside the always-on iperf3 container (same
+// memory-safe docker-exec path as runIPerf3Test) and interprets datagram loss.
+// No throughput floor.
+//
+// NOTE: iperf3 -u first opens a TCP control channel on the target port, so this
+// cannot distinguish "TCP denied + UDP allowed" on the same port (the control
+// channel would be blocked). Callers must avoid that combination.
+func checkUDPPort(ctx context.Context, opts TestConnectivityOpts, sem *semaphore.Weighted, from string, fromSSH *sshutil.Config, toIP netip.Addr, port uint16, expected bool) *IperfError {
+	target := fmt.Sprintf("%s:%d", toIP.String(), port)
+	ie := &IperfError{Source: from, Destination: target}
+
+	if sem != nil {
+		if err := sem.Acquire(ctx, 1); err != nil {
+			ie.ClientMsg = fmt.Sprintf("acquiring iperf3 semaphore: %s", err)
+
+			return ie
+		}
+		defer sem.Release(1)
+	}
+
+	secs := opts.IPerfsSeconds
+	if secs <= 0 {
+		secs = 3
+	}
+	ctx, cancel := context.WithTimeout(ctx, time.Duration(secs+30)*time.Second)
+	defer cancel()
+
+	cmd := fmt.Sprintf("sudo docker exec iperf3 timeout %d iperf3 -u -J -c %s -p %d -t %d -b 10M -l 1000", secs+25, toIP.String(), port, secs)
+	stdout, stderr, err := retrySSHCmd(ctx, fromSSH, cmd, from)
+	report, parseErr := parseIPerf3Report([]byte(stdout))
+
+	// No parseable iperf3 JSON means the probe did not run to completion
+	// (docker/exec/SSH failure, OOM, or a killed process) — surface it rather
+	// than treating it as a traffic verdict. iperf3 -J emits a JSON report with
+	// an "error" field even when the control channel is refused, so a genuine
+	// denial still parses and is classified below.
+	if parseErr != nil {
+		ie.ClientMsg = fmt.Sprintf("iperf3 UDP probe produced no parseable report (cmd err: %v, stderr: %q): %s", err, strings.TrimSpace(stderr), parseErr)
+
+		return ie
+	}
+
+	reportErr := report.Error
+	packets := report.End.Sum.Packets
+	lost := report.End.Sum.LostPackets
+	lostPercent := report.End.Sum.LostPercent
+	delivered := reportErr == "" && packets > 0 && lostPercent < udpAllowLossThreshold
+	// A blocked path is a validated control-channel denial (report.Error, e.g.
+	// "Connection refused"), zero datagrams, or near-total loss — never an
+	// unexplained command/parse error.
+	blocked := reportErr != "" || packets == 0 || lostPercent >= udpDenyLossThreshold
+
+	slog.Debug("UDP port probe result", "from", from, "to", target, "expected", expected,
+		"delivered", delivered, "blocked", blocked, "packets", packets, "lost", lost, "lostPercent", lostPercent,
+		"err", err, "reportErr", reportErr, "stderr", stderr)
+
+	if expected {
+		if !delivered {
+			if reportErr != "" {
+				ie.ClientMsg = fmt.Sprintf("should be reachable but UDP probe reported error: %s", reportErr)
+			} else {
+				ie.ClientMsg = fmt.Sprintf("should be reachable but UDP datagrams not delivered (packets %d, loss %.1f%%)", packets, lostPercent)
+			}
+
+			return ie
+		}
+
+		return nil
+	}
+
+	// expected == deny.
+	if !blocked {
+		ie.ClientMsg = fmt.Sprintf("should not be reachable but UDP datagrams delivered (packets %d, loss %.1f%%)", packets, lostPercent)
+
+		return ie
+	}
+
+	return nil
+}
+
 func checkCurl(ctx context.Context, opts TestConnectivityOpts, curls *semaphore.Weighted, from string, fromSSH *sshutil.Config, toIP string, expected bool) *CurlError {
 	if opts.CurlsCount <= 0 {
 		return nil
@@ -3432,11 +3624,19 @@ type iperf3ReportEnd struct {
 	SumReceived             iperf3ReportSum `json:"sum_received"`
 	SumSentBidirReverse     iperf3ReportSum `json:"sum_sent_bidir_reverse"`
 	SumReceivedBidirReverse iperf3ReportSum `json:"sum_received_bidir_reverse"`
+	// Sum holds the UDP client summary (iperf3 -u reports it here rather than
+	// in sum_sent/sum_received). Carries the datagram/loss counters.
+	Sum iperf3ReportSum `json:"sum"`
 }
 
 type iperf3ReportSum struct {
 	Bytes         int64   `json:"bytes"`
 	BitsPerSecond float64 `json:"bits_per_second"`
+	// UDP-only fields (populated by iperf3 -u).
+	Packets     int64   `json:"packets"`
+	LostPackets int64   `json:"lost_packets"`
+	LostPercent float64 `json:"lost_percent"`
+	JitterMs    float64 `json:"jitter_ms"`
 }
 
 func parseIPerf3Report(data []byte) (*iperf3Report, error) {