From 3d3e570eefcca2b8f3cac029a6da7a6a174a853a Mon Sep 17 00:00:00 2001 From: chen21019 Date: Mon, 14 Sep 2026 20:41:12 +0800 Subject: [PATCH] Harden CNI provider and host-port convergence --- COMPATIBILITY.md | 28 ++++ README.md | 44 ++++-- binexec/watcher.go | 266 +++++++++++++++++++++++++++++------- binexec/watcher_test.go | 166 +++++++++++++++++++++- hostports/readiness_test.go | 161 ++++++++++++++++++++++ hostports/watcher.go | 129 ++++++++++++++++- 6 files changed, 729 insertions(+), 65 deletions(-) diff --git a/COMPATIBILITY.md b/COMPATIBILITY.md index 39509ec..fc2d4c7 100644 --- a/COMPATIBILITY.md +++ b/COMPATIBILITY.md @@ -24,6 +24,34 @@ These strings are compatibility identifiers, not product names, image names, pub The compatibility CA path is read only when the PastureStack-native path is absent. Catalog templates must use PastureStack image names and reviewed numeric tags, with release digests checked separately, even while retained labels are required by the control-plane wire contract. +## CNI provider handoff and Metadata convergence + +Network drivers own their CNI executable and data-plane configuration. Network +Plugin Manager only selects the local provider described by Metadata and +installs the host-side handoff wrapper; it must not copy implementation logic +between IPsec, VXLAN, per-host-subnet, or flat Layer 2 plugins. When providers +overlap during a rolling upgrade, the highest numeric +`org.opencontainers.image.version` wins and the immutable container ID is the +deterministic tie-breaker. Unversioned providers remain compatible but cannot +displace a valid newer numeric version. Unsafe binary names fail before a +wrapper is written. + +Each wrapper is bound to the exact container that was inspected during +reconciliation. Invocation inspects only that immutable ID, never a fresh +same-service container listing, and prefers the selected provider's private +`/opt/cni/bin` bundle with `CNI_PATH` scoped to the same directory. The shared +binary path remains only as a compatibility fallback for older providers. +Wrappers are atomically installed as regular mode-0700 files; content, type, +and permission drift causes the selected wrappers to be restored. + +For a host-port container whose Metadata primary IP has not converged, the +manager may read only that running container's network namespace. The Docker +PID must remain identical across the read, the network must already expose a +validated managed IPv4 subnet, and exactly one non-loopback address may match +that subnet. Otherwise reconciliation fails closed and retries without +replacing the previously applied firewall rules. This fallback does not infer +an address from an arbitrary interface, another container, or a service label. + ## Firewall backend migration The `iptables-legacy` frontend remains an explicit compatibility mode diff --git a/README.md b/README.md index aca01bc..01fae8b 100644 --- a/README.md +++ b/README.md @@ -68,7 +68,7 @@ It restores target-scoped authorization for every packet in an owned DNAT flow, including later UDP datagrams, without accepting unrelated Docker traffic. -The current release is `v0.8.20`. Managed bridge subnets can initiate +The current release is `v0.8.21`. Managed bridge subnets can initiate outbound traffic and receive established or related replies. Shared overlay subnets used by IPsec and VXLAN can also receive new connections from the same validated subnet through the exact managed bridge. Existing templates @@ -93,20 +93,44 @@ image identity from the release's checksum-covered [`published.txt`](https://github.com/PastureStack/network-plugin-manager/releases/latest/download/published.txt) rather than copying an older release digest. +`v0.8.21` also closes two control-plane convergence gaps without moving +responsibility between plugins. If Metadata temporarily omits the primary IP +of a running container that publishes a host port, the manager reads that +exact container's network namespace and accepts an address only when exactly +one IPv4 address belongs to the already validated managed bridge subnet. It +inspects the Docker PID before and after the namespace read; a stopped or +replaced process, an absent subnet, or zero/multiple matching addresses fails +closed and leaves the previously working rule set in place for retry. + +When more than one local network driver provides the same CNI executable, the +manager deterministically selects the highest numeric OCI image version, with +the immutable container ID as a stable tie-breaker. The host-side wrapper is +bound to that exact inspected container ID and executes the selected driver's +private `/opt/cni/bin` bundle with a private `CNI_PATH`; it does not list and +reselect a same-labelled container at invocation time. Binary names are +strictly validated, wrappers are installed atomically as regular mode-0700 +files, and content, type, or permission drift is repaired. Older drivers that +do not yet contain a private bundle retain the existing shared-binary fallback. +The driver still owns its CNI data plane; Network Plugin Manager continues to +own only host NAT, forwarding, and host-port reconciliation. + The current preflight inspects already loaded legacy tables using an independent iptables-legacy executable. Active old platform or Docker hooks in the other frontend block startup; an unhooked chain declaration alone does not select or block a backend. The manager never migrates host rules or switches Docker's selected backend automatically. -A bounded two-host upgrade gate passed after operator-controlled cleanup of -old platform hooks: both hosts retained nft Docker hooks, the new manager was -healthy, Metadata and IPsec services ran, and metadata network namespaces -resolved DNS and reached the management ping. A service port on the second -host returned HTTP 200. This component does not migrate or remove old rules -automatically. Verify the official image digest and perform controlled host -migration for each deployment. The complete Ubuntu 26.04 native-nft -control-plane gate remains pending Catalog/Server integration. +A bounded two-host Ubuntu 26.04 / Docker 29 gate exercised the release +candidate through a managed-service upgrade. Docker native nftables on one +host and `iptables-nft` on the other passed IPsec, VXLAN, per-host-subnet, and +flat Layer 2 cross-host traffic, Metadata, DNS, platform egress, and host-port +checks. The second host passed the same checks after a Docker restart and +after an explicit switch to `iptables-legacy`, then was restored to its +original `iptables-nft` frontend. Injected CNI-wrapper drift was restored to +the same exact selected provider on both hosts. This component does not +migrate or remove old rules automatically. Verify the official image digest +and perform controlled host migration for each deployment; Catalog and Server +integration are separate release gates. The maintained image coordinate is: @@ -188,7 +212,7 @@ The Alpine 3.23 base image is digest-pinned. Direct runtime packages are exact-v make test make validate bash scripts/check-build-downloads -VERSION_OVERRIDE=v0.8.20 IMAGE_NAMESPACE=local/pasturestack make package +VERSION_OVERRIDE=v0.8.21 IMAGE_NAMESPACE=local/pasturestack make package ``` Pull requests and `main` run one non-publishing gate: tests, vet/format checks, govulncheck, a reproducible binary build, one runtime image build, and Trivy scans plus CycloneDX SBOMs for the source, binary, and image. All reported vulnerabilities and secrets fail the gate. Publishing remains a separate, explicitly authorized operation. diff --git a/binexec/watcher.go b/binexec/watcher.go index c850299..e061813 100644 --- a/binexec/watcher.go +++ b/binexec/watcher.go @@ -1,11 +1,13 @@ package binexec import ( + "bytes" "context" "fmt" "os" "path/filepath" "reflect" + "strconv" "strings" "sync" "time" @@ -19,10 +21,42 @@ import ( ) var ( - reapplyEvery = 5 * time.Minute - binDir = cniglue.CniPath[0] + reapplyEvery = 5 * time.Minute + wrapperDriftCheckEvery = 10 * time.Second + binDir = cniglue.CniPath[0] ) +const driverWrapperScript = `#!/bin/sh +set -eu +target=%s +binary_name=%s +socket=/var/run/docker.sock +api_prefix="" +if [ -n "${DOCKER_API_VERSION:-}" ]; then + case "${DOCKER_API_VERSION}" in + *[!0-9.]*|'') echo '{"code":100,"msg":"invalid Docker API version"}' >&2; exit 1 ;; + esac + api_prefix="/v${DOCKER_API_VERSION}" +fi +case "${target}" in + *[!0-9a-fA-F]*|'') echo '{"code":100,"msg":"invalid CNI driver container id"}' >&2; exit 1 ;; +esac +pid="$(curl -fsS --max-time 10 --unix-socket "${socket}" \ + "http://localhost${api_prefix}/containers/${target}/json" | jq -r '.State.Pid // 0' || true)" +case "${pid}" in + ''|0|*[!0-9]*) + echo "{\"code\":100,\"msg\":\"selected cni driver container not running: ${target}\"}" >&2 + exit 1 + ;; +esac +private_binary="/opt/cni/bin/${binary_name}" +if /usr/bin/nsenter -m -u -i -n -p -t "${pid}" -- test -x "${private_binary}"; then + exec /usr/bin/nsenter -m -u -i -n -p -t "${pid}" -- \ + /usr/bin/env CNI_PATH=/opt/cni/bin "${private_binary}" "$@" +fi +exec /usr/bin/nsenter -m -u -i -n -p -t "${pid}" -- "$0" "$@" +` + func Watch(c metadata.Client, dc *client.Client) *Watcher { w := &Watcher{ c: c, @@ -31,6 +65,7 @@ func Watch(c metadata.Client, dc *client.Client) *Watcher { } w.onChange("") go c.OnChange(5, w.onChangeNoError) + go w.watchWrapperDrift() return w } @@ -48,6 +83,21 @@ func (w *Watcher) onChangeNoError(version string) { } } +func (w *Watcher) watchWrapperDrift() { + ticker := time.NewTicker(wrapperDriftCheckEvery) + defer ticker.Stop() + for range ticker.C { + w.Lock() + if len(w.applied) != 0 && !w.wrapperFilesMatch(w.applied) { + logrus.Warn("CNI driver wrapper drift detected; restoring selected providers") + if err := w.apply(w.applied); err != nil { + logrus.Errorf("Failed to restore CNI driver wrappers: %v", err) + } + } + w.Unlock() + } +} + func (w *Watcher) Handle(event *events.Message) error { w.Lock() @@ -112,59 +162,57 @@ func (w *Watcher) onChange(version string) error { if container.ExternalId != "" && container.HostUUID == hostUUID && hasDriverLabel(container) { binName := getBinaryName(container) if binName != "" { - binaries[binName] = container.ExternalId + if !validBinaryName(binName) { + return fmt.Errorf("invalid CNI driver binary name %q", binName) + } + current, exists := binaries[binName] + if !exists || w.preferBinaryProvider(current, container.ExternalId) { + binaries[binName] = container.ExternalId + } } } } } - if time.Now().Sub(w.lastApplied) > reapplyEvery || !reflect.DeepEqual(binaries, w.applied) { + needsApply := time.Now().Sub(w.lastApplied) > reapplyEvery || !reflect.DeepEqual(binaries, w.applied) + if !needsApply && !w.wrapperFilesMatch(binaries) { + logrus.Warn("CNI driver wrapper drift detected; restoring selected providers") + needsApply = true + } + if needsApply { return w.apply(binaries) } return nil } +func (w *Watcher) wrapperFilesMatch(binaries map[string]string) bool { + for name, target := range binaries { + expected := renderDriverWrapper(target, name) + if !wrapperFileMatches(filepath.Join(binDir, name), expected) { + return false + } + } + return true +} + +func wrapperFileMatches(path string, expected []byte) bool { + info, err := os.Lstat(path) + if err != nil || !info.Mode().IsRegular() || info.Mode().Perm() != 0700 { + return false + } + actual, err := os.ReadFile(path) + return err == nil && bytes.Equal(actual, expected) +} + func (w *Watcher) apply(binaries map[string]string) error { if !reflect.DeepEqual(binaries, w.applied) { logrus.Infof("Setting up binaries for: %v", binaries) } - const script = `#!/bin/sh -set -eu -target=%s -service_label=%s -socket=/var/run/docker.sock -api_prefix="" -if [ -n "${DOCKER_API_VERSION:-}" ]; then - case "${DOCKER_API_VERSION}" in - *[!0-9.]*|'') echo '{"code":100,"msg":"invalid Docker API version"}' >&2; exit 1 ;; - esac - api_prefix="/v${DOCKER_API_VERSION}" -fi -cid="" -if [ -n "${service_label}" ]; then - filters="$(jq -cn --arg label "io.rancher.stack_service.name=${service_label}" '{label:[$label]}')" - cid="$(curl -fsS --max-time 10 --unix-socket "${socket}" --get \ - --data-urlencode "filters=${filters}" \ - "http://localhost${api_prefix}/containers/json" | jq -r '.[0].Id // empty')" -fi -if [ -z "${cid}" ]; then - cid="${target}" -fi -case "${cid}" in - *[!0-9a-fA-F]*|'') echo '{"code":100,"msg":"invalid CNI driver container id"}' >&2; exit 1 ;; -esac -pid="$(curl -fsS --max-time 10 --unix-socket "${socket}" \ - "http://localhost${api_prefix}/containers/${cid}/json" | jq -r '.State.Pid // 0' || true)" -if [ -z "${pid}" ] || [ "${pid}" = "0" ]; then - echo "{\"code\":100,\"msg\":\"cni driver container not running: ${service_label:-${target}}\"}" >&2 - exit 1 -fi -exec /usr/bin/nsenter -m -u -i -n -p -t "${pid}" -- $0 "$@" -` - - os.MkdirAll(binDir, 0700) + if err := os.MkdirAll(binDir, 0700); err != nil { + return fmt.Errorf("create CNI wrapper directory: %w", err) + } var lastErr error for name, target := range binaries { @@ -180,23 +228,13 @@ exec /usr/bin/nsenter -m -u -i -n -p -t "${pid}" -- $0 "$@" break } - serviceLabel := "" - if container.Config != nil && container.Config.Labels != nil { - serviceLabel = container.Config.Labels["io.rancher.stack_service.name"] - } - - ptmp := filepath.Join(binDir, name+".tmp") p := filepath.Join(binDir, name) - content := []byte(fmt.Sprintf(script, shellQuote(target), shellQuote(serviceLabel))) + content := renderDriverWrapper(target, name) logrus.Debugf("Writing %s:\n%s", p, content) - if err := os.WriteFile(ptmp, content, 0700); err != nil { + if err := writeWrapperAtomic(p, content); err != nil { lastErr = err break } - - if err := os.Rename(ptmp, p); err != nil { - lastErr = err - } } if lastErr == nil { @@ -207,6 +245,132 @@ exec /usr/bin/nsenter -m -u -i -n -p -t "${pid}" -- $0 "$@" return lastErr } +func (w *Watcher) preferBinaryProvider(current, candidate string) bool { + currentVersion := w.containerImageVersion(current) + candidateVersion := w.containerImageVersion(candidate) + return preferProvider(current, currentVersion, candidate, candidateVersion) +} + +func preferProvider(current, currentVersion, candidate, candidateVersion string) bool { + comparison, comparable := compareNumericVersions(candidateVersion, currentVersion) + if comparable && comparison != 0 { + return comparison > 0 + } + _, currentNumeric := numericVersionParts(currentVersion) + _, candidateNumeric := numericVersionParts(candidateVersion) + if currentNumeric != candidateNumeric { + return candidateNumeric + } + return candidate < current +} + +func (w *Watcher) containerImageVersion(containerID string) string { + if w.dc == nil { + return "" + } + result, err := w.dc.ContainerInspect(context.Background(), containerID, client.ContainerInspectOptions{}) + if err != nil || result.Container.Config == nil || result.Container.Config.Labels == nil { + return "" + } + return strings.TrimSpace(result.Container.Config.Labels["org.opencontainers.image.version"]) +} + +func writeWrapperAtomic(path string, content []byte) (err error) { + temporary, err := os.CreateTemp(filepath.Dir(path), "."+filepath.Base(path)+".tmp-*") + if err != nil { + return fmt.Errorf("create private CNI wrapper temporary file: %w", err) + } + temporaryPath := temporary.Name() + committed := false + defer func() { + if !committed { + _ = temporary.Close() + _ = os.Remove(temporaryPath) + } + }() + if _, err := temporary.Write(content); err != nil { + return fmt.Errorf("write private CNI wrapper temporary file: %w", err) + } + if err := temporary.Chmod(0700); err != nil { + return fmt.Errorf("set private CNI wrapper permissions: %w", err) + } + if err := temporary.Sync(); err != nil { + return fmt.Errorf("sync private CNI wrapper temporary file: %w", err) + } + if err := temporary.Close(); err != nil { + return fmt.Errorf("close private CNI wrapper temporary file: %w", err) + } + if err := os.Rename(temporaryPath, path); err != nil { + return fmt.Errorf("install private CNI wrapper: %w", err) + } + committed = true + return nil +} + +func renderDriverWrapper(target, name string) []byte { + return []byte(fmt.Sprintf(driverWrapperScript, shellQuote(target), shellQuote(name))) +} + +func compareNumericVersions(left, right string) (int, bool) { + leftParts, leftOK := numericVersionParts(left) + rightParts, rightOK := numericVersionParts(right) + if !leftOK || !rightOK { + return 0, false + } + for index := 0; index < len(leftParts) || index < len(rightParts); index++ { + leftValue, rightValue := 0, 0 + if index < len(leftParts) { + leftValue = leftParts[index] + } + if index < len(rightParts) { + rightValue = rightParts[index] + } + if leftValue < rightValue { + return -1, true + } + if leftValue > rightValue { + return 1, true + } + } + return 0, true +} + +func numericVersionParts(value string) ([]int, bool) { + value = strings.TrimPrefix(strings.TrimSpace(value), "v") + if value == "" { + return nil, false + } + parts := strings.Split(value, ".") + parsed := make([]int, len(parts)) + for index, part := range parts { + if part == "" { + return nil, false + } + value, err := strconv.Atoi(part) + if err != nil || value < 0 { + return nil, false + } + parsed[index] = value + } + return parsed, true +} + +func validBinaryName(value string) bool { + if value == "" || value == "." || value == ".." || strings.ContainsAny(value, `/\\`) { + return false + } + for _, character := range value { + if (character >= 'a' && character <= 'z') || + (character >= 'A' && character <= 'Z') || + (character >= '0' && character <= '9') || + character == '.' || character == '_' || character == '-' { + continue + } + return false + } + return true +} + func shellQuote(value string) string { return "'" + strings.ReplaceAll(value, "'", "'\"'\"'") + "'" } diff --git a/binexec/watcher_test.go b/binexec/watcher_test.go index 91ae5a7..585e1ff 100644 --- a/binexec/watcher_test.go +++ b/binexec/watcher_test.go @@ -1,6 +1,12 @@ package binexec -import "testing" +import ( + "os" + "path/filepath" + "runtime" + "strings" + "testing" +) func TestShellQuoteKeepsMetadataInsideOneLiteral(t *testing.T) { input := "service'$(touch /tmp/should-not-run)" @@ -9,3 +15,161 @@ func TestShellQuoteKeepsMetadataInsideOneLiteral(t *testing.T) { t.Fatalf("shellQuote() = %q, want %q", got, want) } } + +func TestNumericVersionComparison(t *testing.T) { + tests := []struct { + left, right string + want int + ok bool + }{ + {"0.14.37", "0.14.34", 1, true}, + {"v0.14.34", "0.14.37", -1, true}, + {"0.14.37", "0.14.37.0", 0, true}, + {"dev", "0.14.37", 0, false}, + } + for _, test := range tests { + got, ok := compareNumericVersions(test.left, test.right) + if got != test.want || ok != test.ok { + t.Fatalf("compareNumericVersions(%q, %q) = %d, %v", test.left, test.right, got, ok) + } + } +} + +func TestProviderPreferenceKeepsNumericVersionsAheadOfDevelopmentLabels(t *testing.T) { + tests := []struct { + name string + currentID, currentVersion string + candidateID, candidateVersion string + want bool + }{ + {"newer numeric version", "bbbb", "0.14.34", "cccc", "0.14.37", true}, + {"older numeric version", "bbbb", "0.14.37", "aaaa", "0.14.34", false}, + {"numeric replaces development label", "bbbb", "dev", "cccc", "0.14.37", true}, + {"development label cannot replace numeric", "bbbb", "0.14.37", "aaaa", "dev", false}, + {"equal version uses lower immutable id", "bbbb", "0.14.37", "aaaa", "0.14.37", true}, + {"equal version retains lower immutable id", "aaaa", "0.14.37", "bbbb", "0.14.37", false}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got := preferProvider(test.currentID, test.currentVersion, test.candidateID, test.candidateVersion) + if got != test.want { + t.Fatalf("preferProvider(%q, %q, %q, %q) = %v, want %v", test.currentID, test.currentVersion, test.candidateID, test.candidateVersion, got, test.want) + } + }) + } +} + +func TestBinaryNameValidation(t *testing.T) { + for _, valid := range []string{"pasture-bridge", "driver_1.2"} { + if !validBinaryName(valid) { + t.Fatalf("valid binary name rejected: %q", valid) + } + } + for _, invalid := range []string{"", ".", "..", "../driver", `dir\\driver`, "driver name", "$(id)"} { + if validBinaryName(invalid) { + t.Fatalf("invalid binary name accepted: %q", invalid) + } + } +} + +func TestDriverWrapperUsesSelectedContainersPrivateBundle(t *testing.T) { + wrapper := string(renderDriverWrapper("0123456789abcdef", "pasture-bridge")) + for _, required := range []string{ + "containers/${target}/json", + "private_binary=\"/opt/cni/bin/${binary_name}\"", + "CNI_PATH=/opt/cni/bin", + "\"${private_binary}\" \"$@\"", + "-- \"$0\" \"$@\"", + } { + if !strings.Contains(wrapper, required) { + t.Fatalf("driver wrapper is missing %q:\n%s", required, wrapper) + } + } + for _, forbidden := range []string{"/containers/json", "service_label", "filters="} { + if strings.Contains(wrapper, forbidden) { + t.Fatalf("driver wrapper can reselect an unverified provider through %q:\n%s", forbidden, wrapper) + } + } +} + +func TestWrapperFileMatchesDetectsOverwrite(t *testing.T) { + path := filepath.Join(t.TempDir(), "pasture-bridge") + expected := renderDriverWrapper("0123456789abcdef", "pasture-bridge") + if err := os.WriteFile(path, expected, 0700); err != nil { + t.Fatal(err) + } + if !wrapperFileMatches(path, expected) { + t.Fatal("matching wrapper was reported as drifted") + } + if err := os.WriteFile(path, []byte("stale wrapper\n"), 0700); err != nil { + t.Fatal(err) + } + if wrapperFileMatches(path, expected) { + t.Fatal("overwritten wrapper was reported as current") + } + if err := os.WriteFile(path, expected, 0600); err != nil { + t.Fatal(err) + } + if err := os.Chmod(path, 0600); err != nil { + t.Fatal(err) + } + if wrapperFileMatches(path, expected) { + t.Fatal("wrapper with non-executable permissions was reported as current") + } + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + target := filepath.Join(t.TempDir(), "wrapper-target") + if err := os.WriteFile(target, expected, 0700); err != nil { + t.Fatal(err) + } + if err := os.Symlink(target, path); err != nil { + if runtime.GOOS == "windows" { + t.Skipf("symlink creation is unavailable: %v", err) + } + t.Fatal(err) + } + if wrapperFileMatches(path, expected) { + t.Fatal("symlinked wrapper was reported as a regular managed file") + } +} + +func TestAtomicWrapperWriteDoesNotFollowPredictableTemporarySymlink(t *testing.T) { + directory := t.TempDir() + path := filepath.Join(directory, "pasture-bridge") + victim := filepath.Join(directory, "victim") + if err := os.WriteFile(victim, []byte("unchanged"), 0600); err != nil { + t.Fatal(err) + } + if err := os.Symlink(victim, path+".tmp"); err != nil { + if runtime.GOOS == "windows" { + t.Skipf("symlink creation is unavailable: %v", err) + } + t.Fatal(err) + } + content := []byte("#!/bin/sh\nexit 0\n") + if err := writeWrapperAtomic(path, content); err != nil { + t.Fatal(err) + } + gotVictim, err := os.ReadFile(victim) + if err != nil { + t.Fatal(err) + } + if string(gotVictim) != "unchanged" { + t.Fatalf("predictable temporary symlink target was overwritten: %q", gotVictim) + } + gotWrapper, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if string(gotWrapper) != string(content) { + t.Fatalf("installed wrapper = %q, want %q", gotWrapper, content) + } + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm() != 0700 { + t.Fatalf("installed wrapper permissions = %o, want 700", info.Mode().Perm()) + } +} diff --git a/hostports/readiness_test.go b/hostports/readiness_test.go index 8ba4856..724e2f7 100644 --- a/hostports/readiness_test.go +++ b/hostports/readiness_test.go @@ -1,11 +1,15 @@ package hostports import ( + "net" + "net/netip" "strings" "testing" + "github.com/PastureStack/network-plugin-manager/internal/firewall" "github.com/PastureStack/network-plugin-manager/internal/metadata" "github.com/moby/moby/client" + "github.com/vishvananda/netlink" ) type badPortMetadata struct{ metadata.Client } @@ -14,6 +18,163 @@ func (badPortMetadata) GetNetworks() ([]metadata.Network, error) { return []metadata.Network{{UUID: "net-1", HostPorts: true}}, nil } +func netlinkAddress(cidr string) netlink.Addr { + ip, network, err := net.ParseCIDR(cidr) + if err != nil { + panic(err) + } + return netlink.Addr{IPNet: &net.IPNet{IP: ip, Mask: network.Mask}} +} + +func TestSelectContainerIPv4IsScopedAndUnambiguous(t *testing.T) { + prefix := netip.MustParsePrefix("10.50.1.0/24") + addresses := []netlink.Addr{ + netlinkAddress("127.0.0.1/8"), + netlinkAddress("192.0.2.9/24"), + netlinkAddress("10.50.1.2/24"), + netlinkAddress("10.50.1.2/32"), + } + got, err := selectContainerIPv4(addresses, prefix) + if err != nil || got != "10.50.1.2" { + t.Fatalf("scoped address = %q, %v", got, err) + } + addresses = append(addresses, netlinkAddress("10.50.1.3/24")) + if _, err := selectContainerIPv4(addresses, prefix); err == nil || !strings.Contains(err.Error(), "found 2") { + t.Fatalf("ambiguous managed addresses were accepted: %v", err) + } +} + +func TestResolveContainerIPv4RevalidatesContainerPID(t *testing.T) { + pidCalls := 0 + w := &watcher{ + containerPID: func(string) (int, error) { + pidCalls++ + if pidCalls == 1 { + return 101, nil + } + return 202, nil + }, + namespaceIPv4: func(pid int) ([]netlink.Addr, error) { + if pid != 101 { + t.Fatalf("namespace reader PID = %d, want 101", pid) + } + return []netlink.Addr{netlinkAddress("10.50.1.2/24")}, nil + }, + } + if _, err := w.resolveContainerIPv4("container-1", "10.50.1.0/24"); err == nil || !strings.Contains(err.Error(), "PID changed") { + t.Fatalf("PID lifecycle race was not rejected: %v", err) + } +} + +func TestResolveContainerIPv4AcceptsStableContainerPID(t *testing.T) { + pidCalls := 0 + w := &watcher{ + containerPID: func(string) (int, error) { + pidCalls++ + return 101, nil + }, + namespaceIPv4: func(pid int) ([]netlink.Addr, error) { + return []netlink.Addr{netlinkAddress("10.50.1.2/24")}, nil + }, + } + got, err := w.resolveContainerIPv4("container-1", "10.50.1.0/24") + if err != nil || got != "10.50.1.2" { + t.Fatalf("stable namespace address = %q, %v", got, err) + } + if pidCalls != 2 { + t.Fatalf("container PID inspected %d times, want 2", pidCalls) + } +} + +type missingPrimaryIPMetadata struct{ metadata.Client } + +func (missingPrimaryIPMetadata) GetNetworks() ([]metadata.Network, error) { + return []metadata.Network{{ + UUID: "net-1", HostPorts: true, + Metadata: map[string]interface{}{"cniConfig": map[string]interface{}{ + "10-per-host.conf": map[string]interface{}{ + "type": "pasture-bridge", "bridge": "cattle0", "bridgeSubnet": "10.50.1.0/24", + }, + }}, + }}, nil +} + +func (missingPrimaryIPMetadata) GetContainers() ([]metadata.Container, error) { + return []metadata.Container{{ + ExternalId: "container-1", HostUUID: "host-1", NetworkUUID: "net-1", + Name: "probe", State: "running", Ports: []string{"0.0.0.0:18084:8080/tcp"}, + }}, nil +} + +func TestMissingMetadataIPUsesManagedNamespaceAddress(t *testing.T) { + var gotContainerID, gotSubnet string + w := &watcher{ + c: missingPrimaryIPMetadata{}, + localHost: func(metadata.Client, *client.Client) (metadata.Host, error) { + return metadata.Host{UUID: "host-1", AgentIP: "192.0.2.10"}, nil + }, + containerIPv4: func(containerID, subnet string) (string, error) { + gotContainerID, gotSubnet = containerID, subnet + return "10.50.1.2", nil + }, + applied: ruleSet{ + Ports: map[string]PortRule{}, ForwardSubnets: map[string]string{}, ForwardBridges: map[string]string{}, RouteLocalnetBridges: map[string]bool{}, + }, + backend: firewall.Backend{Mode: firewall.NFTables, Command: "nft"}, + output: func(_ ...string) ([]byte, error) { + return nil, nil + }, + getRouteLocalnet: func(string) (bool, error) { return false, nil }, + setRouteLocalnet: func(string, bool) error { return nil }, + routeLocalnetOriginal: map[string]bool{}, + routeLocalnetStatePath: t.TempDir() + "/route-localnet.json", + restoreRules: func(_ string, _ []string, data []byte) error { + if !strings.Contains(string(data), "10.50.1.2:8080") { + t.Fatalf("resolved CNI address was not used in host-port rules: %s", data) + } + return nil + }, + } + if err := w.onChange("test"); err != nil { + t.Fatal(err) + } + if gotContainerID != "container-1" || gotSubnet != "10.50.1.0/24" { + t.Fatalf("resolver arguments = %q, %q", gotContainerID, gotSubnet) + } +} + +func TestMissingMetadataIPWithoutManagedSubnetFailsClosed(t *testing.T) { + clientWithoutSubnet := badPortMetadata{} + w := &watcher{ + c: clientWithoutSubnet, + localHost: func(metadata.Client, *client.Client) (metadata.Host, error) { + return metadata.Host{UUID: "host-1", AgentIP: "192.0.2.10"}, nil + }, + applied: ruleSet{Ports: map[string]PortRule{}, ForwardSubnets: map[string]string{}, ForwardBridges: map[string]string{}, RouteLocalnetBridges: map[string]bool{}}, + } + containers, _ := clientWithoutSubnet.GetContainers() + containers[0].PrimaryIp = "" + containers[0].Ports = []string{"0.0.0.0:18084:8080/tcp"} + w.c = &singleContainerMetadata{network: metadata.Network{UUID: "net-1", HostPorts: true}, container: containers[0]} + if err := w.onChange("test"); err == nil || !strings.Contains(err.Error(), "without a metadata IP or managed subnet") { + t.Fatalf("missing address boundary was not rejected: %v", err) + } +} + +type singleContainerMetadata struct { + metadata.Client + network metadata.Network + container metadata.Container +} + +func (m *singleContainerMetadata) GetNetworks() ([]metadata.Network, error) { + return []metadata.Network{m.network}, nil +} + +func (m *singleContainerMetadata) GetContainers() ([]metadata.Container, error) { + return []metadata.Container{m.container}, nil +} + func (badPortMetadata) GetContainers() ([]metadata.Container, error) { return []metadata.Container{{ ExternalId: "container-1", HostUUID: "host-1", NetworkUUID: "net-1", diff --git a/hostports/watcher.go b/hostports/watcher.go index 8b361c6..82491e3 100644 --- a/hostports/watcher.go +++ b/hostports/watcher.go @@ -2,6 +2,7 @@ package hostports import ( "bytes" + "context" "fmt" "net/netip" "os" @@ -19,6 +20,8 @@ import ( "github.com/PastureStack/network-plugin-manager/internal/metadata" "github.com/moby/moby/client" "github.com/sirupsen/logrus" + "github.com/vishvananda/netlink" + "github.com/vishvananda/netns" ) var ( @@ -89,6 +92,9 @@ type watcher struct { routeLocalnetStatePath string report func(error) localHost func(metadata.Client, *client.Client) (metadata.Host, error) + containerIPv4 func(containerID, subnet string) (string, error) + containerPID func(containerID string) (int, error) + namespaceIPv4 func(pid int) ([]netlink.Addr, error) } type ruleSet struct { @@ -411,8 +417,10 @@ func (w *watcher) onChangeLocked(version string, force bool) error { } if container.HostUUID != host.UUID || - !(network.HostPorts || (container.System && container.Labels[hostPortsLabel] == "true")) || - container.PrimaryIp == "" { + !(network.HostPorts || (container.System && container.Labels[hostPortsLabel] == "true")) { + continue + } + if len(container.Ports) == 0 { continue } @@ -421,9 +429,24 @@ func (w *watcher) onChangeLocked(version string, force bool) error { return fmt.Errorf("network %s bridge configuration: %w", container.NetworkUUID, err) } bridge, masqueradeDNAT = bridgeConfig.Bridge, bridgeConfig.ExternalGateway + targetIP := container.PrimaryIp + if targetIP == "" { + subnet := newRules.ForwardSubnets[container.NetworkUUID] + if subnet == "" { + return fmt.Errorf("container %s (%s) publishes host ports without a metadata IP or managed subnet", container.Name, container.ExternalId) + } + resolveContainerIPv4 := w.containerIPv4 + if resolveContainerIPv4 == nil { + resolveContainerIPv4 = w.resolveContainerIPv4 + } + targetIP, err = resolveContainerIPv4(container.ExternalId, subnet) + if err != nil { + return fmt.Errorf("resolve host-port address for container %s (%s): %w", container.Name, container.ExternalId, err) + } + } for _, port := range container.Ports { - rule, ok := parsePortRule(bridge, host.AgentIP, container.PrimaryIp, port) + rule, ok := parsePortRule(bridge, host.AgentIP, targetIP, port) if !ok { return fmt.Errorf("invalid host port definition for container %s (%s): %q", container.Name, container.ExternalId, port) } @@ -449,6 +472,106 @@ func (w *watcher) onChangeLocked(version string, force bool) error { return nil } +func (w *watcher) resolveContainerIPv4(containerID, subnet string) (string, error) { + if containerID == "" { + return "", fmt.Errorf("Docker container ID is empty") + } + prefix, err := netip.ParsePrefix(subnet) + if err != nil || !prefix.Addr().Is4() { + return "", fmt.Errorf("managed subnet %q is not a valid IPv4 prefix", subnet) + } + prefix = prefix.Masked() + + inspectPID := w.containerPID + if inspectPID == nil { + inspectPID = w.inspectRunningContainerPID + } + pid, err := inspectPID(containerID) + if err != nil { + return "", err + } + readAddresses := w.namespaceIPv4 + if readAddresses == nil { + readAddresses = readContainerNamespaceIPv4 + } + addresses, err := readAddresses(pid) + if err != nil { + return "", err + } + confirmedPID, err := inspectPID(containerID) + if err != nil { + return "", fmt.Errorf("revalidate Docker container after network namespace read: %w", err) + } + if confirmedPID != pid { + return "", fmt.Errorf("Docker container PID changed during network namespace read: %d to %d", pid, confirmedPID) + } + return selectContainerIPv4(addresses, prefix) +} + +func (w *watcher) inspectRunningContainerPID(containerID string) (int, error) { + if w.dc == nil { + return 0, fmt.Errorf("Docker client is unavailable") + } + inspectResult, err := w.dc.ContainerInspect(context.Background(), containerID, client.ContainerInspectOptions{}) + if err != nil { + return 0, fmt.Errorf("inspect Docker container: %w", err) + } + state := inspectResult.Container.State + if state == nil || !state.Running || state.Pid <= 0 { + return 0, fmt.Errorf("Docker container is not running with a network namespace") + } + return state.Pid, nil +} + +func readContainerNamespaceIPv4(pid int) ([]netlink.Addr, error) { + ns, err := netns.GetFromPid(pid) + if err != nil { + return nil, fmt.Errorf("open network namespace: %w", err) + } + defer ns.Close() + handle, err := netlink.NewHandleAt(ns) + if err != nil { + return nil, fmt.Errorf("open netlink handle: %w", err) + } + defer handle.Delete() + addresses, err := handle.AddrList(nil, netlink.FAMILY_V4) + if err != nil { + return nil, fmt.Errorf("list network addresses: %w", err) + } + return addresses, nil +} + +func selectContainerIPv4(addresses []netlink.Addr, prefix netip.Prefix) (string, error) { + candidates := make([]netip.Addr, 0, len(addresses)) + for _, address := range addresses { + if address.IPNet == nil { + continue + } + parsed, ok := netip.AddrFromSlice(address.IP) + if !ok { + continue + } + parsed = parsed.Unmap() + if !parsed.Is4() || parsed.IsLoopback() || !prefix.Contains(parsed) { + continue + } + duplicate := false + for _, candidate := range candidates { + if candidate == parsed { + duplicate = true + break + } + } + if !duplicate { + candidates = append(candidates, parsed) + } + } + if len(candidates) != 1 { + return "", fmt.Errorf("expected exactly one address in managed subnet %s, found %d", prefix, len(candidates)) + } + return candidates[0].String(), nil +} + func (w *watcher) apply(rules ruleSet) error { w.baseRuleMu.Lock() defer w.baseRuleMu.Unlock()