Merge branch 'main' into feat/execd-websocket-pty

Author: Pangjiping

components/egress/pkg/nftables/dynamic.go

@@ -24,9 +24,12 @@ import (
 const (
 	dynAllowV4Set  = "dyn_allow_v4"
 	dynAllowV6Set  = "dyn_allow_v6"
-	dynSetTimeoutS = 300
+	dynSetTimeoutS = 360
+	// nftTTLSlackSec is added to the DNS TTL before clamping, so allow entries
+	// slightly outlive the resolver cache and reduce races with short TTLs.
+	nftTTLSlackSec = 60
 	minTTLSec      = 60
-	maxTTLSec      = 300
+	maxTTLSec      = 360 // max DNS TTL (300) + nftTTLSlackSec
 )
 
 // ResolvedIP is a single IP learned from DNS with TTL for dynamic nft set.
@@ -58,7 +61,7 @@ func buildAddResolvedIPsScript(table string, ips []ResolvedIP) string {
 }
 
 func clampTTL(d time.Duration) int {
-	sec := int(d.Seconds())
+	sec := int(d.Seconds()) + nftTTLSlackSec
 	if sec < minTTLSec {
 		return minTTLSec
 	}

components/egress/pkg/nftables/interval.go

@@ -0,0 +1,112 @@
+// Copyright 2026 Alibaba Group Holding Ltd.
+//
+// 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 nftables
+
+import (
+	"fmt"
+	"net/netip"
+)
+
+// normalizeNFTIntervalSet drops entries that are redundant for nftables sets with
+// `flags interval`: a host or smaller CIDR that falls strictly inside another
+// listed CIDR would make `nft add element` fail with "conflicting intervals specified".
+// Used on every ApplyStatic (startup and /policy updates).
+func normalizeNFTIntervalSet(elems []string) ([]string, error) {
+	if len(elems) == 0 {
+		return nil, nil
+	}
+	prefs := make([]netip.Prefix, 0, len(elems))
+	for _, s := range elems {
+		p, err := parseAsPrefix(s)
+		if err != nil {
+			return nil, err
+		}
+		prefs = append(prefs, p.Masked())
+	}
+	prefs = uniquePrefixes(prefs)
+	prefs = removeStrictSubnets(prefs)
+	out := make([]string, 0, len(prefs))
+	for _, p := range prefs {
+		out = append(out, formatPrefixForNFT(p))
+	}
+	return out, nil
+}
+
+func parseAsPrefix(s string) (netip.Prefix, error) {
+	if p, err := netip.ParsePrefix(s); err == nil {
+		return p, nil
+	}
+	if a, err := netip.ParseAddr(s); err == nil {
+		if a.Is4() {
+			return a.Prefix(32)
+		}
+		return a.Prefix(128)
+	}
+	return netip.Prefix{}, fmt.Errorf("nftables: invalid IP or CIDR %q", s)
+}
+
+func uniquePrefixes(prefs []netip.Prefix) []netip.Prefix {
+	seen := make(map[string]struct{}, len(prefs))
+	out := make([]netip.Prefix, 0, len(prefs))
+	for _, p := range prefs {
+		k := p.String()
+		if _, ok := seen[k]; ok {
+			continue
+		}
+		seen[k] = struct{}{}
+		out = append(out, p)
+	}
+	return out
+}
+
+// strictSupernet reports whether super covers all addresses of sub, with super
+// strictly larger (smaller prefix length) than sub.
+func strictSupernet(super, sub netip.Prefix) bool {
+	if super.Addr().Is4() != sub.Addr().Is4() {
+		return false
+	}
+	if super.Bits() >= sub.Bits() {
+		return false
+	}
+	return super.Contains(sub.Addr())
+}
+
+func removeStrictSubnets(prefs []netip.Prefix) []netip.Prefix {
+	out := make([]netip.Prefix, 0, len(prefs))
+	for _, p := range prefs {
+		redundant := false
+		for _, q := range prefs {
+			if strictSupernet(q, p) {
+				redundant = true
+				break
+			}
+		}
+		if !redundant {
+			out = append(out, p)
+		}
+	}
+	return out
+}
+
+func formatPrefixForNFT(p netip.Prefix) string {
+	p = p.Masked()
+	if p.Addr().Is4() && p.Bits() == 32 {
+		return p.Addr().String()
+	}
+	if p.Addr().Is6() && p.Bits() == 128 {
+		return p.Addr().String()
+	}
+	return p.String()
+}

components/egress/pkg/nftables/interval_test.go

@@ -0,0 +1,58 @@
+// Copyright 2026 Alibaba Group Holding Ltd.
+//
+// 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 nftables
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNormalizeNFTIntervalSet_cgnatContainsNameservers(t *testing.T) {
+	got, err := normalizeNFTIntervalSet([]string{
+		"100.64.0.0/10",
+		"127.0.0.1",
+		"100.100.2.136",
+		"100.100.2.138",
+	})
+	require.NoError(t, err)
+	require.Equal(t, []string{"100.64.0.0/10", "127.0.0.1"}, got)
+}
+
+func TestNormalizeNFTIntervalSet_noChange(t *testing.T) {
+	got, err := normalizeNFTIntervalSet([]string{"1.1.1.1", "2.2.0.0/16"})
+	require.NoError(t, err)
+	require.Equal(t, []string{"1.1.1.1", "2.2.0.0/16"}, got)
+}
+
+func TestNormalizeNFTIntervalSet_dedupe(t *testing.T) {
+	got, err := normalizeNFTIntervalSet([]string{"8.8.8.8", "8.8.8.8"})
+	require.NoError(t, err)
+	require.Equal(t, []string{"8.8.8.8"}, got)
+}
+
+func TestNormalizeNFTIntervalSet_ipv6(t *testing.T) {
+	got, err := normalizeNFTIntervalSet([]string{
+		"2001:db8::/32",
+		"2001:db8::1",
+	})
+	require.NoError(t, err)
+	require.Equal(t, []string{"2001:db8::/32"}, got)
+}
+
+func TestNormalizeNFTIntervalSet_invalid(t *testing.T) {
+	_, err := normalizeNFTIntervalSet([]string{"not-an-ip"})
+	require.Error(t, err)
+}

components/egress/pkg/nftables/manager.go

@@ -82,6 +82,9 @@ func NewManagerWithOptions(opts Options) *Manager {
 // Uses the same mutex as AddResolvedIPs so a /policy update never overlaps a DNS
 // callback: without this, add-element could run while the table is being deleted/recreated
 // and fail, causing a transient deny for a client that already got an allowed DNS answer.
+//
+// On every call (startup and /policy updates), static allow/deny and DoH blocklist
+// interval sets are normalized so overlapping CIDR/host pairs do not make nft fail.
 func (m *Manager) ApplyStatic(ctx context.Context, p *policy.NetworkPolicy) error {
 	if p == nil {
 		p = policy.DefaultDenyPolicy()
@@ -91,7 +94,10 @@ func (m *Manager) ApplyStatic(ctx context.Context, p *policy.NetworkPolicy) erro
 		p.DefaultAction, len(allowV4), len(allowV6), len(denyV4), len(denyV6))
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	script := buildRuleset(p, m.opts)
+	script, err := buildRuleset(p, m.opts)
+	if err != nil {
+		return err
+	}
 	if _, err := m.run(ctx, script); err != nil {
 		// On a fresh host the delete-table may fail; retry once without the delete line.
 		if isMissingTableError(err) {
@@ -109,7 +115,8 @@ func (m *Manager) ApplyStatic(ctx context.Context, p *policy.NetworkPolicy) erro
 }
 
 // AddResolvedIPs adds DNS-learned IPs to dynamic allow sets with TTL-based timeout.
-// TTL is clamped to minTTLSec–maxTTLSec. Call only when table exists (dns+nft mode).
+// Each element timeout is DNS TTL + 60s, then clamped to minTTLSec–maxTTLSec.
+// Call only when table exists (dns+nft mode).
 func (m *Manager) AddResolvedIPs(ctx context.Context, ips []ResolvedIP) error {
 	if len(ips) == 0 {
 		return nil
@@ -126,8 +133,33 @@ func (m *Manager) AddResolvedIPs(ctx context.Context, ips []ResolvedIP) error {
 	return err
 }
 
-func buildRuleset(p *policy.NetworkPolicy, opts Options) string {
+func buildRuleset(p *policy.NetworkPolicy, opts Options) (string, error) {
 	allowV4, allowV6, denyV4, denyV6 := p.StaticIPSets()
+	var err error
+	if allowV4, err = normalizeNFTIntervalSet(allowV4); err != nil {
+		return "", err
+	}
+	if allowV6, err = normalizeNFTIntervalSet(allowV6); err != nil {
+		return "", err
+	}
+	if denyV4, err = normalizeNFTIntervalSet(denyV4); err != nil {
+		return "", err
+	}
+	if denyV6, err = normalizeNFTIntervalSet(denyV6); err != nil {
+		return "", err
+	}
+	dohBlockV4 := opts.DoHBlocklistV4
+	dohBlockV6 := opts.DoHBlocklistV6
+	if len(dohBlockV4) > 0 {
+		if dohBlockV4, err = normalizeNFTIntervalSet(dohBlockV4); err != nil {
+			return "", err
+		}
+	}
+	if len(dohBlockV6) > 0 {
+		if dohBlockV6, err = normalizeNFTIntervalSet(dohBlockV6); err != nil {
+			return "", err
+		}
+	}
 
 	var b strings.Builder
 	// Reset and re-create table, sets, and chain.
@@ -141,19 +173,19 @@ func buildRuleset(p *policy.NetworkPolicy, opts Options) string {
 	fmt.Fprintf(&b, "add set inet %s %s { type ipv4_addr; timeout %ds; }\n", tableName, dynAllowV4Set, dynSetTimeoutS)
 	fmt.Fprintf(&b, "add set inet %s %s { type ipv6_addr; timeout %ds; }\n", tableName, dynAllowV6Set, dynSetTimeoutS)
 
-	if len(opts.DoHBlocklistV4) > 0 {
+	if len(dohBlockV4) > 0 {
 		fmt.Fprintf(&b, "add set inet %s %s { type ipv4_addr; flags interval; }\n", tableName, dohBlockV4Set)
 	}
-	if len(opts.DoHBlocklistV6) > 0 {
+	if len(dohBlockV6) > 0 {
 		fmt.Fprintf(&b, "add set inet %s %s { type ipv6_addr; flags interval; }\n", tableName, dohBlockV6Set)
 	}
 
 	writeElements(&b, allowV4Set, allowV4)
 	writeElements(&b, denyV4Set, denyV4)
 	writeElements(&b, allowV6Set, allowV6)
 	writeElements(&b, denyV6Set, denyV6)
-	writeElements(&b, dohBlockV4Set, opts.DoHBlocklistV4)
-	writeElements(&b, dohBlockV6Set, opts.DoHBlocklistV6)
+	writeElements(&b, dohBlockV4Set, dohBlockV4)
+	writeElements(&b, dohBlockV6Set, dohBlockV6)
 
 	chainPolicy := "drop"
 	if p.DefaultAction == policy.ActionAllow {
@@ -168,14 +200,14 @@ func buildRuleset(p *policy.NetworkPolicy, opts Options) string {
 		fmt.Fprintf(&b, "add rule inet %s %s udp dport 853 drop\n", tableName, chainName)
 	}
 	if opts.BlockDoH443 {
-		if len(opts.DoHBlocklistV4) == 0 && len(opts.DoHBlocklistV6) == 0 {
+		if len(dohBlockV4) == 0 && len(dohBlockV6) == 0 {
 			// strict: drop all 443 when enabled but no blocklist provided
 			fmt.Fprintf(&b, "add rule inet %s %s tcp dport 443 drop\n", tableName, chainName)
 		} else {
-			if len(opts.DoHBlocklistV4) > 0 {
+			if len(dohBlockV4) > 0 {
 				fmt.Fprintf(&b, "add rule inet %s %s ip daddr @%s tcp dport 443 drop\n", tableName, chainName, dohBlockV4Set)
 			}
-			if len(opts.DoHBlocklistV6) > 0 {
+			if len(dohBlockV6) > 0 {
 				fmt.Fprintf(&b, "add rule inet %s %s ip6 daddr @%s tcp dport 443 drop\n", tableName, chainName, dohBlockV6Set)
 			}
 		}
@@ -190,7 +222,7 @@ func buildRuleset(p *policy.NetworkPolicy, opts Options) string {
 		fmt.Fprintf(&b, "add rule inet %s %s counter drop\n", tableName, chainName)
 	}
 
-	return b.String()
+	return b.String(), nil
 }
 
 func writeElements(b *strings.Builder, setName string, elems []string) {

components/egress/pkg/nftables/manager_test.go

@@ -50,8 +50,8 @@ func TestApplyStatic_BuildsRuleset_DefaultDeny(t *testing.T) {
 	expectContains(t, rendered, "add rule inet opensandbox egress oifname \"lo\" accept")
 	expectContains(t, rendered, "add rule inet opensandbox egress tcp dport 853 drop")
 	expectContains(t, rendered, "add rule inet opensandbox egress udp dport 853 drop")
-	expectContains(t, rendered, "add set inet opensandbox dyn_allow_v4 { type ipv4_addr; timeout 300s; }")
-	expectContains(t, rendered, "add set inet opensandbox dyn_allow_v6 { type ipv6_addr; timeout 300s; }")
+	expectContains(t, rendered, "add set inet opensandbox dyn_allow_v4 { type ipv4_addr; timeout 360s; }")
+	expectContains(t, rendered, "add set inet opensandbox dyn_allow_v6 { type ipv6_addr; timeout 360s; }")
 	expectContains(t, rendered, "add element inet opensandbox allow_v4 { 1.1.1.1, 2.2.0.0/16 }")
 	expectContains(t, rendered, "add element inet opensandbox deny_v6 { 2001:db8::/32 }")
 	expectContains(t, rendered, "add rule inet opensandbox egress ip daddr @dyn_allow_v4 accept")
@@ -137,8 +137,8 @@ func TestAddResolvedIPs_BuildsDynamicElements(t *testing.T) {
 		{Addr: netip.MustParseAddr("2001:db8::1"), TTL: 60 * time.Second},
 	}
 	require.NoError(t, m.AddResolvedIPs(context.Background(), ips), "AddResolvedIPs returned error")
-	expectContains(t, rendered, "add element inet opensandbox dyn_allow_v4 { 1.1.1.1 timeout 120s }")
-	expectContains(t, rendered, "add element inet opensandbox dyn_allow_v6 { 2001:db8::1 timeout 60s }")
+	expectContains(t, rendered, "add element inet opensandbox dyn_allow_v4 { 1.1.1.1 timeout 180s }")
+	expectContains(t, rendered, "add element inet opensandbox dyn_allow_v6 { 2001:db8::1 timeout 120s }")
 }
 
 func TestAddResolvedIPs_ClampsTTL(t *testing.T) {
@@ -152,8 +152,8 @@ func TestAddResolvedIPs_ClampsTTL(t *testing.T) {
 		{Addr: netip.MustParseAddr("10.0.0.2"), TTL: 9999 * time.Second},
 	}
 	require.NoError(t, m.AddResolvedIPs(context.Background(), ips), "AddResolvedIPs returned error")
-	expectContains(t, rendered, "10.0.0.1 timeout 60s")
-	expectContains(t, rendered, "10.0.0.2 timeout 300s")
+	expectContains(t, rendered, "10.0.0.1 timeout 70s")
+	expectContains(t, rendered, "10.0.0.2 timeout 360s")
 }
 
 func TestAddResolvedIPs_EmptyNoOp(t *testing.T) {
@@ -164,3 +164,21 @@ func TestAddResolvedIPs_EmptyNoOp(t *testing.T) {
 	require.NoError(t, m.AddResolvedIPs(context.Background(), nil), "AddResolvedIPs returned error")
 	require.NoError(t, m.AddResolvedIPs(context.Background(), []ResolvedIP{}), "AddResolvedIPs returned error")
 }
+
+func TestApplyStatic_NormalizesOverlappingAllow(t *testing.T) {
+	var rendered string
+	m := NewManagerWithRunner(func(_ context.Context, script string) ([]byte, error) {
+		rendered = script
+		return nil, nil
+	})
+	p, err := policy.ParsePolicy(`{
+		"defaultAction":"deny",
+		"egress":[
+			{"action":"allow","target":"100.64.0.0/10"},
+			{"action":"allow","target":"100.100.2.136"}
+		]
+	}`)
+	require.NoError(t, err)
+	require.NoError(t, m.ApplyStatic(context.Background(), p))
+	expectContains(t, rendered, "add element inet opensandbox allow_v4 { 100.64.0.0/10 }")
+}