Merge pull request #54011 from akasakariko/fix/53466-batch-port-firewall
bridge: batch firewall updates for published ports
This commit is contained in:
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 }
|
||||
|
||||
|
||||
Reference in new issue
Block a user