bridge: batch firewall updates for published ports

Allocate all requested port bindings before applying their firewall rules, so nftables rewrites each affected chain once per address family rather than once per port group

Keep grouped port allocation, start proxies after the firewall update, and roll back resources and installed rules if either stage fails

Extend the port-mapping tests to check a single firewall update and verify cleanup and rebinding after setup failures

Signed-off-by: akasakariko <148060537+akasakariko@users.noreply.github.com>
This commit is contained in:
akasakariko committed 2026-10-27 00:28:11 +09:00
1 parent b711d21792
commit e83b503590
2 files changed
+110 -62

No files matched your search

@@ -10,7 +10,6 @@ import (
"slices"
"github.com/containerd/log"
"github.com/moby/moby/v2/daemon/libnetwork/drvregistry"
"github.com/moby/moby/v2/daemon/libnetwork/netutils"
"github.com/moby/moby/v2/daemon/libnetwork/portallocator"
"github.com/moby/moby/v2/daemon/libnetwork/portmapper"
@@ -59,7 +58,7 @@ func (n *bridgeNetwork) addPortMappings(
bindings := make([]portmapperapi.PortBinding, 0, len(cfg)*2)
defer func() {
if retErr != nil {
if err := n.unmapPBs(ctx, bindings); err != nil {
if err := n.unmapPorts(ctx, bindings); err != nil {
log.G(ctx).WithFields(log.Fields{
"bindings": bindings,
"error": err,
@@ -89,55 +88,36 @@ func (n *bridgeNetwork) addPortMappings(
continue
}
newB, err := n.mapPorts(ctx, pms, toBind)
mapper := c.Mapper
pm, err := pms.Get(mapper)
if err != nil {
return nil, err
}
newB, err := pm.MapPorts(ctx, toBind)
if err != nil {
return nil, err
}
for i := range newB {
// Record the mapper so rollback and release use the same port mapper.
newB[i].Mapper = mapper
}
bindings = append(bindings, newB...)
// Reset toBind now the ports are bound.
toBind = toBind[:0]
}
return bindings, nil
}
// mapPorts calls the port mapper used to map the ports in reqs, applies the firewall rules requested by that portmapper,
// and starts userland proxies if needed. It returns an error if it fails on any of these steps, and rolls back any
// changes it made. Caller must ensure that reqs is non-empty and all requests have the same Mapper set.
func (n *bridgeNetwork) mapPorts(ctx context.Context, pms *drvregistry.PortMappers, reqs []portmapperapi.PortBindingReq) (_ []portmapperapi.PortBinding, retErr error) {
mapper := reqs[0].Mapper
pm, err := pms.Get(mapper)
if err != nil {
return nil, err
}
bindings, err := pm.MapPorts(ctx, reqs)
if err != nil {
return nil, err
}
defer func() {
if retErr != nil {
if err := pm.UnmapPorts(ctx, bindings); err != nil {
log.G(ctx).WithFields(log.Fields{
"bindings": bindings,
"error": err,
"origErr": retErr,
}).Warn("Failed to unmap port bindings after error")
}
return
}
}()
for i := range bindings {
// Make sure that Mapper is correctly set such that UnmapPorts call the right portmapper.
bindings[i].Mapper = mapper
if len(bindings) == 0 {
return bindings, nil
}
// Apply all port rules together, so nftables only rewrites the affected chains
// once per address family, rather than once for each group of bindings.
fwPorts := collectFirewallPorts(bindings)
if err := n.firewallerNetwork.AddPorts(ctx, fwPorts); err != nil {
return nil, err
}
// AddPorts can fail after installing rules for one address family.
// With nftables, a failed per-family transaction leaves no rules for that
// family. Deleting those missing rules may log a warning below even when
// all successfully installed rules have been removed.
defer func() {
if retErr != nil {
if err := n.firewallerNetwork.DelPorts(ctx, fwPorts); err != nil {
@@ -149,23 +129,11 @@ func (n *bridgeNetwork) mapPorts(ctx context.Context, pms *drvregistry.PortMappe
}
}
}()
if err := n.firewallerNetwork.AddPorts(ctx, fwPorts); err != nil {
return nil, err
}
// Start userland proxy processes.
defer func() {
if retErr != nil {
for _, pb := range bindings {
if pb.StopProxy == nil {
continue
}
if err := pb.StopProxy(); err != nil {
log.G(ctx).WithFields(log.Fields{
"binding": pb.PortBinding,
"error": err,
}).Warnf("failed to stop userland proxy for port mapping")
}
}
}
}()
// Start userland proxies only after all firewall rules have been installed.
if n.driver.config.EnableProxy {
for i, pb := range bindings {
if pb.BoundSocket == nil || pb.RootlesskitUnsupported || pb.StopProxy != nil {
@@ -426,6 +394,14 @@ func (n *bridgeNetwork) releasePorts(ep *bridgeEndpoint) error {
}
func (n *bridgeNetwork) unmapPBs(ctx context.Context, bindings []portmapperapi.PortBinding) error {
err := n.unmapPorts(ctx, bindings)
fwErr := n.firewallerNetwork.DelPorts(ctx, collectFirewallPorts(bindings))
return errors.Join(err, fwErr)
}
// unmapPorts releases port mapper resources and stops proxies without changing
// firewall rules, which may not have been installed yet during rollback.
func (n *bridgeNetwork) unmapPorts(ctx context.Context, bindings []portmapperapi.PortBinding) error {
pms := n.portMappers()
var errs []error
@@ -446,10 +422,6 @@ func (n *bridgeNetwork) unmapPBs(ctx context.Context, bindings []portmapperapi.P
}
}
if err := n.firewallerNetwork.DelPorts(ctx, collectFirewallPorts(bindings)); err != nil {
return err
}
return errors.Join(errs...)
}
@@ -220,6 +220,9 @@ func TestAddPortMappings(t *testing.T) {
enableProxy bool
hairpin bool
busyPortIPv4 int
proxyErrPort uint16
firewallErr bool
delPortsErr error
newPDC func() nat.PortDriverClient
hostAddrs []string
noProxy6To4 bool
@@ -283,6 +286,30 @@ func TestAddPortMappings(t *testing.T) {
busyPortIPv4: 8080,
expErr: "failed to bind host port 0.0.0.0:8080/tcp: address already in use",
},
{
name: "firewall failure after installing port rules",
epAddrV4: ctrIP4,
epAddrV6: ctrIP6,
cfg: []portmapperapi.PortBindingReq{
{PortBinding: types.PortBinding{Proto: types.TCP, Port: 22, HostPort: 2222}},
{PortBinding: types.PortBinding{Proto: types.TCP, Port: 80, HostPort: 8080}},
},
enableProxy: true,
firewallErr: true,
expErr: "firewall failed",
},
{
name: "proxy failure after installing all port rules",
epAddrV4: ctrIP4,
epAddrV6: ctrIP6,
cfg: []portmapperapi.PortBindingReq{
{PortBinding: types.PortBinding{Proto: types.TCP, Port: 22, HostPort: 2222}},
{PortBinding: types.PortBinding{Proto: types.TCP, Port: 80, HostPort: 8080}},
},
enableProxy: true,
proxyErrPort: 80,
expErr: "failed to start userland proxy for port mapping 0.0.0.0:8080:172.19.0.2:80/tcp: proxy failed",
},
{
name: "ipv4 mapped container address with specific host port",
epAddrV4: ctrIP4Mapped,
@@ -463,6 +490,7 @@ func TestAddPortMappings(t *testing.T) {
{PortBinding: types.PortBinding{Proto: types.TCP, Port: 22, HostPort: 2222}},
},
enableProxy: true,
delPortsErr: errors.New("firewall cleanup failed"),
expPBs: []types.PortBinding{
{Proto: types.TCP, IP: ctrIP4.IP, Port: 22, HostIP: net.IPv4zero, HostPort: 2222},
{Proto: types.TCP, IP: ctrIP6.IP, Port: 22, HostIP: net.IPv6zero, HostPort: 2222},
@@ -472,7 +500,8 @@ func TestAddPortMappings(t *testing.T) {
expReleaseErr: "unmapping port binding 0.0.0.0:2222:172.19.0.2:22/tcp: failed to stop userland proxy: can't stop now\n" +
"unmapping port binding [::]:2222:[fdf8:b88e:bb5c:3483::2]:22/tcp: failed to stop userland proxy: can't stop now\n" +
"unmapping port binding 0.0.0.0:8080:172.19.0.2:80/tcp: failed to stop userland proxy: can't stop now\n" +
"unmapping port binding [::]:8080:[fdf8:b88e:bb5c:3483::2]:80/tcp: failed to stop userland proxy: can't stop now",
"unmapping port binding [::]:8080:[fdf8:b88e:bb5c:3483::2]:80/tcp: failed to stop userland proxy: can't stop now\n" +
"firewall cleanup failed",
},
{
name: "disable nat6",
@@ -717,6 +746,9 @@ func TestAddPortMappings(t *testing.T) {
// Mock the startProxy function used by the code under test.
proxies := map[proxyCall]bool{} // proxy -> is not stopped
startProxy = func(pb types.PortBinding, _ string, listenSock *os.File) (stop func() error, retErr error) {
if pb.Port == tc.proxyErrPort {
return nil, errors.New("proxy failed")
}
if tc.busyPortIPv4 > 0 && tc.busyPortIPv4 == int(pb.HostPort) && pb.HostIP.To4() != nil {
return nil, errors.New("busy port")
}
@@ -792,7 +824,11 @@ func TestAddPortMappings(t *testing.T) {
fwn, err := n.newFirewallerNetwork(t.Context())
assert.NilError(t, err)
assert.Check(t, fwn != nil, "no firewaller network")
n.firewallerNetwork = fwn
fwCalls := &portMappingFirewaller{Network: fwn, delPortsErr: tc.delPortsErr}
if tc.firewallErr {
fwCalls.addPortsErr = errors.New("firewall failed")
}
n.firewallerNetwork = fwCalls
expChildIP := func(hostIP net.IP) net.IP {
if pdc == nil {
@@ -833,9 +869,27 @@ func TestAddPortMappings(t *testing.T) {
pbs, err := n.addPortMappings(ctx, ep, tc.cfg, tc.defHostIP, pbm, pbm)
if tc.expErr != "" {
assert.ErrorContains(t, err, tc.expErr)
assert.Check(t, is.Len(fwn.(*firewaller.StubFirewallerNetwork).Ports, 0))
for proxy, running := range proxies {
assert.Check(t, !running, "proxy still running: %v", proxy)
}
if tc.proxyErrPort != 0 || tc.firewallErr {
// Retry the same bindings to check all host ports were released.
tc.proxyErrPort = 0
fwCalls.addPortsErr = nil
clear(proxies)
pbs, err = n.addPortMappings(ctx, ep, tc.cfg, tc.defHostIP, pbm, pbm)
assert.NilError(t, err)
assert.Check(t, n.releasePorts(&bridgeEndpoint{portMapping: pbs}))
}
return
}
assert.NilError(t, err)
expAddPortsCalls := 0
if len(pbs) > 0 {
expAddPortsCalls = 1
}
assert.Check(t, is.Equal(fwCalls.addPortsCalls, expAddPortsCalls))
for _, expLog := range tc.expLogs {
assert.Check(t, is.Contains(sb.String(), expLog))
}
@@ -846,7 +900,7 @@ func TestAddPortMappings(t *testing.T) {
assert.Check(t, fw.IPv4)
assert.Check(t, fw.IPv6)
fnw := n.firewallerNetwork.(*firewaller.StubFirewallerNetwork)
fnw := fwn.(*firewaller.StubFirewallerNetwork)
assert.Check(t, !fnw.Internal)
assert.Check(t, !fnw.ICC)
assert.Check(t, !fnw.Masquerade)
@@ -924,6 +978,28 @@ func TestAddPortMappings(t *testing.T) {
}
}
type portMappingFirewaller struct {
firewaller.Network
addPortsCalls int
addPortsErr error
delPortsErr error
}
func (fw *portMappingFirewaller) AddPorts(ctx context.Context, pbs []types.PortBinding) error {
fw.addPortsCalls++
if err := fw.Network.AddPorts(ctx, pbs); err != nil {
return err
}
return fw.addPortsErr
}
func (fw *portMappingFirewaller) DelPorts(ctx context.Context, pbs []types.PortBinding) error {
if err := fw.Network.DelPorts(ctx, pbs); err != nil {
return err
}
return fw.delPortsErr
}
// Type for tracking calls to StartProxy.
type proxyCall struct{ proto, host, container string }