From 6ad3f9b18f33e641885b6d7c593fc2068f196821 Mon Sep 17 00:00:00 2001 From: Abigailliang Date: Mon, 26 Oct 2026 17:23:45 -0700 Subject: [PATCH 1/3] feat: add cloud platform and trust-store capabilities [1/5] Separate the unchanged cloud runtime and trust-store installation helpers from certificate retrieval and orchestration. Add direct capability coverage without activating provisioning. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: be499bac-1283-498e-b5b1-59d8c91047a4 --- aks-node-controller/README.md | 16 ++ .../cloud_certificate_store.go | 126 ++++++++++++ .../cloud_certificate_store_behavior_test.go | 103 ++++++++++ .../cloud_certificate_store_test.go | 18 ++ aks-node-controller/cloud_runtime.go | 183 ++++++++++++++++++ .../cloud_runtime_events_test.go | 27 +++ aks-node-controller/cloud_runtime_test.go | 68 +++++++ .../cloud_test_helpers_test.go | 59 ++++++ aks-node-controller/embeddednodecustomdata.go | 3 +- 9 files changed, 602 insertions(+), 1 deletion(-) create mode 100644 aks-node-controller/cloud_certificate_store.go create mode 100644 aks-node-controller/cloud_certificate_store_behavior_test.go create mode 100644 aks-node-controller/cloud_certificate_store_test.go create mode 100644 aks-node-controller/cloud_runtime.go create mode 100644 aks-node-controller/cloud_runtime_events_test.go create mode 100644 aks-node-controller/cloud_runtime_test.go create mode 100644 aks-node-controller/cloud_test_helpers_test.go diff --git a/aks-node-controller/README.md b/aks-node-controller/README.md index 0539665d43b..89ea6e29688 100644 --- a/aks-node-controller/README.md +++ b/aks-node-controller/README.md @@ -54,6 +54,22 @@ the provisioning config in CustomData; the automatic size fallback uses the same config-only payload. Leaving the new flag false preserves legacy, Phase 2, and Phase 2.5 behavior. +### Azure cloud trust-store capabilities + +The shared cloud runtime provides platform detection, bounded HTTP and command +execution, public-file writes, and guest-agent events. The trust-store helpers +write downloaded contents without PEM/X.509 filtering, update the platform's +existing CA tools, and preserve Ubuntu's OpenSSL bundle/symlink behavior. + +Cache cleanup removes flat staging files, including hidden files, without +recursively deleting directories or removing installed system-trust certificates. +`AKS_CLOUD_LOG_CERTIFICATES=true` enables public-certificate-only diagnostics; +parsing for logging does not gate installation. + +These are internal capabilities only. No certificate-retrieval flow, standalone +refresh command, refresh schedule, or native provisioning hook is enabled here. +Existing provisioning still uses `init-aks-cloud.sh`. + ### Validating and normalizing AKSNodeConfig Before serializing a typed config, producers such as RP should call: diff --git a/aks-node-controller/cloud_certificate_store.go b/aks-node-controller/cloud_certificate_store.go new file mode 100644 index 00000000000..9c559bb1e53 --- /dev/null +++ b/aks-node-controller/cloud_certificate_store.go @@ -0,0 +1,126 @@ +package main + +import ( + "bytes" + "context" + "crypto/x509" + "encoding/pem" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "sort" + "strings" +) + +func (c *cloudInitializer) clearCertificateCache() { + directory := c.path("/root/AzureCACertificates") + if err := os.MkdirAll(directory, 0o755); err != nil { + slog.Warn("could not create Azure cloud certificate cache", "error", err) + return + } + entries, err := os.ReadDir(directory) + if err != nil { + slog.Warn("could not read Azure cloud certificate cache", "error", err) + return + } + // Certificate downloads are written directly here as files, without subdirectories. + for _, entry := range entries { + if err := os.Remove(filepath.Join(directory, entry.Name())); err != nil { + slog.Warn("could not remove cached Azure cloud certificate", "name", entry.Name(), "error", err) + } + } +} + +func publicCloudCertificates(data []byte) ([]byte, error) { + var result bytes.Buffer + for { + block, rest := pem.Decode(data) + if block == nil { + break + } + data = rest + if block.Type != "CERTIFICATE" { + continue + } + if _, err := x509.ParseCertificate(block.Bytes); err != nil { + return nil, fmt.Errorf("invalid CA certificate: %w", err) + } + if err := pem.Encode(&result, &pem.Block{Type: "CERTIFICATE", Bytes: block.Bytes}); err != nil { + return nil, err + } + } + if result.Len() == 0 { + return nil, errors.New("response contains no PEM certificates") + } + return result.Bytes(), nil +} + +func (c *cloudInitializer) installCertificates(ctx context.Context, platform cloudPlatform, certificates map[string][]byte) error { + destination, update := "/usr/local/share/ca-certificates", "update-ca-certificates" + switch platform.ID { + case osReleaseIDAzureLinux, osReleaseIDMariner, osReleaseIDAzureContainerLinux: + destination, update = "/etc/pki/ca-trust/source/anchors", "update-ca-trust" + case osReleaseIDFlatcar: + destination = "/etc/ssl/certs" + } + names := make([]string, 0, len(certificates)) + for name := range certificates { + names = append(names, name) + } + sort.Strings(names) + installed := 0 + for _, name := range names { + data := certificates[name] + if !strings.HasSuffix(name, ".crt") { + continue + } + targetName := name + if platform.ID == osReleaseIDFlatcar { + targetName = strings.TrimSuffix(name, ".crt") + ".pem" + } + if err := c.writeFile(filepath.Join(destination, targetName), data); err != nil { + return err + } + installed++ + if os.Getenv("AKS_CLOUD_LOG_CERTIFICATES") == "true" { + public, err := publicCloudCertificates(data) + if err != nil { + slog.Warn("could not decode public Azure cloud CA certificates for logging", "name", name, "error", err) + continue + } + slog.Info("Azure cloud public CA certificate", "name", name, "certificate", string(public)) + } + } + if installed == 0 { + return errors.New("no *.crt files in Azure cloud certificate cache to install") + } + if _, err := c.command(ctx, nil, update); err != nil { + return err + } + if platform.ID == osReleaseIDUbuntu { + return c.copyOpenSSLBundle() + } + return nil +} + +func (c *cloudInitializer) copyOpenSSLBundle() error { + source, target := c.path("/etc/ssl/certs/ca-certificates.crt"), c.path("/usr/lib/ssl/cert.pem") + sourceInfo, err := os.Stat(source) + if err != nil { + return err + } + targetInfo, err := os.Stat(target) + if err == nil && os.SameFile(sourceInfo, targetInfo) { + return nil + } + if err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + data, err := os.ReadFile(source) + if err != nil { + return err + } + return c.writeFile("/usr/lib/ssl/cert.pem", data) +} diff --git a/aks-node-controller/cloud_certificate_store_behavior_test.go b/aks-node-controller/cloud_certificate_store_behavior_test.go new file mode 100644 index 00000000000..01e807de391 --- /dev/null +++ b/aks-node-controller/cloud_certificate_store_behavior_test.go @@ -0,0 +1,103 @@ +package main + +import ( + "encoding/pem" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestCloudCertificateStorePlatformInstallation(t *testing.T) { + for _, tc := range []struct{ id, target, command string }{ + {osReleaseIDUbuntu, "/usr/local/share/ca-certificates/root.crt", "update-ca-certificates"}, + {osReleaseIDMariner, "/etc/pki/ca-trust/source/anchors/root.crt", "update-ca-trust"}, + {osReleaseIDAzureLinux, "/etc/pki/ca-trust/source/anchors/root.crt", "update-ca-trust"}, + {osReleaseIDAzureContainerLinux, "/etc/pki/ca-trust/source/anchors/root.crt", "update-ca-trust"}, + {osReleaseIDFlatcar, "/etc/ssl/certs/root.pem", "update-ca-certificates"}, + } { + t.Run(tc.id, func(t *testing.T) { + c := newTestCloudInitializer(t, "ID="+tc.id+"\n") + platform, err := c.platform() + require.NoError(t, err) + var commands []string + c.run = func(cmd *exec.Cmd) error { + commands = append(commands, cmd.Args[0]) + return nil + } + payload := []byte("unparsed certificate payload\n") + require.NoError(t, c.installCertificates(t.Context(), platform, map[string][]byte{ + "root.crt": payload, + "skip.cer": []byte("ignored"), + })) + got, err := os.ReadFile(c.path(tc.target)) + require.NoError(t, err) + require.Equal(t, payload, got) + require.Equal(t, []string{tc.command}, commands) + _, err = os.Stat(c.path(filepath.Join(filepath.Dir(tc.target), "skip.cer"))) + require.ErrorIs(t, err, os.ErrNotExist) + if tc.id == osReleaseIDUbuntu { + bundle, err := os.ReadFile(c.path("/usr/lib/ssl/cert.pem")) + require.NoError(t, err) + require.Equal(t, "system-ca-bundle", string(bundle)) + } + }) + } +} + +func TestCloudCertificateStoreRejectsMissingCRTFiles(t *testing.T) { + c := newTestCloudInitializer(t, cloudTestUbuntu) + c.run = func(*exec.Cmd) error { + t.Fatal("empty certificate sets must not invoke trust-store tools") + return nil + } + err := c.installCertificates(t.Context(), cloudPlatform{ID: osReleaseIDUbuntu}, map[string][]byte{ + "root.cer": []byte("ignored"), + }) + require.ErrorContains(t, err, "no *.crt files") +} + +func TestCloudCertificateStoreClearsOnlyFlatStagingFiles(t *testing.T) { + c := newTestCloudInitializer(t, cloudTestUbuntu) + for _, name := range []string{"root.crt", ".hidden.crt", "nested/keep.crt"} { + require.NoError(t, c.writeFile("/root/AzureCACertificates/"+name, []byte(name))) + } + installed := "/usr/local/share/ca-certificates/installed.crt" + require.NoError(t, c.writeFile(installed, []byte("installed CA"))) + c.clearCertificateCache() + for _, name := range []string{"root.crt", ".hidden.crt"} { + _, err := os.Stat(c.path("/root/AzureCACertificates/" + name)) + require.ErrorIs(t, err, os.ErrNotExist) + } + for _, name := range []string{"/root/AzureCACertificates/nested/keep.crt", installed} { + _, err := os.Stat(c.path(name)) + require.NoError(t, err) + } +} + +func TestCloudCertificateStoreLoggingPreservesRawPayload(t *testing.T) { + t.Setenv("AKS_CLOUD_LOG_CERTIFICATES", "true") + logs := installLogCapturer(t) + c := newTestCloudInitializer(t, "ID=azurelinux\n") + payload := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: []byte("not-a-real-private-key")}) + payload = append(payload, cloudTestCertificate(t)...) + require.NoError(t, c.installCertificates(t.Context(), cloudPlatform{ID: osReleaseIDAzureLinux}, map[string][]byte{ + "root.crt": payload, + })) + got, err := os.ReadFile(c.path("/etc/pki/ca-trust/source/anchors/root.crt")) + require.NoError(t, err) + require.Equal(t, payload, got) + found := false + for _, record := range logs.getRecords() { + for _, value := range record.Attrs { + require.NotContains(t, value, "PRIVATE KEY") + if strings.Contains(value, "BEGIN CERTIFICATE") { + found = true + } + } + } + require.True(t, found) +} diff --git a/aks-node-controller/cloud_certificate_store_test.go b/aks-node-controller/cloud_certificate_store_test.go new file mode 100644 index 00000000000..f8f37c6198d --- /dev/null +++ b/aks-node-controller/cloud_certificate_store_test.go @@ -0,0 +1,18 @@ +package main + +import ( + "os" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestCloudCertificateUbuntuBundleSymlink(t *testing.T) { + c := newTestCloudInitializer(t, cloudTestUbuntu) + require.NoError(t, os.MkdirAll(c.path("/usr/lib/ssl"), 0o755)) + require.NoError(t, os.Symlink(c.path("/etc/ssl/certs/ca-certificates.crt"), c.path("/usr/lib/ssl/cert.pem"))) + require.NoError(t, c.copyOpenSSLBundle()) + info, err := os.Lstat(c.path("/usr/lib/ssl/cert.pem")) + require.NoError(t, err) + require.NotZero(t, info.Mode()&os.ModeSymlink) +} diff --git a/aks-node-controller/cloud_runtime.go b/aks-node-controller/cloud_runtime.go new file mode 100644 index 00000000000..2bf139083c7 --- /dev/null +++ b/aks-node-controller/cloud_runtime.go @@ -0,0 +1,183 @@ +package main + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "os" + "os/exec" + "path/filepath" + "strings" + "time" + + "github.com/Azure/agentbaker/aks-node-controller/common" + "github.com/Azure/agentbaker/aks-node-controller/helpers" +) + +const cloudResponseLimit = 8 << 20 + +const cloudCommandTimeout = 2 * time.Minute + +type cloudPlatform struct { + ID, Version, Codename string +} + +type cloudInitializer struct { + run func(*exec.Cmd) error + wait func(context.Context, time.Duration) error + wireClient *http.Client + repoClient *http.Client + wireEndpoint string + eventLogger *helpers.EventLogger + rootDir string + osReleasePath string + executable string +} + +func newCloudInitializer(a *App) *cloudInitializer { + transport := common.NewBaseTransport(common.HTTPTransportOptions{ + DialTimeout: 10 * time.Second, + TLSHandshakeTimeout: 10 * time.Second, + ResponseHeaderTimeout: 30 * time.Second, + }) + repoTransport := transport.Clone() + repoTransport.Proxy = http.ProxyFromEnvironment + return &cloudInitializer{ + run: a.cmdRun, + wait: waitCloudRetry, + wireEndpoint: wireServerEndpoint, + wireClient: &http.Client{ + Transport: transport, + Timeout: 30 * time.Second, + // WireServer responses must not redirect to another endpoint. + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, + }, + repoClient: &http.Client{Transport: repoTransport, Timeout: 30 * time.Second}, + eventLogger: a.eventLogger, + osReleasePath: a.osReleasePath, + } +} + +func (c *cloudInitializer) close() { + c.wireClient.CloseIdleConnections() + c.repoClient.CloseIdleConnections() +} + +func (c *cloudInitializer) platform() (cloudPlatform, error) { + path := c.osReleasePath + if path == "" { + path = c.path(defaultOSReleasePath) + } + data, err := os.ReadFile(path) + if err != nil { + return cloudPlatform{}, fmt.Errorf("read cloud initialization OS release: %w", err) + } + values := parseNodeCustomDataOSRelease(data) + id := strings.ToLower(values["ID"]) + switch { + case strings.Contains(values["NAME"], "Ubuntu"), id == osReleaseIDUbuntu: + id = osReleaseIDUbuntu + case strings.Contains(id, osReleaseIDFlatcar): + id = osReleaseIDFlatcar + case id == osReleaseIDAzureContainerLinux, + id == osReleaseIDAzureLinux && values["VARIANT_ID"] == osReleaseIDAzureContainerLinux: + id = osReleaseIDAzureContainerLinux + case strings.Contains(values["NAME"], "Mariner"), id == osReleaseIDMariner: + id = osReleaseIDMariner + case strings.Contains(values["NAME"], "Microsoft Azure Linux"), id == osReleaseIDAzureLinux: + id = osReleaseIDAzureLinux + default: + return cloudPlatform{}, fmt.Errorf("unsupported cloud initialization OS %q", id) + } + return cloudPlatform{ID: id, Version: values["VERSION_ID"], Codename: values["VERSION_CODENAME"]}, nil +} + +func (c *cloudInitializer) path(path string) string { + if c.rootDir == "" { + return path + } + return filepath.Join(c.rootDir, strings.TrimPrefix(path, "/")) +} + +func (c *cloudInitializer) writeFile(path string, data []byte) error { + return writeCloudFile(c.path(path), data) +} + +func writeCloudFile(path string, data []byte) error { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + root, err := os.OpenRoot(filepath.Dir(path)) + if err != nil { + return err + } + defer root.Close() + // #nosec G306 -- Public CAs, package keys, and repository metadata must be readable by unprivileged clients. + return root.WriteFile(filepath.Base(path), data, 0o644) +} + +func (c *cloudInitializer) withEvent(task string, action func() error) error { + start := time.Now() + err := action() + level, message := helpers.EventLevelInformational, "Completed" + if err != nil { + level, message = helpers.EventLevelError, err.Error() + slog.Error("Azure cloud operation failed", "task", task, "error", err) + } + if c.eventLogger != nil { + c.eventLogger.LogEvent(task, message, level, start, time.Now()) + } + return err +} + +func (c *cloudInitializer) command(ctx context.Context, input []byte, name string, args ...string) ([]byte, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + ctx, cancel := context.WithTimeout(ctx, cloudCommandTimeout) + defer cancel() + cmd := exec.CommandContext(ctx, name, args...) + cmd.Env = append(os.Environ(), "LC_ALL=C") + cmd.Stdin = bytes.NewReader(input) + var stdout, stderr bytes.Buffer + cmd.Stdout, cmd.Stderr = &stdout, &stderr + run := c.run + if run == nil { + run = cmdRunner + } + if err := run(cmd); err != nil { + return stdout.Bytes(), fmt.Errorf("%s: %w: %s", name, err, strings.TrimSpace(stderr.String())) + } + if name == "dnf" { + // DNF can report unavailable repositories on stderr despite exiting zero. + _, _ = stdout.Write(stderr.Bytes()) + } + return stdout.Bytes(), nil +} + +func waitCloudRetry(ctx context.Context, delay time.Duration) error { + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + +func readCloudResponse(resp *http.Response) ([]byte, error) { + defer resp.Body.Close() + body, err := io.ReadAll(io.LimitReader(resp.Body, cloudResponseLimit+1)) + if err != nil { + return nil, err + } + if len(body) > cloudResponseLimit { + return nil, errors.New("cloud response exceeds size limit") + } + return body, nil +} diff --git a/aks-node-controller/cloud_runtime_events_test.go b/aks-node-controller/cloud_runtime_events_test.go new file mode 100644 index 00000000000..a85583a5010 --- /dev/null +++ b/aks-node-controller/cloud_runtime_events_test.go @@ -0,0 +1,27 @@ +package main + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestCloudRuntimeOperationEvents(t *testing.T) { + for _, tc := range []struct { + name string + err error + }{ + {"completed", nil}, + {"failed", errors.New("operation failed")}, + } { + t.Run(tc.name, func(t *testing.T) { + c := newTestCloudInitializer(t, cloudTestUbuntu) + err := c.withEvent("CloudRuntimeProbe", func() error { return tc.err }) + require.ErrorIs(t, err, tc.err) + events := c.eventLogger.Events() + require.Len(t, events, 1) + require.Equal(t, "AKS.AKSNodeController.CloudRuntimeProbe", events[0].TaskName) + }) + } +} diff --git a/aks-node-controller/cloud_runtime_test.go b/aks-node-controller/cloud_runtime_test.go new file mode 100644 index 00000000000..1f8669ff1bf --- /dev/null +++ b/aks-node-controller/cloud_runtime_test.go @@ -0,0 +1,68 @@ +package main + +import ( + "io" + "net/http" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestCloudFileWrites(t *testing.T) { + t.Run("create and replace", func(t *testing.T) { + path := filepath.Join(t.TempDir(), "certificates", "root.crt") + require.NoError(t, writeCloudFile(path, []byte("original"))) + require.NoError(t, writeCloudFile(path, []byte("replacement"))) + data, err := os.ReadFile(path) + require.NoError(t, err) + require.Equal(t, "replacement", string(data)) + }) + t.Run("reject symlinks outside the destination directory", func(t *testing.T) { + directory := t.TempDir() + outside := filepath.Join(t.TempDir(), "outside.crt") + require.NoError(t, os.WriteFile(outside, []byte("original"), 0o600)) + path := filepath.Join(directory, "root.crt") + require.NoError(t, os.Symlink(outside, path)) + require.Error(t, writeCloudFile(path, []byte("replacement"))) + data, err := os.ReadFile(outside) + require.NoError(t, err) + require.Equal(t, "original", string(data)) + }) +} + +func TestCloudPlatform(t *testing.T) { + for _, tc := range []struct { + name, release, want string + }{ + {"Ubuntu", cloudTestUbuntu, osReleaseIDUbuntu}, + {"Mariner", "ID=mariner\nNAME=CBL-Mariner\nVERSION_ID=2.0\n", "mariner"}, + {"AzureLinux", "ID=azurelinux\nNAME=\"Microsoft Azure Linux\"\nVERSION_ID=3.0\n", osReleaseIDAzureLinux}, + {"OSGuard", "ID=azurelinux\nVARIANT_ID=osguard\nVERSION_ID=3.0\n", osReleaseIDAzureLinux}, + {"Flatcar", "ID=flatcar\n", osReleaseIDFlatcar}, + {"ACL", "ID=azurecontainerlinux\n", osReleaseIDAzureContainerLinux}, + {"ACL variant", "ID=azurelinux\nVARIANT_ID=azurecontainerlinux\n", osReleaseIDAzureContainerLinux}, + {"unsupported", "ID=debian\n", ""}, + {"missing ID", "", ""}, + } { + t.Run(tc.name, func(t *testing.T) { + c := newTestCloudInitializer(t, tc.release) + platform, err := c.platform() + if tc.want == "" { + require.Error(t, err) + } else { + require.NoError(t, err) + require.Equal(t, tc.want, platform.ID) + } + }) + } +} + +func TestCloudCertificateResponseLimits(t *testing.T) { + _, err := readCloudResponse(&http.Response{ + Body: io.NopCloser(strings.NewReader(strings.Repeat("x", cloudResponseLimit+1))), + }) + require.ErrorContains(t, err, "size limit") +} diff --git a/aks-node-controller/cloud_test_helpers_test.go b/aks-node-controller/cloud_test_helpers_test.go new file mode 100644 index 00000000000..d99bac48b43 --- /dev/null +++ b/aks-node-controller/cloud_test_helpers_test.go @@ -0,0 +1,59 @@ +package main + +import ( + "context" + "crypto/ed25519" + "crypto/rand" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "io" + "math/big" + "os/exec" + "path/filepath" + "testing" + "time" + + "github.com/Azure/agentbaker/aks-node-controller/helpers" + "github.com/stretchr/testify/require" +) + +const cloudTestUbuntu = "ID=ubuntu\nNAME=Ubuntu\nVERSION_ID=22.04\nVERSION_CODENAME=jammy\n" + +func newTestCloudInitializer(t *testing.T, release string) *cloudInitializer { + t.Helper() + c := newCloudInitializer(&App{ + cmdRun: func(cmd *exec.Cmd) error { + if filepath.Base(cmd.Path) == "gpg" { + _, _ = io.WriteString(cmd.Stdout, "dearmored-key") + } + return nil + }, + eventLogger: helpers.NewEventLogger(t.TempDir()), + }) + c.rootDir = t.TempDir() + c.executable = hotfixBinaryPath + c.wait = func(ctx context.Context, _ time.Duration) error { return ctx.Err() } + require.NoError(t, c.writeFile(defaultOSReleasePath, []byte(release))) + require.NoError(t, c.writeFile("/etc/ssl/certs/ca-certificates.crt", []byte("system-ca-bundle"))) + t.Cleanup(c.close) + return c +} + +func cloudTestCertificate(t *testing.T) []byte { + t.Helper() + public, private, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + template := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "ANC cloud initialization test CA"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + IsCA: true, + BasicConstraintsValid: true, + KeyUsage: x509.KeyUsageCertSign, + } + der, err := x509.CreateCertificate(rand.Reader, template, template, public, private) + require.NoError(t, err) + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) +} diff --git a/aks-node-controller/embeddednodecustomdata.go b/aks-node-controller/embeddednodecustomdata.go index 67615d54fcb..1f1ee70c0e5 100644 --- a/aks-node-controller/embeddednodecustomdata.go +++ b/aks-node-controller/embeddednodecustomdata.go @@ -20,6 +20,7 @@ const ( osReleaseIDAzureLinux = "azurelinux" osReleaseIDFlatcar = "flatcar" osReleaseIDUbuntu = "ubuntu" + osReleaseIDMariner = "mariner" osVariantIDOSGuard = "osguard" ) @@ -27,7 +28,7 @@ type nodeCustomDataPlatform string const ( nodeCustomDataPlatformUbuntu nodeCustomDataPlatform = "ubuntu" - nodeCustomDataPlatformMariner nodeCustomDataPlatform = "mariner" + nodeCustomDataPlatformMariner nodeCustomDataPlatform = osReleaseIDMariner nodeCustomDataPlatformUnsupported nodeCustomDataPlatform = "unsupported" ) From 04f264a1f8e7f17b65e415223d51a127c42e01d0 Mon Sep 17 00:00:00 2001 From: Abigailliang Date: Wed, 28 Oct 2026 17:13:23 -0700 Subject: [PATCH 2/3] feat: enhance cloud command handling with process group management and cancellation support --- aks-node-controller/README.md | 5 + aks-node-controller/cloud_runtime.go | 18 ++- aks-node-controller/cloud_runtime_test.go | 143 ++++++++++++++++++ .../cloud-init/artifacts/init-aks-cloud.sh | 4 +- .../artifacts/init_aks_cloud_logging_spec.sh | 8 + 5 files changed, 174 insertions(+), 4 deletions(-) diff --git a/aks-node-controller/README.md b/aks-node-controller/README.md index 89ea6e29688..f803ffa5a7e 100644 --- a/aks-node-controller/README.md +++ b/aks-node-controller/README.md @@ -61,6 +61,11 @@ execution, public-file writes, and guest-agent events. The trust-store helpers write downloaded contents without PEM/X.509 filtering, update the platform's existing CA tools, and preserve Ubuntu's OpenSSL bundle/symlink behavior. +Cloud commands run in their own process groups with a two-minute timeout, or +the caller's earlier cancellation. Cancellation kills the entire group. A +five-second `WaitDelay` bounds output draining after cancellation or command +exit, including when a descendant leaves the group but retains an output pipe. + Cache cleanup removes flat staging files, including hidden files, without recursively deleting directories or removing installed system-trust certificates. `AKS_CLOUD_LOG_CERTIFICATES=true` enables public-certificate-only diagnostics; diff --git a/aks-node-controller/cloud_runtime.go b/aks-node-controller/cloud_runtime.go index 2bf139083c7..4781d41b8e4 100644 --- a/aks-node-controller/cloud_runtime.go +++ b/aks-node-controller/cloud_runtime.go @@ -12,6 +12,7 @@ import ( "os/exec" "path/filepath" "strings" + "syscall" "time" "github.com/Azure/agentbaker/aks-node-controller/common" @@ -20,7 +21,10 @@ import ( const cloudResponseLimit = 8 << 20 -const cloudCommandTimeout = 2 * time.Minute +const ( + cloudCommandTimeout = 2 * time.Minute + cloudCommandWaitDelay = 5 * time.Second +) type cloudPlatform struct { ID, Version, Codename string @@ -84,7 +88,7 @@ func (c *cloudInitializer) platform() (cloudPlatform, error) { case strings.Contains(id, osReleaseIDFlatcar): id = osReleaseIDFlatcar case id == osReleaseIDAzureContainerLinux, - id == osReleaseIDAzureLinux && values["VARIANT_ID"] == osReleaseIDAzureContainerLinux: + id == osReleaseIDAzureLinux && strings.ToLower(values["VARIANT_ID"]) == osReleaseIDAzureContainerLinux: id = osReleaseIDAzureContainerLinux case strings.Contains(values["NAME"], "Mariner"), id == osReleaseIDMariner: id = osReleaseIDMariner @@ -141,6 +145,16 @@ func (c *cloudInitializer) command(ctx context.Context, input []byte, name strin ctx, cancel := context.WithTimeout(ctx, cloudCommandTimeout) defer cancel() cmd := exec.CommandContext(ctx, name, args...) + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + cmd.Cancel = func() error { + err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) + if errors.Is(err, syscall.ESRCH) { + return os.ErrProcessDone + } + return err + } + // A descendant can leave the process group but still hold an output pipe open. + cmd.WaitDelay = cloudCommandWaitDelay cmd.Env = append(os.Environ(), "LC_ALL=C") cmd.Stdin = bytes.NewReader(input) var stdout, stderr bytes.Buffer diff --git a/aks-node-controller/cloud_runtime_test.go b/aks-node-controller/cloud_runtime_test.go index 1f8669ff1bf..5ef41cfb035 100644 --- a/aks-node-controller/cloud_runtime_test.go +++ b/aks-node-controller/cloud_runtime_test.go @@ -1,12 +1,19 @@ package main import ( + "context" + "errors" + "fmt" "io" "net/http" "os" + "os/exec" "path/filepath" + "strconv" "strings" + "syscall" "testing" + "time" "github.com/stretchr/testify/require" ) @@ -44,6 +51,7 @@ func TestCloudPlatform(t *testing.T) { {"Flatcar", "ID=flatcar\n", osReleaseIDFlatcar}, {"ACL", "ID=azurecontainerlinux\n", osReleaseIDAzureContainerLinux}, {"ACL variant", "ID=azurelinux\nVARIANT_ID=azurecontainerlinux\n", osReleaseIDAzureContainerLinux}, + {"Mixed-case ACL variant", "ID=azurelinux\nVARIANT_ID=AzureContainerLinux\n", osReleaseIDAzureContainerLinux}, {"unsupported", "ID=debian\n", ""}, {"missing ID", "", ""}, } { @@ -66,3 +74,138 @@ func TestCloudCertificateResponseLimits(t *testing.T) { }) require.ErrorContains(t, err, "size limit") } + +func TestCloudCommandSuccess(t *testing.T) { + c := &cloudInitializer{} + output, err := c.command(t.Context(), []byte("input"), os.Args[0], + "-test.run=^TestCloudCommandHelperProcess$", "--", "echo", "") + require.NoError(t, err) + require.Equal(t, "C:input", string(output)) +} + +func TestCloudCommandAlreadyCanceled(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + called := false + c := &cloudInitializer{run: func(cmd *exec.Cmd) error { + called = true + return cmd.Run() + }} + output, err := c.command(ctx, nil, os.Args[0], + "-test.run=^TestCloudCommandHelperProcess$", "--", "echo", "") + require.ErrorIs(t, err, context.Canceled) + require.False(t, called, "an already-canceled context must not invoke the command runner") + require.Nil(t, output) +} + +func TestCloudCommandCancellation(t *testing.T) { + for _, tc := range []struct { + name string + mode string + returnWithin time.Duration + }{ + {"kills descendants with inherited output", "group", 2 * time.Second}, + {"bounds output draining after a descendant leaves the group", "detached", 7 * time.Second}, + } { + t.Run(tc.name, func(t *testing.T) { + readyPath := filepath.Join(t.TempDir(), "child.pid") + ctx, cancel := context.WithCancel(t.Context()) + started := make(chan *os.Process, 1) + finished := make(chan struct{}) + c := &cloudInitializer{run: func(cmd *exec.Cmd) error { + if err := cmd.Start(); err != nil { + return err + } + started <- cmd.Process + return cmd.Wait() + }} + var parent, child *os.Process + t.Cleanup(func() { + cancel() + for _, process := range []*os.Process{child, parent} { + if process == nil { + continue + } + if err := process.Kill(); err != nil && + !errors.Is(err, os.ErrProcessDone) && !errors.Is(err, syscall.ESRCH) { + t.Errorf("clean up process %d: %v", process.Pid, err) + } + } + if child != nil { + defer child.Release() + } + select { + case <-finished: + case <-time.After(5 * time.Second): + t.Error("command did not finish after subprocess cleanup") + } + }) + var output []byte + var commandErr error + go func() { + defer close(finished) + output, commandErr = c.command(ctx, nil, os.Args[0], + "-test.run=^TestCloudCommandHelperProcess$", "--", tc.mode, readyPath) + }() + select { + case parent = <-started: + case <-finished: + t.Fatalf("command exited before starting its child: %v", commandErr) + case <-time.After(5 * time.Second): + t.Fatal("command did not start") + } + var childPID int + require.Eventually(t, func() bool { + data, err := os.ReadFile(readyPath) + if err != nil { + return false + } + childPID, err = strconv.Atoi(strings.TrimSpace(string(data))) + return err == nil && childPID > 0 + }, 5*time.Second, 10*time.Millisecond, "command did not start its output-holding child") + var err error + child, err = os.FindProcess(childPID) + require.NoError(t, err) + + cancel() + select { + case <-finished: + require.Error(t, commandErr) + require.Contains(t, string(output), "stdout-marker") + require.ErrorContains(t, commandErr, "stderr-marker") + case <-time.After(tc.returnWithin): + t.Fatalf("command did not return within %s of cancellation", tc.returnWithin) + } + }) + } +} + +func TestCloudCommandHelperProcess(t *testing.T) { + if len(os.Args) != 5 || os.Args[2] != "--" { + return + } + switch os.Args[3] { + case "echo": + input, err := io.ReadAll(os.Stdin) + require.NoError(t, err) + _, err = fmt.Fprintf(os.Stdout, "%s:%s", os.Getenv("LC_ALL"), input) + require.NoError(t, err) + case "hold": + time.Sleep(time.Minute) + case "group", "detached": + child := exec.Command(os.Args[0], os.Args[1], "--", "hold", os.Args[4]) + child.Stdout, child.Stderr = os.Stdout, os.Stderr + if os.Args[3] == "detached" { + child.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + } + require.NoError(t, child.Start()) + defer child.Process.Kill() + fmt.Fprintln(os.Stdout, "stdout-marker") + fmt.Fprintln(os.Stderr, "stderr-marker") + require.NoError(t, os.WriteFile(os.Args[4], []byte(strconv.Itoa(child.Process.Pid)), 0o600)) + require.NoError(t, child.Wait()) + default: + t.Fatalf("unknown command helper mode %q", os.Args[3]) + } + os.Exit(0) +} diff --git a/parts/linux/cloud-init/artifacts/init-aks-cloud.sh b/parts/linux/cloud-init/artifacts/init-aks-cloud.sh index 76fdf3c9ea7..b2abd419d60 100644 --- a/parts/linux/cloud-init/artifacts/init-aks-cloud.sh +++ b/parts/linux/cloud-init/artifacts/init-aks-cloud.sh @@ -271,7 +271,7 @@ function retrieve_rcv1p_certs { } function install_certs_to_trust_store { - mkdir -p /root/AzureCACertificates + mkdir -p /root/AzureCACertificates || return 1 # Guard against empty glob: if no *.crt files exist, bash leaves the literal # '*.crt', which would fail cp. @@ -280,7 +280,7 @@ function install_certs_to_trust_store { return 1 fi - local cert + local cert destcert for cert in /root/AzureCACertificates/*.crt; do log_certificates "$cert" done diff --git a/spec/parts/linux/cloud-init/artifacts/init_aks_cloud_logging_spec.sh b/spec/parts/linux/cloud-init/artifacts/init_aks_cloud_logging_spec.sh index 9c9993034f1..1845878f892 100644 --- a/spec/parts/linux/cloud-init/artifacts/init_aks_cloud_logging_spec.sh +++ b/spec/parts/linux/cloud-init/artifacts/init_aks_cloud_logging_spec.sh @@ -167,6 +167,14 @@ Describe 'init-aks-cloud.sh certificate logging' install_certs_to_trust_store } + It 'stops immediately when creating the certificate directory fails' + mkdir() { echo 'mkdir failed' >&2; return 1; } + When call install_with_diagnostics + The status should eq 1 + The stdout should be blank + The stderr should eq 'mkdir failed' + End + It 'logs public PEMs through the actual installation path when opted in' make_certificate mkdir "${TEST_DIR}/certs" From e9c3c0749a01fcb4bac67d72651777edbb05002f Mon Sep 17 00:00:00 2001 From: Abigailliang Date: Wed, 28 Oct 2026 17:52:31 -0700 Subject: [PATCH 3/3] lint fix --- aks-node-controller/cloud_runtime_test.go | 72 +++++++++++++++-------- 1 file changed, 48 insertions(+), 24 deletions(-) diff --git a/aks-node-controller/cloud_runtime_test.go b/aks-node-controller/cloud_runtime_test.go index 5ef41cfb035..6cff4f13d66 100644 --- a/aks-node-controller/cloud_runtime_test.go +++ b/aks-node-controller/cloud_runtime_test.go @@ -76,14 +76,18 @@ func TestCloudCertificateResponseLimits(t *testing.T) { } func TestCloudCommandSuccess(t *testing.T) { + executable, err := os.Executable() + require.NoError(t, err) c := &cloudInitializer{} - output, err := c.command(t.Context(), []byte("input"), os.Args[0], + output, err := c.command(t.Context(), []byte("input"), executable, "-test.run=^TestCloudCommandHelperProcess$", "--", "echo", "") require.NoError(t, err) require.Equal(t, "C:input", string(output)) } func TestCloudCommandAlreadyCanceled(t *testing.T) { + executable, err := os.Executable() + require.NoError(t, err) ctx, cancel := context.WithCancel(t.Context()) cancel() called := false @@ -91,7 +95,7 @@ func TestCloudCommandAlreadyCanceled(t *testing.T) { called = true return cmd.Run() }} - output, err := c.command(ctx, nil, os.Args[0], + output, err := c.command(ctx, nil, executable, "-test.run=^TestCloudCommandHelperProcess$", "--", "echo", "") require.ErrorIs(t, err, context.Canceled) require.False(t, called, "an already-canceled context must not invoke the command runner") @@ -99,6 +103,8 @@ func TestCloudCommandAlreadyCanceled(t *testing.T) { } func TestCloudCommandCancellation(t *testing.T) { + executable, err := os.Executable() + require.NoError(t, err) for _, tc := range []struct { name string mode string @@ -121,30 +127,13 @@ func TestCloudCommandCancellation(t *testing.T) { }} var parent, child *os.Process t.Cleanup(func() { - cancel() - for _, process := range []*os.Process{child, parent} { - if process == nil { - continue - } - if err := process.Kill(); err != nil && - !errors.Is(err, os.ErrProcessDone) && !errors.Is(err, syscall.ESRCH) { - t.Errorf("clean up process %d: %v", process.Pid, err) - } - } - if child != nil { - defer child.Release() - } - select { - case <-finished: - case <-time.After(5 * time.Second): - t.Error("command did not finish after subprocess cleanup") - } + cleanupCloudCommand(t, cancel, finished, parent, child) }) var output []byte var commandErr error go func() { defer close(finished) - output, commandErr = c.command(ctx, nil, os.Args[0], + output, commandErr = c.command(ctx, nil, executable, "-test.run=^TestCloudCommandHelperProcess$", "--", tc.mode, readyPath) }() select { @@ -180,10 +169,44 @@ func TestCloudCommandCancellation(t *testing.T) { } } +func cleanupCloudCommand(t *testing.T, cancel context.CancelFunc, finished <-chan struct{}, parent, child *os.Process) { + t.Helper() + cancel() + killCloudCommandTestProcess(t, child) + killCloudCommandTestProcess(t, parent) + select { + case <-finished: + case <-time.After(5 * time.Second): + t.Error("command did not finish after subprocess cleanup") + } + if child != nil { + if err := child.Release(); err != nil { + t.Errorf("release child process: %v", err) + } + } +} + +func killCloudCommandTestProcess(t *testing.T, process *os.Process) { + t.Helper() + if process == nil { + return + } + if err := process.Kill(); err != nil && + !errors.Is(err, os.ErrProcessDone) && !errors.Is(err, syscall.ESRCH) { + t.Errorf("clean up process %d: %v", process.Pid, err) + } +} + func TestCloudCommandHelperProcess(t *testing.T) { if len(os.Args) != 5 || os.Args[2] != "--" { return } + runCloudCommandHelperProcess(t) + os.Exit(0) +} + +func runCloudCommandHelperProcess(t *testing.T) { + t.Helper() switch os.Args[3] { case "echo": input, err := io.ReadAll(os.Stdin) @@ -193,13 +216,15 @@ func TestCloudCommandHelperProcess(t *testing.T) { case "hold": time.Sleep(time.Minute) case "group", "detached": - child := exec.Command(os.Args[0], os.Args[1], "--", "hold", os.Args[4]) + executable, err := os.Executable() + require.NoError(t, err) + child := exec.Command(executable, os.Args[1], "--", "hold", os.Args[4]) child.Stdout, child.Stderr = os.Stdout, os.Stderr if os.Args[3] == "detached" { child.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} } require.NoError(t, child.Start()) - defer child.Process.Kill() + defer killCloudCommandTestProcess(t, child.Process) fmt.Fprintln(os.Stdout, "stdout-marker") fmt.Fprintln(os.Stderr, "stderr-marker") require.NoError(t, os.WriteFile(os.Args[4], []byte(strconv.Itoa(child.Process.Pid)), 0o600)) @@ -207,5 +232,4 @@ func TestCloudCommandHelperProcess(t *testing.T) { default: t.Fatalf("unknown command helper mode %q", os.Args[3]) } - os.Exit(0) }