From 18b4c86232ce6fadfb3bb7e3ce7547679af60381 Mon Sep 17 00:00:00 2001 From: Matteo Bunino <48362942+matbun@users.noreply.github.com> Date: Wed, 29 Jul 2026 18:32:55 +0200 Subject: [PATCH 1/7] refactor: rename wstunnel shadow-pod naming to shadow The per-pod Deployment/Service the VK creates for offloaded pods with exposed ports is a shadow pod, not a wstunnel: wstunnel is only one of the things that can run inside it, and an SSH port-forward shadow runs no wstunnel at all. Rendered resource names change accordingly: Deployment/Service wstunnel-- -> shadow-- Namespace -wstunnel -> -shadow Go identifiers follow (WstunnelTemplateData -> ShadowTemplateData, createDummyPod -> createShadowPod, ...), and the naming/identity helpers move out of mesh.go into shadow.go, where they sit next to the code that uses them rather than next to WireGuard key handling. Pod annotations (interlink.eu/wstunnel-extra-ports, interlink.virtual-kubelet.io/wstunnel-timeout, ...) and the Wstunnel* config keys are deliberately untouched: renaming those breaks existing manifests and deployments for no user-visible gain. Names that genuinely refer to wstunnel - the client command, the embedded wstunnel templates, the download URLs - keep their names. Upgrade note: shadows created before this change keep their old names until their pods are recreated. Anything hard-coding a shadow Service name (Ingress, VirtualService, admission policies) needs updating. Signed-off-by: Matteo Bunino <48362942+matbun@users.noreply.github.com> Co-Authored-By: Claude Opus 5 (1M context) --- pkg/virtualkubelet/mesh.go | 217 +---------------- pkg/virtualkubelet/shadow.go | 228 ++++++++++++++++++ pkg/virtualkubelet/virtualkubelet.go | 162 ++++++------- .../virtualkubelet_mesh_test.go | 52 ++-- 4 files changed, 337 insertions(+), 322 deletions(-) create mode 100644 pkg/virtualkubelet/shadow.go diff --git a/pkg/virtualkubelet/mesh.go b/pkg/virtualkubelet/mesh.go index e4c05b25..4e38c520 100644 --- a/pkg/virtualkubelet/mesh.go +++ b/pkg/virtualkubelet/mesh.go @@ -4,9 +4,7 @@ import ( "bytes" "context" "crypto/rand" - "crypto/sha256" "encoding/base64" - "encoding/hex" "encoding/json" "fmt" "os" @@ -20,217 +18,6 @@ import ( k8stypes "k8s.io/apimachinery/pkg/types" ) -func sanitizeDNSName(name string) string { - // Convert to lowercase - name = strings.ToLower(name) - - // Replace any invalid characters with hyphens - var builder strings.Builder - for _, r := range name { - if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') { - builder.WriteRune(r) - } else { - builder.WriteRune('-') - } - } - name = builder.String() - - // Remove leading and trailing hyphens - name = strings.Trim(name, "-") - - // Collapse consecutive hyphens into a single hyphen - for strings.Contains(name, "--") { - name = strings.ReplaceAll(name, "--", "-") - } - - // Truncate to 63 characters (max label length) - if len(name) > 63 { - name = name[:63] - // Ensure we don't end with a hyphen after truncation - name = strings.TrimRight(name, "-") - } - - // If the result is empty, provide a default - if name == "" { - name = "default" - } - - return name -} - -// sanitizeFullDNSName sanitizes a full DNS name (with dots) to ensure it meets RFC 1123 requirements -func sanitizeFullDNSName(fullName string) string { - // Split by dots to handle each label separately - labels := strings.Split(fullName, ".") - - // Sanitize each label - sanitizedLabels := make([]string, 0, len(labels)) - for _, label := range labels { - if label == "" { - continue - } - sanitized := sanitizeDNSName(label) - if sanitized != "" { - sanitizedLabels = append(sanitizedLabels, sanitized) - } - } - - // Rejoin with dots - result := strings.Join(sanitizedLabels, ".") - - // Ensure total length doesn't exceed 253 characters - if len(result) > 253 { - // Truncate from the beginning (keeping the domain suffix) - excess := len(result) - 253 - result = result[excess:] - // Make sure we don't start with a dot after truncation - result = strings.TrimLeft(result, ".") - } - - return result -} - -// uniqueTruncate shortens a name and adds an 8-character hash to prevent naming collisions. -func uniqueTruncate(s string, maxLen int, full string) string { - if len(s) <= maxLen { - return s - } - h := sha256.Sum256([]byte(full)) - suffix := hex.EncodeToString(h[:4]) - keep := maxLen - len(suffix) - 1 - return strings.TrimRight(s[:keep], "-") + "-" + suffix -} - -type wstunnelResourceIdentity struct { - Name string - Namespace string -} - -func isShadowSameNamespace(pod *v1.Pod) bool { - if pod == nil || pod.Annotations == nil { - return false - } - return pod.Annotations["interlink.eu/shadow-same-ns"] == "true" -} - -func computeWstunnelResourceIdentity(pod *v1.Pod) (wstunnelResourceIdentity, error) { - if pod == nil { - return wstunnelResourceIdentity{}, fmt.Errorf("pod is nil") - } - if pod.Namespace == "" { - return wstunnelResourceIdentity{}, fmt.Errorf("pod namespace is empty") - } - - var name, namespace string - if isShadowSameNamespace(pod) { - name, namespace = computeWstunnelResourceNamesForSameNamespace(pod.Name, pod.Namespace) - } else { - name, namespace = computeWstunnelResourceNames(pod.Name, pod.Namespace) - } - - identity := wstunnelResourceIdentity{Name: name, Namespace: namespace} - if len(fmt.Sprintf("%s-%s", identity.Name, identity.Namespace)) > 63 { - return wstunnelResourceIdentity{}, fmt.Errorf("wstunnel ingress hostname label %q exceeds 63 characters; shorten pod/namespace or disable interlink.eu/shadow-same-ns", fmt.Sprintf("%s-%s", identity.Name, identity.Namespace)) - } - - return identity, nil -} - -func computeWstunnelResourceNamesForSameNamespace(podName, podNamespace string) (resourceBaseName, namespace string) { - // Sanitize namespace and pod name for DNS compliance - sanitizedNamespace := sanitizeDNSName(podNamespace) - sanitizedPodName := sanitizeDNSName(podName) - - // Use the original namespace. Do not truncate it: in same-namespace mode - // resources must be created in the pod's real namespace. - namespace = podNamespace - - // Create a unique resource name to avoid conflicts in the same namespace - // Add "wstunnel-" prefix to distinguish shadow pod resources - resourceBaseName = "wstunnel-" + sanitizedPodName + "-" + sanitizedNamespace - // Hash on the unsanitized names: sanitizeDNSName truncates to 63 chars, long pods could still collide - fullBaseName := "wstunnel-" + podName + "-" + podNamespace - - // Ensure resourceBaseName doesn't exceed 63 characters - if len(resourceBaseName) > 63 { - // Truncate while keeping some of both names - maxPodNameLen := 28 - maxNsLen := 28 - if len(sanitizedPodName) > maxPodNameLen { - sanitizedPodName = uniqueTruncate(sanitizedPodName, maxPodNameLen, fullBaseName) - } - if len(sanitizedNamespace) > maxNsLen { - sanitizedNamespace = sanitizedNamespace[:maxNsLen] - } - resourceBaseName = "wstunnel-" + sanitizedPodName + "-" + sanitizedNamespace - resourceBaseName = strings.TrimRight(resourceBaseName, "-") - } - - // Additional check for total length after combining with namespace - ingressFirstLabel := fmt.Sprintf("%s-%s", resourceBaseName, namespace) - if len(ingressFirstLabel) > 63 { - maxNameLen := 63 - len(namespace) - 1 - if maxNameLen > 9 && len(resourceBaseName) > maxNameLen { // >9: room for the 8-char hash suffix - resourceBaseName = uniqueTruncate(resourceBaseName, maxNameLen, fullBaseName) - } - } - - return resourceBaseName, namespace -} - -func computeWstunnelResourceNames(podName, podNamespace string) (resourceBaseName, wstunnelNamespace string) { - // Sanitize namespace and pod name for DNS compliance - sanitizedNamespace := sanitizeDNSName(podNamespace) - sanitizedPodName := sanitizeDNSName(podName) - - wstunnelNamespace = sanitizedNamespace + "-wstunnel" - // Ensure wstunnelNamespace is valid (max 63 chars for namespace) - if len(wstunnelNamespace) > 63 { - wstunnelNamespace = sanitizedNamespace[:min(54, len(sanitizedNamespace))] + "-wstunnel" - } - - resourceBaseName = sanitizedPodName + "-" + sanitizedNamespace - fullBaseName := podName + "-" + podNamespace - // Ensure resourceBaseName doesn't exceed 63 characters - if len(resourceBaseName) > 63 { - // Truncate while keeping some of both names - maxPodNameLen := 31 - maxNsLen := 31 - if len(sanitizedPodName) > maxPodNameLen { - sanitizedPodName = uniqueTruncate(sanitizedPodName, maxPodNameLen, fullBaseName) - } - if len(sanitizedNamespace) > maxNsLen { - sanitizedNamespace = sanitizedNamespace[:maxNsLen] - } - resourceBaseName = sanitizedPodName + "-" + sanitizedNamespace - resourceBaseName = strings.TrimRight(resourceBaseName, "-") - } - - ingressFirstLabel := fmt.Sprintf("%s-%s", resourceBaseName, wstunnelNamespace) - if len(ingressFirstLabel) > 63 { - // If combined length exceeds 63, we need to truncate - // Strategy: keep both parts but truncate proportionally - maxNameLen := 31 - maxNsLen := 31 - - truncatedName := resourceBaseName - if len(truncatedName) > maxNameLen { - truncatedName = uniqueTruncate(truncatedName, maxNameLen, fullBaseName) - } - - truncatedNs := wstunnelNamespace - if len(truncatedNs) > maxNsLen { - truncatedNs = truncatedNs[:maxNsLen] - truncatedNs = strings.TrimRight(truncatedNs, "-") - } - - resourceBaseName = truncatedName - wstunnelNamespace = truncatedNs - } - - return resourceBaseName, wstunnelNamespace -} - func generateWGKeypair() (string, string, error) { // 32 random bytes -> clamp per X25519 rules -> public = X25519(priv, basepoint) privRaw := make([]byte, 32) @@ -273,7 +60,7 @@ func deriveWGPublicKey(privB64 string) (string, error) { } // addWstunnelClientAnnotation adds the wstunnel client command annotation to the original pod -func (p *Provider) addWstunnelClientAnnotation(ctx context.Context, pod *v1.Pod, td *WstunnelTemplateData) error { +func (p *Provider) addWstunnelClientAnnotation(ctx context.Context, pod *v1.Pod, td *ShadowTemplateData) error { if pod.Annotations == nil { pod.Annotations = make(map[string]string) } @@ -399,7 +186,7 @@ func clearConflictingNetworkAnnotations(pod *v1.Pod, fullMeshEnabledForPod bool) delete(pod.Annotations, annWGClientSnippet) } -func (p *Provider) generateFullMeshScript(ctx context.Context, td *WstunnelTemplateData, ingressEndpoint string, podUID string) (string, error) { +func (p *Provider) generateFullMeshScript(ctx context.Context, td *ShadowTemplateData, ingressEndpoint string, podUID string) (string, error) { serverPub, err := deriveWGPublicKey(td.WGPrivateKey) if err != nil { diff --git a/pkg/virtualkubelet/shadow.go b/pkg/virtualkubelet/shadow.go new file mode 100644 index 00000000..aa6273f8 --- /dev/null +++ b/pkg/virtualkubelet/shadow.go @@ -0,0 +1,228 @@ +package virtualkubelet + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "strings" + + v1 "k8s.io/api/core/v1" +) + +// shadowResourcePrefix is prepended to shadow resource names created in the +// offloaded pod's own namespace, so they cannot collide with the pod itself. +const shadowResourcePrefix = "shadow-" + +// shadowNamespaceSuffix is appended to the offloaded pod's namespace to build +// the dedicated namespace shadow resources live in by default. +const shadowNamespaceSuffix = "-shadow" + +func sanitizeDNSName(name string) string { + // Convert to lowercase + name = strings.ToLower(name) + + // Replace any invalid characters with hyphens + var builder strings.Builder + for _, r := range name { + if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') { + builder.WriteRune(r) + } else { + builder.WriteRune('-') + } + } + name = builder.String() + + // Remove leading and trailing hyphens + name = strings.Trim(name, "-") + + // Collapse consecutive hyphens into a single hyphen + for strings.Contains(name, "--") { + name = strings.ReplaceAll(name, "--", "-") + } + + // Truncate to 63 characters (max label length) + if len(name) > 63 { + name = name[:63] + // Ensure we don't end with a hyphen after truncation + name = strings.TrimRight(name, "-") + } + + // If the result is empty, provide a default + if name == "" { + name = "default" + } + + return name +} + +// sanitizeFullDNSName sanitizes a full DNS name (with dots) to ensure it meets RFC 1123 requirements +func sanitizeFullDNSName(fullName string) string { + // Split by dots to handle each label separately + labels := strings.Split(fullName, ".") + + // Sanitize each label + sanitizedLabels := make([]string, 0, len(labels)) + for _, label := range labels { + if label == "" { + continue + } + sanitized := sanitizeDNSName(label) + if sanitized != "" { + sanitizedLabels = append(sanitizedLabels, sanitized) + } + } + + // Rejoin with dots + result := strings.Join(sanitizedLabels, ".") + + // Ensure total length doesn't exceed 253 characters + if len(result) > 253 { + // Truncate from the beginning (keeping the domain suffix) + excess := len(result) - 253 + result = result[excess:] + // Make sure we don't start with a dot after truncation + result = strings.TrimLeft(result, ".") + } + + return result +} + +// uniqueTruncate shortens a name and adds an 8-character hash to prevent naming collisions. +func uniqueTruncate(s string, maxLen int, full string) string { + if len(s) <= maxLen { + return s + } + h := sha256.Sum256([]byte(full)) + suffix := hex.EncodeToString(h[:4]) + keep := maxLen - len(suffix) - 1 + return strings.TrimRight(s[:keep], "-") + "-" + suffix +} + +type shadowResourceIdentity struct { + Name string + Namespace string +} + +func isShadowSameNamespace(pod *v1.Pod) bool { + if pod == nil || pod.Annotations == nil { + return false + } + return pod.Annotations["interlink.eu/shadow-same-ns"] == "true" +} + +func computeShadowResourceIdentity(pod *v1.Pod) (shadowResourceIdentity, error) { + if pod == nil { + return shadowResourceIdentity{}, fmt.Errorf("pod is nil") + } + if pod.Namespace == "" { + return shadowResourceIdentity{}, fmt.Errorf("pod namespace is empty") + } + + var name, namespace string + if isShadowSameNamespace(pod) { + name, namespace = computeShadowResourceNamesForSameNamespace(pod.Name, pod.Namespace) + } else { + name, namespace = computeShadowResourceNames(pod.Name, pod.Namespace) + } + + identity := shadowResourceIdentity{Name: name, Namespace: namespace} + if len(fmt.Sprintf("%s-%s", identity.Name, identity.Namespace)) > 63 { + return shadowResourceIdentity{}, fmt.Errorf("shadow ingress hostname label %q exceeds 63 characters; shorten pod/namespace or disable interlink.eu/shadow-same-ns", fmt.Sprintf("%s-%s", identity.Name, identity.Namespace)) + } + + return identity, nil +} + +func computeShadowResourceNamesForSameNamespace(podName, podNamespace string) (resourceBaseName, namespace string) { + // Sanitize namespace and pod name for DNS compliance + sanitizedNamespace := sanitizeDNSName(podNamespace) + sanitizedPodName := sanitizeDNSName(podName) + + // Use the original namespace. Do not truncate it: in same-namespace mode + // resources must be created in the pod's real namespace. + namespace = podNamespace + + // Create a unique resource name to avoid conflicts in the same namespace + resourceBaseName = shadowResourcePrefix + sanitizedPodName + "-" + sanitizedNamespace + // Hash on the unsanitized names: sanitizeDNSName truncates to 63 chars, long pods could still collide + fullBaseName := shadowResourcePrefix + podName + "-" + podNamespace + + // Ensure resourceBaseName doesn't exceed 63 characters + if len(resourceBaseName) > 63 { + // Truncate while keeping some of both names + maxPodNameLen := 28 + maxNsLen := 28 + if len(sanitizedPodName) > maxPodNameLen { + sanitizedPodName = uniqueTruncate(sanitizedPodName, maxPodNameLen, fullBaseName) + } + if len(sanitizedNamespace) > maxNsLen { + sanitizedNamespace = sanitizedNamespace[:maxNsLen] + } + resourceBaseName = shadowResourcePrefix + sanitizedPodName + "-" + sanitizedNamespace + resourceBaseName = strings.TrimRight(resourceBaseName, "-") + } + + // Additional check for total length after combining with namespace + ingressFirstLabel := fmt.Sprintf("%s-%s", resourceBaseName, namespace) + if len(ingressFirstLabel) > 63 { + maxNameLen := 63 - len(namespace) - 1 + if maxNameLen > 9 && len(resourceBaseName) > maxNameLen { // >9: room for the 8-char hash suffix + resourceBaseName = uniqueTruncate(resourceBaseName, maxNameLen, fullBaseName) + } + } + + return resourceBaseName, namespace +} + +func computeShadowResourceNames(podName, podNamespace string) (resourceBaseName, shadowNamespace string) { + // Sanitize namespace and pod name for DNS compliance + sanitizedNamespace := sanitizeDNSName(podNamespace) + sanitizedPodName := sanitizeDNSName(podName) + + shadowNamespace = sanitizedNamespace + shadowNamespaceSuffix + // Ensure shadowNamespace is valid (max 63 chars for namespace) + if len(shadowNamespace) > 63 { + shadowNamespace = sanitizedNamespace[:min(63-len(shadowNamespaceSuffix), len(sanitizedNamespace))] + shadowNamespaceSuffix + } + + resourceBaseName = sanitizedPodName + "-" + sanitizedNamespace + fullBaseName := podName + "-" + podNamespace + // Ensure resourceBaseName doesn't exceed 63 characters + if len(resourceBaseName) > 63 { + // Truncate while keeping some of both names + maxPodNameLen := 31 + maxNsLen := 31 + if len(sanitizedPodName) > maxPodNameLen { + sanitizedPodName = uniqueTruncate(sanitizedPodName, maxPodNameLen, fullBaseName) + } + if len(sanitizedNamespace) > maxNsLen { + sanitizedNamespace = sanitizedNamespace[:maxNsLen] + } + resourceBaseName = sanitizedPodName + "-" + sanitizedNamespace + resourceBaseName = strings.TrimRight(resourceBaseName, "-") + } + + ingressFirstLabel := fmt.Sprintf("%s-%s", resourceBaseName, shadowNamespace) + if len(ingressFirstLabel) > 63 { + // If combined length exceeds 63, we need to truncate + // Strategy: keep both parts but truncate proportionally + maxNameLen := 31 + maxNsLen := 31 + + truncatedName := resourceBaseName + if len(truncatedName) > maxNameLen { + truncatedName = uniqueTruncate(truncatedName, maxNameLen, fullBaseName) + } + + truncatedNs := shadowNamespace + if len(truncatedNs) > maxNsLen { + truncatedNs = truncatedNs[:maxNsLen] + truncatedNs = strings.TrimRight(truncatedNs, "-") + } + + resourceBaseName = truncatedName + shadowNamespace = truncatedNs + } + + return resourceBaseName, shadowNamespace +} diff --git a/pkg/virtualkubelet/virtualkubelet.go b/pkg/virtualkubelet/virtualkubelet.go index e55b831b..47d3bc4a 100644 --- a/pkg/virtualkubelet/virtualkubelet.go +++ b/pkg/virtualkubelet/virtualkubelet.go @@ -45,7 +45,7 @@ import ( ) //go:embed templates/wstunnel-template.yaml templates/wstunnel-wireguard-template.yaml -var defaultWstunnelTemplate embed.FS +var defaultShadowTemplates embed.FS //go:embed all:templates/mesh.sh var meshScriptTemplate embed.FS @@ -86,7 +86,7 @@ const ( annMeshNetworkDisabled = "interlink.eu/mesh-network" // set to "disabled" to opt out of mesh networking ) -type WstunnelTemplateData struct { +type ShadowTemplateData struct { Name string Namespace string RandomPassword string @@ -891,16 +891,16 @@ func copyPodLabelsAndAnnotations(pod *v1.Pod) (map[string]string, map[string]str return labels, annotations } -// createDummyPod creates wstunnel infrastructure from template for containers with exposed ports -func (p *Provider) createDummyPod(ctx context.Context, originalPod *v1.Pod) (*v1.Pod, *WstunnelTemplateData, error) { - log.G(ctx).Infof("Creating wstunnel infrastructure for %s/%s with exposed ports", originalPod.Namespace, originalPod.Name) +// createShadowPod creates shadow infrastructure from template for containers with exposed ports +func (p *Provider) createShadowPod(ctx context.Context, originalPod *v1.Pod) (*v1.Pod, *ShadowTemplateData, error) { + log.G(ctx).Infof("Creating shadow infrastructure for %s/%s with exposed ports", originalPod.Namespace, originalPod.Name) - // If not exists, create the namespace for wstunnel + // If not exists, create the namespace for the shadow if originalPod.Namespace == "" { return nil, nil, fmt.Errorf("pod namespace is empty") } - identity, err := computeWstunnelResourceIdentity(originalPod) + identity, err := computeShadowResourceIdentity(originalPod) if err != nil { return nil, nil, err } @@ -912,7 +912,7 @@ func (p *Provider) createDummyPod(ctx context.Context, originalPod *v1.Pod) (*v1 _, err = p.clientSet.CoreV1().Namespaces().Get(ctx, identity.Namespace, metav1.GetOptions{}) if err != nil { if !apierrors.IsNotFound(err) { - return nil, nil, fmt.Errorf("failed to get wstunnel namespace %s: %w", identity.Namespace, err) + return nil, nil, fmt.Errorf("failed to get shadow namespace %s: %w", identity.Namespace, err) } // Create the namespace if it doesn't exist ns := &v1.Namespace{ @@ -922,9 +922,9 @@ func (p *Provider) createDummyPod(ctx context.Context, originalPod *v1.Pod) (*v1 } _, err = p.clientSet.CoreV1().Namespaces().Create(ctx, ns, metav1.CreateOptions{}) if err != nil { - return nil, nil, fmt.Errorf("failed to create wstunnel namespace %s: %w", identity.Namespace, err) + return nil, nil, fmt.Errorf("failed to create shadow namespace %s: %w", identity.Namespace, err) } - log.G(ctx).Infof("Created wstunnel namespace %s", identity.Namespace) + log.G(ctx).Infof("Created shadow namespace %s", identity.Namespace) } // Reuse existing random path prefix if the Deployment already exists, otherwise generate it once @@ -950,7 +950,7 @@ func (p *Provider) createDummyPod(ctx context.Context, originalPod *v1.Pod) (*v1 log.G(ctx).Infof("Copied %d labels and %d annotations from original pod to shadow pod", len(podLabels), len(podAnnotations)) fullMeshEnabledForPod := p.config.Network.FullMesh && !isMeshNetworkingDisabled(originalPod) - templateData := WstunnelTemplateData{ + templateData := ShadowTemplateData{ Name: identity.Name, Namespace: identity.Namespace, RandomPassword: pathPrefix, @@ -980,22 +980,22 @@ func (p *Provider) createDummyPod(ctx context.Context, originalPod *v1.Pod) (*v1 } } - manifestYAML, err := p.executeWstunnelTemplate(ctx, templateData) + manifestYAML, err := p.executeShadowTemplate(ctx, templateData) if err != nil { - return nil, nil, fmt.Errorf("failed to execute wstunnel template: %w", err) + return nil, nil, fmt.Errorf("failed to execute shadow template: %w", err) } - createdPod, err := p.applyWstunnelManifests(ctx, manifestYAML, &templateData) + createdPod, err := p.applyShadowManifests(ctx, manifestYAML, &templateData) if err != nil { - return nil, nil, fmt.Errorf("failed to apply wstunnel manifests: %w", err) + return nil, nil, fmt.Errorf("failed to apply shadow manifests: %w", err) } - log.G(ctx).Infof("Created wstunnel infrastructure for %s/%s", originalPod.Namespace, originalPod.Name) + log.G(ctx).Infof("Created shadow infrastructure for %s/%s", originalPod.Namespace, originalPod.Name) return createdPod, &templateData, nil } // setupWireGuardConfig populates WireGuard-related fields on templateData using annotations from the original pod. -func (p *Provider) setupWireGuardConfig(ctx context.Context, originalPod *v1.Pod, templateData *WstunnelTemplateData) error { +func (p *Provider) setupWireGuardConfig(ctx context.Context, originalPod *v1.Pod, templateData *ShadowTemplateData) error { wgMTU := 1280 if v := strings.TrimSpace(originalPod.Annotations[annWGMTU]); v != "" { if n, err := strconv.Atoi(v); err == nil && n > 0 { @@ -1064,8 +1064,8 @@ func mergeMaps(dst, src map[string]string) map[string]string { return dst } -// executeWstunnelTemplate loads and executes the wstunnel template -func (p *Provider) executeWstunnelTemplate(ctx context.Context, data WstunnelTemplateData) (string, error) { +// executeShadowTemplate loads and executes the shadow template +func (p *Provider) executeShadowTemplate(ctx context.Context, data ShadowTemplateData) (string, error) { var templateContent string // Try to load from custom path first @@ -1084,7 +1084,7 @@ func (p *Provider) executeWstunnelTemplate(ctx context.Context, data WstunnelTem if data.FullMesh { templatePath = "templates/wstunnel-wireguard-template.yaml" } - content, err := defaultWstunnelTemplate.ReadFile(templatePath) + content, err := defaultShadowTemplates.ReadFile(templatePath) if err != nil { return "", fmt.Errorf("failed to read embedded template: %w", err) } @@ -1092,7 +1092,7 @@ func (p *Provider) executeWstunnelTemplate(ctx context.Context, data WstunnelTem } // Parse and execute template - tmpl, err := template.New("wstunnel").Parse(templateContent) + tmpl, err := template.New("shadow").Parse(templateContent) if err != nil { return "", fmt.Errorf("failed to parse template: %w", err) } @@ -1144,7 +1144,7 @@ func prependContainers(dst []v1.Container, add []v1.Container) []v1.Container { } // applyOrUpdateDeployment creates or updates a Deployment, returning its name and namespace. -func (p *Provider) applyOrUpdateDeployment(ctx context.Context, o *appsv1.Deployment, td *WstunnelTemplateData) (string, string, error) { +func (p *Provider) applyOrUpdateDeployment(ctx context.Context, o *appsv1.Deployment, td *ShadowTemplateData) (string, string, error) { if td != nil { ps := &o.Spec.Template.Spec ps.Volumes = mergeVolumes(ps.Volumes, td.Volumes) @@ -1281,8 +1281,8 @@ func (p *Provider) applyOrUpdateSecret(ctx context.Context, o *v1.Secret) error return nil } -// applyWstunnelManifests applies the generated manifests and returns the first created pod -func (p *Provider) applyWstunnelManifests(ctx context.Context, manifestYAML string, td *WstunnelTemplateData) (*v1.Pod, error) { +// applyShadowManifests applies the generated manifests and returns the first created pod +func (p *Provider) applyShadowManifests(ctx context.Context, manifestYAML string, td *ShadowTemplateData) (*v1.Pod, error) { resources := strings.Split(manifestYAML, "---") decoder := serializer.NewCodecFactory(scheme.Scheme).UniversalDeserializer() var deploymentName string @@ -1305,7 +1305,7 @@ func (p *Provider) applyWstunnelManifests(ctx context.Context, manifestYAML stri case *appsv1.Deployment: name, ns, err := p.applyOrUpdateDeployment(ctx, o, td) if err != nil { - p.cleanupPartialWstunnelResources(ctx, createdResources, o.Namespace) + p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace) return nil, err } deploymentName = name @@ -1314,28 +1314,28 @@ func (p *Provider) applyWstunnelManifests(ctx context.Context, manifestYAML stri case *v1.Service: if err := p.applyOrUpdateService(ctx, o); err != nil { - p.cleanupPartialWstunnelResources(ctx, createdResources, o.Namespace) + p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace) return nil, err } createdResources = append(createdResources, "service:"+o.Name) case *networkingv1.Ingress: if err := p.applyOrUpdateIngress(ctx, o); err != nil { - p.cleanupPartialWstunnelResources(ctx, createdResources, o.Namespace) + p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace) return nil, err } createdResources = append(createdResources, "ingress:"+o.Name) case *v1.ConfigMap: if err := p.applyOrUpdateConfigMap(ctx, o); err != nil { - p.cleanupPartialWstunnelResources(ctx, createdResources, o.Namespace) + p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace) return nil, err } createdResources = append(createdResources, "configmap:"+o.Name) case *v1.Secret: if err := p.applyOrUpdateSecret(ctx, o); err != nil { - p.cleanupPartialWstunnelResources(ctx, createdResources, o.Namespace) + p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace) return nil, err } createdResources = append(createdResources, "secret:"+o.Name) @@ -1377,57 +1377,57 @@ func (p *Provider) waitForDeploymentPod(ctx context.Context, deploymentName, nam return nil, fmt.Errorf("no pod found for deployment %s within timeout", deploymentName) } -// cleanupWstunnelResources removes all wstunnel resources for a given name and namespace -func (p *Provider) cleanupWstunnelResources(ctx context.Context, wstunnelName, namespace string) { - log.G(ctx).Infof("Cleaning up wstunnel resources for %s/%s", namespace, wstunnelName) +// cleanupShadowResources removes all shadow resources for a given name and namespace +func (p *Provider) cleanupShadowResources(ctx context.Context, shadowName, namespace string) { + log.G(ctx).Infof("Cleaning up shadow resources for %s/%s", namespace, shadowName) // Delete deployment - err := p.clientSet.AppsV1().Deployments(namespace).Delete(ctx, wstunnelName, metav1.DeleteOptions{}) + err := p.clientSet.AppsV1().Deployments(namespace).Delete(ctx, shadowName, metav1.DeleteOptions{}) if err != nil { - log.G(ctx).Warningf("Failed to delete wstunnel deployment %s/%s: %v", namespace, wstunnelName, err) + log.G(ctx).Warningf("Failed to delete shadow deployment %s/%s: %v", namespace, shadowName, err) } else { - log.G(ctx).Infof("Successfully deleted wstunnel deployment %s/%s", namespace, wstunnelName) + log.G(ctx).Infof("Successfully deleted shadow deployment %s/%s", namespace, shadowName) } // Delete service - err = p.clientSet.CoreV1().Services(namespace).Delete(ctx, wstunnelName, metav1.DeleteOptions{}) + err = p.clientSet.CoreV1().Services(namespace).Delete(ctx, shadowName, metav1.DeleteOptions{}) if err != nil { - log.G(ctx).Warningf("Failed to delete wstunnel service %s/%s: %v", namespace, wstunnelName, err) + log.G(ctx).Warningf("Failed to delete shadow service %s/%s: %v", namespace, shadowName, err) } else { - log.G(ctx).Infof("Successfully deleted wstunnel service %s/%s", namespace, wstunnelName) + log.G(ctx).Infof("Successfully deleted shadow service %s/%s", namespace, shadowName) } // Delete ingress - err = p.clientSet.NetworkingV1().Ingresses(namespace).Delete(ctx, wstunnelName, metav1.DeleteOptions{}) + err = p.clientSet.NetworkingV1().Ingresses(namespace).Delete(ctx, shadowName, metav1.DeleteOptions{}) if err != nil { - log.G(ctx).Warningf("Failed to delete wstunnel ingress %s/%s: %v", namespace, wstunnelName, err) + log.G(ctx).Warningf("Failed to delete shadow ingress %s/%s: %v", namespace, shadowName, err) } else { - log.G(ctx).Infof("Successfully deleted wstunnel ingress %s/%s", namespace, wstunnelName) + log.G(ctx).Infof("Successfully deleted shadow ingress %s/%s", namespace, shadowName) } // Delete configmap - err = p.clientSet.CoreV1().ConfigMaps(namespace).Delete(ctx, wstunnelName+"-wg-config", metav1.DeleteOptions{}) + err = p.clientSet.CoreV1().ConfigMaps(namespace).Delete(ctx, shadowName+"-wg-config", metav1.DeleteOptions{}) if err != nil { - log.G(ctx).Warningf("Failed to delete wstunnel configmap %s/%s: %v", namespace, wstunnelName+"-wg-config", err) + log.G(ctx).Warningf("Failed to delete shadow configmap %s/%s: %v", namespace, shadowName+"-wg-config", err) } else { - log.G(ctx).Infof("Successfully deleted wstunnel configmap %s/%s", namespace, wstunnelName+"-wg-config") + log.G(ctx).Infof("Successfully deleted shadow configmap %s/%s", namespace, shadowName+"-wg-config") } // Delete cert-manager-provisioned TLS secret. if p.config.Network.IngressTLS { - secretName := wstunnelName + "-tls" + secretName := shadowName + "-tls" err = p.clientSet.CoreV1().Secrets(namespace).Delete(ctx, secretName, metav1.DeleteOptions{}) if err != nil { - log.G(ctx).Warningf("Failed to delete wstunnel TLS secret %s/%s: %v", namespace, secretName, err) + log.G(ctx).Warningf("Failed to delete shadow TLS secret %s/%s: %v", namespace, secretName, err) } else { - log.G(ctx).Infof("Successfully deleted wstunnel TLS secret %s/%s", namespace, secretName) + log.G(ctx).Infof("Successfully deleted shadow TLS secret %s/%s", namespace, secretName) } } } -// cleanupPartialWstunnelResources removes specific resources that were created before a failure -func (p *Provider) cleanupPartialWstunnelResources(ctx context.Context, createdResources []string, namespace string) { - log.G(ctx).Infof("Cleaning up partial wstunnel resources in namespace %s", namespace) +// cleanupPartialShadowResources removes specific resources that were created before a failure +func (p *Provider) cleanupPartialShadowResources(ctx context.Context, createdResources []string, namespace string) { + log.G(ctx).Infof("Cleaning up partial shadow resources in namespace %s", namespace) for _, resource := range createdResources { parts := strings.Split(resource, ":") @@ -1602,8 +1602,8 @@ func hasExtraPortsAnnotation(pod *v1.Pod) bool { return exists && strings.TrimSpace(extraPorts) != "" } -// shouldCreateWstunnel checks if wstunnel infrastructure should be created -func (p *Provider) shouldCreateWstunnel(pod *v1.Pod) bool { +// shouldCreateShadow checks if shadow infrastructure should be created +func (p *Provider) shouldCreateShadow(pod *v1.Pod) bool { return p.config.Network.EnableTunnel && (hasExposedPorts(pod) || hasExtraPortsAnnotation(pod)) && pod.Annotations["interlink.eu/pod-vpn"] == "" } @@ -1617,23 +1617,23 @@ func isMeshNetworkingDisabled(pod *v1.Pod) bool { return strings.EqualFold(strings.TrimSpace(pod.Annotations[annMeshNetworkDisabled]), "disabled") } -// handleWstunnelCreation creates wstunnel infrastructure and returns the pod IP -func (p *Provider) handleWstunnelCreation(ctx context.Context, pod *v1.Pod) (string, error) { - identity, err := computeWstunnelResourceIdentity(pod) +// handleShadowCreation creates shadow infrastructure and returns the pod IP +func (p *Provider) handleShadowCreation(ctx context.Context, pod *v1.Pod) (string, error) { + identity, err := computeShadowResourceIdentity(pod) if err != nil { return "", err } - // Create wstunnel infrastructure outside virtual node for port exposure - dummyPod, templateData, err := p.createDummyPod(ctx, pod) + // Create shadow infrastructure outside virtual node for port exposure + shadowPod, templateData, err := p.createShadowPod(ctx, pod) if err != nil { - log.G(ctx).Errorf("Failed to create wstunnel infrastructure for %s/%s: %v", pod.Namespace, pod.Name, err) + log.G(ctx).Errorf("Failed to create shadow infrastructure for %s/%s: %v", pod.Namespace, pod.Name, err) // Clean up any partially created resources - p.cleanupWstunnelResources(ctx, identity.Name, identity.Namespace) - return "", fmt.Errorf("failed to create wstunnel infrastructure for exposed ports: %w", err) + p.cleanupShadowResources(ctx, identity.Name, identity.Namespace) + return "", fmt.Errorf("failed to create shadow infrastructure for exposed ports: %w", err) } - // Wait for wstunnel pod to get an IP with timeout + // Wait for shadow pod to get an IP with timeout timeout := 30 * time.Second // Configurable timeout if timeoutStr := pod.Annotations["interlink.virtual-kubelet.io/wstunnel-timeout"]; timeoutStr != "" { if parsedTimeout, err := time.ParseDuration(timeoutStr); err == nil { @@ -1641,50 +1641,50 @@ func (p *Provider) handleWstunnelCreation(ctx context.Context, pod *v1.Pod) (str } } - podIP, err := p.waitForWstunnelPodIP(ctx, dummyPod, timeout, wstunnelResourceIdentity{ + podIP, err := p.waitForShadowPodIP(ctx, shadowPod, timeout, shadowResourceIdentity{ Name: templateData.Name, Namespace: templateData.Namespace, }) if err != nil { - log.G(ctx).Errorf("Failed to get wstunnel pod IP for %s/%s: %v", pod.Namespace, pod.Name, err) + log.G(ctx).Errorf("Failed to get shadow pod IP for %s/%s: %v", pod.Namespace, pod.Name, err) return "", err } // Add wstunnel client command annotation to the original pod if err := p.addWstunnelClientAnnotation(ctx, pod, templateData); err != nil { log.G(ctx).Warningf("Failed to add wstunnel client annotation to pod %s/%s: %v", pod.Namespace, pod.Name, err) - // Note: We don't clean up here since the wstunnel infrastructure is working, + // Note: We don't clean up here since the shadow infrastructure is working, // just the annotation failed (non-critical) } return podIP, nil } -// waitForWstunnelPodIP waits for wstunnel pod to get an IP -func (p *Provider) waitForWstunnelPodIP(ctx context.Context, dummyPod *v1.Pod, timeout time.Duration, identity wstunnelResourceIdentity) (string, error) { - log.G(ctx).Infof("Waiting up to %v for wstunnel pod %s/%s to get an IP", timeout, dummyPod.Namespace, dummyPod.Name) +// waitForShadowPodIP waits for shadow pod to get an IP +func (p *Provider) waitForShadowPodIP(ctx context.Context, shadowPod *v1.Pod, timeout time.Duration, identity shadowResourceIdentity) (string, error) { + log.G(ctx).Infof("Waiting up to %v for shadow pod %s/%s to get an IP", timeout, shadowPod.Namespace, shadowPod.Name) start := time.Now() for time.Since(start) < timeout { - updatedDummyPod, err := p.clientSet.CoreV1().Pods(dummyPod.Namespace).Get(ctx, dummyPod.Name, metav1.GetOptions{}) + updatedDummyPod, err := p.clientSet.CoreV1().Pods(shadowPod.Namespace).Get(ctx, shadowPod.Name, metav1.GetOptions{}) if err != nil { - log.G(ctx).Warningf("Failed to get wstunnel pod status: %v", err) + log.G(ctx).Warningf("Failed to get shadow pod status: %v", err) time.Sleep(1 * time.Second) continue } if updatedDummyPod.Status.PodIP != "" { podIP := updatedDummyPod.Status.PodIP - log.G(ctx).Infof("Using wstunnel pod IP %s for virtual pod %s/%s", podIP, identity.Namespace, dummyPod.Name) + log.G(ctx).Infof("Using shadow pod IP %s for virtual pod %s/%s", podIP, identity.Namespace, shadowPod.Name) return podIP, nil } time.Sleep(1 * time.Second) } - // Clean up the wstunnel infrastructure since it didn't get an IP - p.cleanupWstunnelResources(ctx, identity.Name, identity.Namespace) - return "", fmt.Errorf("wstunnel pod %s/%s failed to get an IP within %v timeout", dummyPod.Namespace, dummyPod.Name, timeout) + // Clean up the shadow infrastructure since it didn't get an IP + p.cleanupShadowResources(ctx, identity.Name, identity.Namespace) + return "", fmt.Errorf("shadow pod %s/%s failed to get an IP within %v timeout", shadowPod.Namespace, shadowPod.Name, timeout) } // buildTerminatedContainerStatuses builds a slice of ContainerStatus entries where @@ -1830,10 +1830,10 @@ func (p *Provider) CreatePod(ctx context.Context, pod *v1.Pod) error { podIP := "127.0.0.1" - // Handle wstunnel creation if needed - if p.shouldCreateWstunnel(pod) || (p.config.Network.FullMesh && !isMeshNetworkingDisabled(pod)) { + // Handle shadow creation if needed + if p.shouldCreateShadow(pod) || (p.config.Network.FullMesh && !isMeshNetworkingDisabled(pod)) { var err error - podIP, err = p.handleWstunnelCreation(ctx, pod) + podIP, err = p.handleShadowCreation(ctx, pod) if err != nil { return err } @@ -1920,13 +1920,13 @@ func (p *Provider) DeletePod(ctx context.Context, pod *v1.Pod) (err error) { return errdefs.NotFound("pod not found") } - // Clean up wstunnel resources if tunnel is enabled and they exist and no VPN annotation - if p.shouldCreateWstunnel(pod) || (p.config.Network.FullMesh && !isMeshNetworkingDisabled(pod)) { - identity, identityErr := computeWstunnelResourceIdentity(pod) + // Clean up shadow resources if tunnel is enabled and they exist and no VPN annotation + if p.shouldCreateShadow(pod) || (p.config.Network.FullMesh && !isMeshNetworkingDisabled(pod)) { + identity, identityErr := computeShadowResourceIdentity(pod) if identityErr != nil { - log.G(ctx).Warningf("Failed to compute wstunnel resource identity for %s/%s: %v", pod.Namespace, pod.Name, identityErr) + log.G(ctx).Warningf("Failed to compute shadow resource identity for %s/%s: %v", pod.Namespace, pod.Name, identityErr) } else { - p.cleanupWstunnelResources(ctx, identity.Name, identity.Namespace) + p.cleanupShadowResources(ctx, identity.Name, identity.Namespace) } } diff --git a/pkg/virtualkubelet/virtualkubelet_mesh_test.go b/pkg/virtualkubelet/virtualkubelet_mesh_test.go index 1d75be4c..4c612ed6 100644 --- a/pkg/virtualkubelet/virtualkubelet_mesh_test.go +++ b/pkg/virtualkubelet/virtualkubelet_mesh_test.go @@ -61,11 +61,11 @@ func TestIsMeshNetworkingDisabled(t *testing.T) { } } -func TestExecuteWstunnelTemplateIngressTLS(t *testing.T) { +func TestExecuteShadowTemplateIngressTLS(t *testing.T) { p := &Provider{} - manifest, err := p.executeWstunnelTemplate(t.Context(), WstunnelTemplateData{ + manifest, err := p.executeShadowTemplate(t.Context(), ShadowTemplateData{ Name: "pod-default", - Namespace: "default-wstunnel", + Namespace: "default-shadow", RandomPassword: testPathPrefix, WildcardDNS: "tunnel.example.com", IngressTLS: true, @@ -74,17 +74,17 @@ func TestExecuteWstunnelTemplateIngressTLS(t *testing.T) { assert.NoError(t, err) assert.Contains(t, manifest, "cert-manager.io/cluster-issuer: lets-issuer") - assert.Contains(t, manifest, "- pod-default-default-wstunnel.tunnel.example.com") - assert.Contains(t, manifest, "host: pod-default-default-wstunnel.tunnel.example.com") + assert.Contains(t, manifest, "- pod-default-default-shadow.tunnel.example.com") + assert.Contains(t, manifest, "host: pod-default-default-shadow.tunnel.example.com") assert.NotContains(t, manifest, "host: ws-pod-default.tunnel.example.com") assert.Equal(t, 1, strings.Count(manifest, "secretName: pod-default-tls")) } -func TestExecuteWstunnelTemplateFullMeshSelectsWireGuardTemplate(t *testing.T) { +func TestExecuteShadowTemplateFullMeshSelectsWireGuardTemplate(t *testing.T) { p := &Provider{} - manifest, err := p.executeWstunnelTemplate(t.Context(), WstunnelTemplateData{ + manifest, err := p.executeShadowTemplate(t.Context(), ShadowTemplateData{ Name: "pod-default", - Namespace: "default-wstunnel", + Namespace: "default-shadow", RandomPassword: testPathPrefix, WildcardDNS: "tunnel.example.com", FullMesh: true, @@ -98,9 +98,9 @@ func TestExecuteWstunnelTemplateFullMeshSelectsWireGuardTemplate(t *testing.T) { assert.Contains(t, manifest, "number: 28080") } -func TestComputeWstunnelResourceIdentityUsesFinalNamespace(t *testing.T) { +func TestComputeShadowResourceIdentityUsesFinalNamespace(t *testing.T) { t.Run("default shadow namespace", func(t *testing.T) { - identity, err := computeWstunnelResourceIdentity(&v1.Pod{ + identity, err := computeShadowResourceIdentity(&v1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: "my-pod", Namespace: testNamespaceDefault, @@ -109,11 +109,11 @@ func TestComputeWstunnelResourceIdentityUsesFinalNamespace(t *testing.T) { assert.NoError(t, err) assert.Equal(t, "my-pod-default", identity.Name) - assert.Equal(t, "default-wstunnel", identity.Namespace) + assert.Equal(t, "default-shadow", identity.Namespace) }) t.Run("same namespace keeps original namespace", func(t *testing.T) { - identity, err := computeWstunnelResourceIdentity(&v1.Pod{ + identity, err := computeShadowResourceIdentity(&v1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: "my-pod", Namespace: testNamespaceDefault, @@ -124,19 +124,19 @@ func TestComputeWstunnelResourceIdentityUsesFinalNamespace(t *testing.T) { }) assert.NoError(t, err) - assert.Equal(t, "wstunnel-my-pod-default", identity.Name) + assert.Equal(t, "shadow-my-pod-default", identity.Name) assert.Equal(t, testNamespaceDefault, identity.Namespace) }) } -func TestComputeWstunnelResourceIdentitySameNamespaceLongNames(t *testing.T) { +func TestComputeShadowResourceIdentitySameNamespaceLongNames(t *testing.T) { t.Run("long pod name preserves full namespace and stays within 63 chars", func(t *testing.T) { // A real, long-lived namespace that must never be truncated in same-namespace // mode (resources are created in the pod's actual namespace). namespace := strings.Repeat("a", 40) podName := strings.Repeat("b", 80) - identity, err := computeWstunnelResourceIdentity(&v1.Pod{ + identity, err := computeShadowResourceIdentity(&v1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: podName, Namespace: namespace, @@ -160,7 +160,7 @@ func TestComputeWstunnelResourceIdentitySameNamespaceLongNames(t *testing.T) { // label, so the identity must error rather than silently truncate the namespace. namespace := strings.Repeat("a", 62) - identity, err := computeWstunnelResourceIdentity(&v1.Pod{ + identity, err := computeShadowResourceIdentity(&v1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: "web", Namespace: namespace, @@ -188,13 +188,13 @@ func TestGenerateFullMeshScriptIncludesRetryAndReadinessLogic(t *testing.T) { }, } - script, err := p.generateFullMeshScript(t.Context(), &WstunnelTemplateData{ + script, err := p.generateFullMeshScript(t.Context(), &ShadowTemplateData{ RandomPassword: testPathPrefix, WGPrivateKey: serverPriv, ClientPrivateKey: "client-private-key", WGMTU: 1280, KeepaliveSecs: 25, - }, "pod-default-default-wstunnel.tunnel.example.com", "1234567890abcdef") + }, "pod-default-default-shadow.tunnel.example.com", "1234567890abcdef") assert.NoError(t, err) assert.Contains(t, script, "download_with_retry") @@ -202,11 +202,11 @@ func TestGenerateFullMeshScriptIncludesRetryAndReadinessLogic(t *testing.T) { assert.Contains(t, script, "ensure_wstunnel_running") assert.Contains(t, script, "wait_for_wireguard_interface") assert.Contains(t, script, `readiness_protocol="https"`) - assert.Contains(t, script, "$readiness_protocol://pod-default-default-wstunnel.tunnel.example.com:443/path-prefix") - assert.Contains(t, script, "wss://pod-default-default-wstunnel.tunnel.example.com:443") + assert.Contains(t, script, "$readiness_protocol://pod-default-default-shadow.tunnel.example.com:443/path-prefix") + assert.Contains(t, script, "wss://pod-default-default-shadow.tunnel.example.com:443") } -func TestShouldCreateWstunnel(t *testing.T) { +func TestShouldCreateShadow(t *testing.T) { basePod := &v1.Pod{ Spec: v1.PodSpec{ Containers: []v1.Container{ @@ -251,7 +251,7 @@ func TestShouldCreateWstunnel(t *testing.T) { expected: true, }, { - name: "pod vpn annotation disables wstunnel", + name: "pod vpn annotation disables the shadow", network: Network{EnableTunnel: true}, pod: &v1.Pod{ ObjectMeta: metav1.ObjectMeta{ @@ -285,12 +285,12 @@ func TestShouldCreateWstunnel(t *testing.T) { Network: tt.network, }, } - assert.Equal(t, tt.expected, p.shouldCreateWstunnel(tt.pod)) + assert.Equal(t, tt.expected, p.shouldCreateShadow(tt.pod)) }) } } -func TestCleanupWstunnelResources(t *testing.T) { +func TestCleanupShadowResources(t *testing.T) { const ( name = "pod-default" ns = testNamespaceDefault @@ -312,7 +312,7 @@ func TestCleanupWstunnelResources(t *testing.T) { client := newClient() p := &Provider{clientSet: client, config: Config{Network: Network{IngressTLS: true}}} - p.cleanupWstunnelResources(t.Context(), name, ns) + p.cleanupShadowResources(t.Context(), name, ns) _, err := client.AppsV1().Deployments(ns).Get(t.Context(), name, metav1.GetOptions{}) assert.True(t, apierrors.IsNotFound(err), "deployment should be deleted") @@ -330,7 +330,7 @@ func TestCleanupWstunnelResources(t *testing.T) { client := newClient() p := &Provider{clientSet: client, config: Config{Network: Network{IngressTLS: false}}} - p.cleanupWstunnelResources(t.Context(), name, ns) + p.cleanupShadowResources(t.Context(), name, ns) // Core resources are still removed regardless of TLS... _, err := client.AppsV1().Deployments(ns).Get(t.Context(), name, metav1.GetOptions{}) From d4e5c7cb9c5595ffe69630cf8192507a0e17ebb8 Mon Sep 17 00:00:00 2001 From: Matteo Bunino <48362942+matbun@users.noreply.github.com> Date: Wed, 29 Jul 2026 18:36:24 +0200 Subject: [PATCH 2/7] feat: propagate the remote compute node to shadow pods A shadow tunnel needs to know which remote node the workload landed on, but the shadow is rendered at CreatePod time - before the job has been submitted, let alone scheduled. Nothing in the pod spec can carry that value. Plugins now report it through the status API: PodStatus.NodeName // omitempty; empty while the job is queued and the VK materialises it into a per-shadow ConfigMap, -node, created empty alongside the shadow and patched once the plugin reports a node. A shadow mounts that ConfigMap and waits for the key to fill in. A ConfigMap rather than an env var because kubelet refreshes mounted ConfigMaps in place: changing the pod spec would restart the shadow and change its pod IP, which the VK has already reported to Kubernetes as the offloaded pod's IP. The value is blanked when a shadow is created, so a pod recreated under a new UID cannot inherit a node whose job is gone, and repeated reports of an unchanged node are dropped so the status loop does not write on every poll. Plugins that do not set NodeName are unaffected: the field is omitempty, so the wire format is unchanged and shadows that do not consume the ConfigMap never look at it. Signed-off-by: Matteo Bunino <48362942+matbun@users.noreply.github.com> Co-Authored-By: Claude Opus 5 (1M context) --- pkg/interlink/types.go | 6 ++ pkg/virtualkubelet/execute.go | 4 + pkg/virtualkubelet/shadow.go | 100 +++++++++++++++++ pkg/virtualkubelet/shadow_test.go | 156 +++++++++++++++++++++++++++ pkg/virtualkubelet/virtualkubelet.go | 31 +++++- 5 files changed, 295 insertions(+), 2 deletions(-) create mode 100644 pkg/virtualkubelet/shadow_test.go diff --git a/pkg/interlink/types.go b/pkg/interlink/types.go index 5356d7cf..1e63d4a9 100644 --- a/pkg/interlink/types.go +++ b/pkg/interlink/types.go @@ -42,6 +42,12 @@ type PodStatus struct { PodNamespace string `json:"namespace"` // JobID is the remote system's job identifier (e.g., SLURM job ID, container ID) JobID string `json:"JID"` + // NodeName is the remote compute node the pod was allocated on, as the plugin's + // site resolves it (typically a fully qualified hostname reachable from the login + // node). Plugins report it once the workload is actually running; it is empty + // while the job is still queued, and empty for plugins that do not track it. + // interLink uses it to point per-pod shadow tunnels at the right host. + NodeName string `json:"nodeName,omitempty"` // Containers holds the status of all regular containers in the pod Containers []v1.ContainerStatus `json:"containers"` // InitContainers holds the status of all init containers in the pod diff --git a/pkg/virtualkubelet/execute.go b/pkg/virtualkubelet/execute.go index de49f901..1f6f5db1 100644 --- a/pkg/virtualkubelet/execute.go +++ b/pkg/virtualkubelet/execute.go @@ -1531,6 +1531,10 @@ func checkPodsStatus(ctx context.Context, p *Provider, pod *v1.Pod, token string // if the PodUID match with the one in etcd we are talking of the same thing. GOOD if podRemoteStatus.PodUID == string(podRefInCluster.UID) { + // The plugin reports the remote compute node only once the job is actually + // running, so this stays a no-op for as long as the job sits in the queue. + p.publishShadowNodeName(ctx, podRefInCluster, podRemoteStatus.NodeName) + // check if the pod is already in a terminal state (Failed or Succeeded) if currentPhase, terminal := p.podTerminalPhase(podRemoteStatus.PodUID); terminal { if podRefInCluster.Status.Phase == currentPhase { diff --git a/pkg/virtualkubelet/shadow.go b/pkg/virtualkubelet/shadow.go index aa6273f8..e5762497 100644 --- a/pkg/virtualkubelet/shadow.go +++ b/pkg/virtualkubelet/shadow.go @@ -1,12 +1,18 @@ package virtualkubelet import ( + "context" "crypto/sha256" "encoding/hex" + "encoding/json" "fmt" "strings" + "github.com/containerd/containerd/log" v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + k8stypes "k8s.io/apimachinery/pkg/types" ) // shadowResourcePrefix is prepended to shadow resource names created in the @@ -17,6 +23,13 @@ const shadowResourcePrefix = "shadow-" // the dedicated namespace shadow resources live in by default. const shadowNamespaceSuffix = "-shadow" +// shadowNodeConfigMapSuffix names the per-shadow ConfigMap carrying the remote +// compute node the workload was allocated on. +const shadowNodeConfigMapSuffix = "-node" + +// shadowNodeNameKey is the key inside that ConfigMap holding the node name. +const shadowNodeNameKey = "compute-node" + func sanitizeDNSName(name string) string { // Convert to lowercase name = strings.ToLower(name) @@ -226,3 +239,90 @@ func computeShadowResourceNames(podName, podNamespace string) (resourceBaseName, return resourceBaseName, shadowNamespace } + +// hasShadow reports whether a shadow Deployment is rendered for this pod, either +// because the pod exposes ports over a tunnel or because mesh networking wraps +// every offloaded pod. +func (p *Provider) hasShadow(pod *v1.Pod) bool { + return p.shouldCreateShadow(pod) || (p.config.Network.FullMesh && !isMeshNetworkingDisabled(pod)) +} + +// shadowNodeConfigMapName returns the ConfigMap carrying the compute node name +// for the given shadow. +func shadowNodeConfigMapName(shadowName string) string { + return shadowName + shadowNodeConfigMapSuffix +} + +// resetShadowNodeConfigMap creates, or blanks, the per-shadow node ConfigMap. +// The shadow is rendered before the remote batch system has allocated anything, +// so the ConfigMap has to exist - and be mountable - while still empty. Blanking +// an existing one stops a previous run's node from being tunnelled to. +func (p *Provider) resetShadowNodeConfigMap(ctx context.Context, identity shadowResourceIdentity) error { + cm := &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: shadowNodeConfigMapName(identity.Name), + Namespace: identity.Namespace, + }, + Data: map[string]string{shadowNodeNameKey: ""}, + } + + _, err := p.clientSet.CoreV1().ConfigMaps(identity.Namespace).Create(ctx, cm, metav1.CreateOptions{}) + if err == nil { + return nil + } + if !apierrors.IsAlreadyExists(err) { + return fmt.Errorf("failed to create shadow node configmap %s/%s: %w", identity.Namespace, cm.Name, err) + } + if _, err := p.clientSet.CoreV1().ConfigMaps(identity.Namespace).Update(ctx, cm, metav1.UpdateOptions{}); err != nil { + return fmt.Errorf("failed to reset shadow node configmap %s/%s: %w", identity.Namespace, cm.Name, err) + } + return nil +} + +// publishShadowNodeName records the remote compute node a pod was allocated on +// into its shadow's node ConfigMap, so the shadow can direct its tunnel at the +// right host. The ConfigMap is mounted rather than passed as an env var on +// purpose: kubelet refreshes it in place, whereas changing the pod spec would +// restart the shadow and change the pod IP already reported to Kubernetes as the +// offloaded pod's IP. +// +// Repeated calls with an unchanged value are dropped, so the status loop does not +// write on every poll. +func (p *Provider) publishShadowNodeName(ctx context.Context, pod *v1.Pod, nodeName string) { + if nodeName == "" || !p.hasShadow(pod) { + return + } + if last, ok := p.shadowNodeNames.Load(string(pod.UID)); ok && last == nodeName { + return + } + + identity, err := computeShadowResourceIdentity(pod) + if err != nil { + log.G(ctx).Warningf("Failed to compute shadow resource identity for %s/%s: %v", pod.Namespace, pod.Name, err) + return + } + + patch, err := json.Marshal(map[string]map[string]string{"data": {shadowNodeNameKey: nodeName}}) + if err != nil { + log.G(ctx).Warningf("Failed to marshal shadow node patch for %s/%s: %v", pod.Namespace, pod.Name, err) + return + } + + name := shadowNodeConfigMapName(identity.Name) + _, err = p.clientSet.CoreV1().ConfigMaps(identity.Namespace).Patch( + ctx, name, k8stypes.StrategicMergePatchType, patch, metav1.PatchOptions{}, + ) + if err != nil { + log.G(ctx).Warningf("Failed to publish compute node %q to shadow configmap %s/%s: %v", nodeName, identity.Namespace, name, err) + return + } + + p.shadowNodeNames.Store(string(pod.UID), nodeName) + log.G(ctx).Infof("Published compute node %q for shadow %s/%s", nodeName, identity.Namespace, identity.Name) +} + +// forgetShadowNodeName drops the cached node name for a pod that is going away, +// so a pod recreated under a new UID starts from a clean slate. +func (p *Provider) forgetShadowNodeName(pod *v1.Pod) { + p.shadowNodeNames.Delete(string(pod.UID)) +} diff --git a/pkg/virtualkubelet/shadow_test.go b/pkg/virtualkubelet/shadow_test.go new file mode 100644 index 00000000..4cb4dd73 --- /dev/null +++ b/pkg/virtualkubelet/shadow_test.go @@ -0,0 +1,156 @@ +package virtualkubelet + +import ( + "testing" + + "github.com/stretchr/testify/assert" + v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + k8stypes "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" +) + +// podWithPort is an offloaded pod that exposes a port, which is what makes the VK +// render a shadow for it. +func podWithPort(name, namespace string, uid string) *v1.Pod { + return &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace, UID: k8stypes.UID(uid)}, + Spec: v1.PodSpec{ + Containers: []v1.Container{{ + Name: "app", + Ports: []v1.ContainerPort{{ContainerPort: 8888}}, + }}, + }, + } +} + +func tunnelProvider(client *fake.Clientset) *Provider { + return &Provider{clientSet: client, config: Config{Network: Network{EnableTunnel: true}}} +} + +// configMapPatches returns the ConfigMap patch calls recorded by the fake client, +// which is how these tests tell "wrote to the API" from "decided not to". +func configMapPatches(client *fake.Clientset) []k8stesting.Action { + var patches []k8stesting.Action + for _, a := range client.Actions() { + if a.Matches("patch", "configmaps") { + patches = append(patches, a) + } + } + return patches +} + +func TestResetShadowNodeConfigMap(t *testing.T) { + identity := shadowResourceIdentity{Name: "pod-default", Namespace: testNamespaceDefault} + + t.Run("creates the configmap with an empty node so the shadow can mount it before the job is scheduled", func(t *testing.T) { + client := fake.NewSimpleClientset() + p := tunnelProvider(client) + + assert.NoError(t, p.resetShadowNodeConfigMap(t.Context(), identity)) + + cm, err := client.CoreV1().ConfigMaps(identity.Namespace).Get(t.Context(), shadowNodeConfigMapName(identity.Name), metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, "", cm.Data[shadowNodeNameKey]) + }) + + t.Run("blanks a leftover node from a previous run", func(t *testing.T) { + client := fake.NewSimpleClientset(&v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: shadowNodeConfigMapName(identity.Name), Namespace: identity.Namespace}, + Data: map[string]string{shadowNodeNameKey: "stale-node.hpc.example.org"}, + }) + p := tunnelProvider(client) + + assert.NoError(t, p.resetShadowNodeConfigMap(t.Context(), identity)) + + cm, err := client.CoreV1().ConfigMaps(identity.Namespace).Get(t.Context(), shadowNodeConfigMapName(identity.Name), metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, "", cm.Data[shadowNodeNameKey], + "a stale node would make the shadow tunnel to a host whose job is gone") + }) +} + +func TestPublishShadowNodeName(t *testing.T) { + pod := podWithPort("nb", testNamespaceDefault, "uid-1") + identity, err := computeShadowResourceIdentity(pod) + assert.NoError(t, err) + cmName := shadowNodeConfigMapName(identity.Name) + + newClient := func() *fake.Clientset { + return fake.NewSimpleClientset(&v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: cmName, Namespace: identity.Namespace}, + Data: map[string]string{shadowNodeNameKey: ""}, + }) + } + + t.Run("writes the node reported by the plugin", func(t *testing.T) { + client := newClient() + p := tunnelProvider(client) + + p.publishShadowNodeName(t.Context(), pod, "node042.hpc.example.org") + + cm, err := client.CoreV1().ConfigMaps(identity.Namespace).Get(t.Context(), cmName, metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, "node042.hpc.example.org", cm.Data[shadowNodeNameKey]) + }) + + t.Run("an empty node name leaves the configmap alone while the job is queued", func(t *testing.T) { + client := newClient() + p := tunnelProvider(client) + + p.publishShadowNodeName(t.Context(), pod, "") + + assert.Empty(t, configMapPatches(client), "waiting for an allocation must not write to the API") + }) + + t.Run("repeated reports of the same node do not write again", func(t *testing.T) { + client := newClient() + p := tunnelProvider(client) + + p.publishShadowNodeName(t.Context(), pod, "node042.hpc.example.org") + before := len(configMapPatches(client)) + for range 5 { + p.publishShadowNodeName(t.Context(), pod, "node042.hpc.example.org") + } + + assert.Len(t, configMapPatches(client), before, + "the status loop polls continuously; only changes should reach the API") + }) + + t.Run("a pod with no shadow is skipped", func(t *testing.T) { + client := newClient() + p := &Provider{clientSet: client, config: Config{Network: Network{EnableTunnel: false}}} + + p.publishShadowNodeName(t.Context(), pod, "node042.hpc.example.org") + + assert.Empty(t, configMapPatches(client)) + }) + + t.Run("forgetting a pod lets the same node be published again", func(t *testing.T) { + client := newClient() + p := tunnelProvider(client) + + p.publishShadowNodeName(t.Context(), pod, "node042.hpc.example.org") + p.forgetShadowNodeName(pod) + p.publishShadowNodeName(t.Context(), pod, "node042.hpc.example.org") + + assert.Len(t, configMapPatches(client), 2) + }) +} + +func TestCleanupShadowResourcesRemovesNodeConfigMap(t *testing.T) { + const name = "pod-default" + ns := testNamespaceDefault + + client := fake.NewSimpleClientset(&v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: shadowNodeConfigMapName(name), Namespace: ns}, + }) + p := tunnelProvider(client) + + p.cleanupShadowResources(t.Context(), name, ns) + + _, err := client.CoreV1().ConfigMaps(ns).Get(t.Context(), shadowNodeConfigMapName(name), metav1.GetOptions{}) + assert.True(t, apierrors.IsNotFound(err), "the node configmap should be deleted with the rest of the shadow") +} diff --git a/pkg/virtualkubelet/virtualkubelet.go b/pkg/virtualkubelet/virtualkubelet.go index 47d3bc4a..d96668cf 100644 --- a/pkg/virtualkubelet/virtualkubelet.go +++ b/pkg/virtualkubelet/virtualkubelet.go @@ -105,6 +105,12 @@ type ShadowTemplateData struct { IngressTLS bool IngressClusterIssuer string FullMesh bool + // NodeConfigMap is the ConfigMap the shadow can mount to learn which remote + // compute node the workload landed on. It exists from the moment the shadow is + // created but holds an empty NodeNameKey until the plugin reports the node. + NodeConfigMap string + // NodeNameKey is the key inside NodeConfigMap holding the node name. + NodeNameKey string } type PortMapping struct { @@ -147,6 +153,9 @@ type Provider struct { clientSet kubernetes.Interface clientHTTPTransport *http.Transport podIPs []string + // shadowNodeNames caches, per pod UID, the last compute node published to + // that pod's shadow, so the status loop only writes on change. + shadowNodeNames sync.Map } // Increment the given IP address @@ -943,6 +952,12 @@ func (p *Provider) createShadowPod(ctx context.Context, originalPod *v1.Pod) (*v // log the path prefix log.G(ctx).Infof("Using wstunnel path prefix %s for %s/%s", pathPrefix, originalPod.Namespace, originalPod.Name) + // The remote node is unknown at this point - the job has not been submitted, let + // alone scheduled - so publish an empty ConfigMap the shadow can already mount. + if err := p.resetShadowNodeConfigMap(ctx, identity); err != nil { + return nil, nil, err + } + localContainers := getLocalContainers(originalPod) localInitContainers := getLocalInitContainers(originalPod) @@ -964,6 +979,8 @@ func (p *Provider) createShadowPod(ctx context.Context, originalPod *v1.Pod) (*v IngressTLS: p.config.Network.IngressTLS, IngressClusterIssuer: p.config.Network.IngressClusterIssuer, FullMesh: fullMeshEnabledForPod, + NodeConfigMap: shadowNodeConfigMapName(identity.Name), + NodeNameKey: shadowNodeNameKey, } log.G(ctx).Debugf("LocalInitContainers count: %d", len(templateData.LocalInitContainers)) @@ -1413,6 +1430,15 @@ func (p *Provider) cleanupShadowResources(ctx context.Context, shadowName, names log.G(ctx).Infof("Successfully deleted shadow configmap %s/%s", namespace, shadowName+"-wg-config") } + // Delete the compute node configmap + nodeCM := shadowNodeConfigMapName(shadowName) + err = p.clientSet.CoreV1().ConfigMaps(namespace).Delete(ctx, nodeCM, metav1.DeleteOptions{}) + if err != nil { + log.G(ctx).Warningf("Failed to delete shadow configmap %s/%s: %v", namespace, nodeCM, err) + } else { + log.G(ctx).Infof("Successfully deleted shadow configmap %s/%s", namespace, nodeCM) + } + // Delete cert-manager-provisioned TLS secret. if p.config.Network.IngressTLS { secretName := shadowName + "-tls" @@ -1831,7 +1857,7 @@ func (p *Provider) CreatePod(ctx context.Context, pod *v1.Pod) error { podIP := "127.0.0.1" // Handle shadow creation if needed - if p.shouldCreateShadow(pod) || (p.config.Network.FullMesh && !isMeshNetworkingDisabled(pod)) { + if p.hasShadow(pod) { var err error podIP, err = p.handleShadowCreation(ctx, pod) if err != nil { @@ -1921,13 +1947,14 @@ func (p *Provider) DeletePod(ctx context.Context, pod *v1.Pod) (err error) { } // Clean up shadow resources if tunnel is enabled and they exist and no VPN annotation - if p.shouldCreateShadow(pod) || (p.config.Network.FullMesh && !isMeshNetworkingDisabled(pod)) { + if p.hasShadow(pod) { identity, identityErr := computeShadowResourceIdentity(pod) if identityErr != nil { log.G(ctx).Warningf("Failed to compute shadow resource identity for %s/%s: %v", pod.Namespace, pod.Name, identityErr) } else { p.cleanupShadowResources(ctx, identity.Name, identity.Namespace) } + p.forgetShadowNodeName(pod) } now := metav1.Now() From c297e8b8ad2a275c963d2d1adaa24d15d35d8c80 Mon Sep 17 00:00:00 2001 From: Matteo Bunino <48362942+matbun@users.noreply.github.com> Date: Wed, 29 Jul 2026 18:43:02 +0200 Subject: [PATCH 3/7] feat: add an SSH port-forward shadow for air-gapped HPC The wstunnel shadow assumes two things air-gapped HPC sites do not have: outbound internet access from the compute nodes, and a publicly reachable ingress on the Kubernetes cluster. The workload dials out; nothing works if it cannot. This inverts the direction. The shadow dials in to the site's SSH login node and runs `ssh -N -L ::`, one -L per exposed port, so cluster traffic reaches services inside an offloaded pod - notebooks, dashboards, web UIs - with the login node as the only required route. The offloaded pod runs nothing on its side: no wstunnel client, no WireGuard config, no pre-exec injection at all. Network: EnableTunnel: true ShadowMode: ssh SSH: LoginHost: login.hpc.example.org User: alice KeySecret: hpc-ssh-key # or Kerberos: KeytabSecret + Principal Selected by Network.ShadowMode, which defaults to "wstunnel", so existing deployments are untouched. Both authentication methods sites actually use are supported: an SSH private key, or a Kerberos keytab with kinit run in an init container and refreshed by a sidecar for the life of the tunnel. The shadow learns its target from the per-pod node ConfigMap: an init container waits on it, so a pod whose job is still queued shows Init rather than a crash-looping tunnel, and the wait is bounded by SSH.NodeWaitTimeout (default 2h) with an error naming the actual cause. Credentials are replicated into the shadow's namespace by default, since shadows follow the offloaded pod into arbitrary per-user namespaces and would otherwise have to be seeded by hand everywhere. Set SSH.ReplicateCredentials: false to manage them yourself. ShadowMode: ssh with FullMesh is rejected at startup: the mesh needs the compute node to dial the cluster, which is exactly what these sites cannot do. Mesh over SSH is tracked in interlink-hq/interLink#548. UDP ports are skipped with a warning, since ssh -L is TCP only. The shadow runs docker/Dockerfile.ssh-tunnel, a stock ssh + krb5 client image published alongside the other release images. It contains no interLink code; SSH.Image overrides it. Signed-off-by: Matteo Bunino <48362942+matbun@users.noreply.github.com> Co-Authored-By: Claude Opus 5 (1M context) --- .github/workflows/build_images.yaml | 12 +- docker/Dockerfile.ssh-tunnel | 14 + pkg/virtualkubelet/config.go | 73 ++++ pkg/virtualkubelet/shadow_ssh.go | 219 ++++++++++++ pkg/virtualkubelet/shadow_ssh_test.go | 337 ++++++++++++++++++ .../templates/shadow-ssh-template.yaml | 248 +++++++++++++ pkg/virtualkubelet/virtualkubelet.go | 42 ++- 7 files changed, 937 insertions(+), 8 deletions(-) create mode 100644 docker/Dockerfile.ssh-tunnel create mode 100644 pkg/virtualkubelet/shadow_ssh.go create mode 100644 pkg/virtualkubelet/shadow_ssh_test.go create mode 100644 pkg/virtualkubelet/templates/shadow-ssh-template.yaml diff --git a/.github/workflows/build_images.yaml b/.github/workflows/build_images.yaml index 7e164188..fd32015d 100644 --- a/.github/workflows/build_images.yaml +++ b/.github/workflows/build_images.yaml @@ -71,7 +71,17 @@ jobs: ghcr.io/${{ steps.get_repo_owner.outputs.repo_owner }}/interlink/interlink:latest file: ./docker/Dockerfile.interlink platforms: linux/amd64, linux/arm64, linux/aarch64 - + - name: Build container base image ssh-tunnel + uses: docker/build-push-action@v6 + with: + context: ./ + outputs: "type=registry,push=true" + tags: | + ghcr.io/${{ steps.get_repo_owner.outputs.repo_owner }}/interlink/ssh-tunnel:${{ env.RELEASE_VERSION }} + ghcr.io/${{ steps.get_repo_owner.outputs.repo_owner }}/interlink/ssh-tunnel:latest + file: ./docker/Dockerfile.ssh-tunnel + platforms: linux/amd64, linux/arm64, linux/aarch64 + virtual-kubelet-refresh-token: runs-on: ubuntu-latest #env: diff --git a/docker/Dockerfile.ssh-tunnel b/docker/Dockerfile.ssh-tunnel new file mode 100644 index 00000000..9a93fb96 --- /dev/null +++ b/docker/Dockerfile.ssh-tunnel @@ -0,0 +1,14 @@ +# Image for the SSH port-forward shadow pod (Network.ShadowMode: ssh). +# +# It runs `ssh -N -L ...` against an HPC login node, so it needs an ssh client and, +# for GSSAPI sites, the Kerberos client tools to obtain and renew a ticket from a +# keytab. Nothing from this repository is installed: the shadow runs no interLink +# binary, only stock ssh. +FROM debian:bookworm-slim + +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + openssh-client krb5-user ca-certificates \ + && rm -rf /var/lib/apt/lists/* + +CMD ["/bin/sh"] diff --git a/pkg/virtualkubelet/config.go b/pkg/virtualkubelet/config.go index db867798..45be7a37 100644 --- a/pkg/virtualkubelet/config.go +++ b/pkg/virtualkubelet/config.go @@ -173,4 +173,77 @@ type Network struct { Slirp4netnsURL string `yaml:"Slirp4netnsURL,omitempty"` // UnsharedMode is the flag for unshared network mode in slirp4netns UnshareMode string `yaml:"UnshareMode,omitempty"` + // ShadowMode selects which shadow implementation is rendered for offloaded pods + // with exposed ports: "wstunnel" (default) or "ssh". + ShadowMode string `yaml:"ShadowMode,omitempty"` + // SSH configures the SSH port-forward shadow, used when ShadowMode is "ssh" + SSH SSHTunnel `yaml:"SSH,omitempty"` +} + +// Shadow implementations selectable through Network.ShadowMode. +const ( + // ShadowModeWstunnel exposes the offloaded pod's ports by having the workload + // dial out to a public ingress and run a wstunnel client. This is the default. + ShadowModeWstunnel = "wstunnel" + // ShadowModeSSH exposes them the other way round: the shadow dials in to an SSH + // login node and forwards each port to the compute node the job landed on. The + // workload runs nothing, and no compute node needs outbound internet access. + ShadowModeSSH = "ssh" +) + +// SSH authentication methods selectable through SSHTunnel.Auth. +const ( + // SSHAuthPublicKey authenticates with a private key from KeySecret. + SSHAuthPublicKey = "publickey" + // SSHAuthKerberos authenticates with GSSAPI, using a keytab from KeytabSecret. + SSHAuthKerberos = "kerberos" +) + +// SSHTunnel configures the SSH port-forward shadow. +// +// The shadow runs `ssh -N -L ::` against the site's login +// node, one -L per exposed port, so cluster traffic reaches services inside an +// offloaded pod without the compute node needing any outbound connectivity. It +// covers the same ground as the wstunnel shadow, in the opposite direction; it does +// not give the offloaded pod access back into the cluster (see Network.FullMesh). +type SSHTunnel struct { + // LoginHost is the SSH login node to forward through (required) + LoginHost string `yaml:"LoginHost,omitempty"` + // Port is the login node's SSH port (default 22) + Port int `yaml:"Port,omitempty"` + // User is the login name on the login node (required) + User string `yaml:"User,omitempty"` + // Image is the container image running in the shadow. It must provide an ssh + // client, and kinit/klist when Auth is "kerberos". + Image string `yaml:"Image,omitempty"` + // Auth selects the authentication method: "publickey" (default) or "kerberos" + Auth string `yaml:"Auth,omitempty"` + // KeySecret is the Secret holding the SSH private key ("publickey" auth) + KeySecret string `yaml:"KeySecret,omitempty"` + // KeySecretKey is the key inside KeySecret holding the private key (default "id_ed25519") + KeySecretKey string `yaml:"KeySecretKey,omitempty"` + // KeytabSecret is the Secret holding the Kerberos keytab ("kerberos" auth) + KeytabSecret string `yaml:"KeytabSecret,omitempty"` + // KeytabSecretKey is the key inside KeytabSecret holding the keytab (default "user.keytab") + KeytabSecretKey string `yaml:"KeytabSecretKey,omitempty"` + // Principal is the Kerberos principal to obtain a ticket for ("kerberos" auth) + Principal string `yaml:"Principal,omitempty"` + // Krb5ConfigMap is an optional ConfigMap holding a krb5.conf to mount at /etc/krb5.conf + Krb5ConfigMap string `yaml:"Krb5ConfigMap,omitempty"` + // KnownHostsConfigMap is an optional ConfigMap holding a known_hosts file. When + // unset the shadow falls back to StrictHostKeyChecking=accept-new, which trusts + // whatever key the login node presents on first contact. + KnownHostsConfigMap string `yaml:"KnownHostsConfigMap,omitempty"` + // ReplicateCredentials copies the referenced Secret and ConfigMaps from the + // virtual kubelet's own namespace into the shadow's namespace, so offloaded pods + // in arbitrary (e.g. per-user) namespaces work without pre-seeding credentials + // everywhere. Defaults to true. Note this makes the credential readable by anyone + // who can read Secrets in those namespaces. + ReplicateCredentials *bool `yaml:"ReplicateCredentials,omitempty"` + // NodeWaitTimeout bounds how long the shadow waits for the plugin to report the + // compute node before failing (default "2h"). Queue waits are normal, so this is + // generous by design. + NodeWaitTimeout string `yaml:"NodeWaitTimeout,omitempty"` + // ExtraOptions are additional ssh client options, each passed verbatim as -o + ExtraOptions []string `yaml:"ExtraOptions,omitempty"` } diff --git a/pkg/virtualkubelet/shadow_ssh.go b/pkg/virtualkubelet/shadow_ssh.go new file mode 100644 index 00000000..524386ab --- /dev/null +++ b/pkg/virtualkubelet/shadow_ssh.go @@ -0,0 +1,219 @@ +package virtualkubelet + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/containerd/containerd/log" + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +const ( + // DefaultSSHPort is the login node port used when none is configured. + DefaultSSHPort = 22 + // DefaultSSHKeySecretKey is the key holding the private key inside KeySecret. + DefaultSSHKeySecretKey = "id_ed25519" + // DefaultSSHKeytabSecretKey is the key holding the keytab inside KeytabSecret. + DefaultSSHKeytabSecretKey = "user.keytab" + // DefaultSSHNodeWaitTimeout bounds the wait for the plugin to report a compute + // node. Batch queues routinely make a pod wait hours before it starts. + DefaultSSHNodeWaitTimeout = "2h" +) + +// defaultSSHNodeWait is DefaultSSHNodeWaitTimeout as a duration, so falling back to +// it needs no parsing. TestDefaultSSHNodeWaitMatchesTimeout keeps the two in step. +const defaultSSHNodeWait = 2 * time.Hour + +// defaultSSHTunnelImage returns the image the SSH shadow runs, pinned to this +// virtual kubelet's own version so the two are released together. +func defaultSSHTunnelImage() string { + return "ghcr.io/interlink-hq/interlink/ssh-tunnel:" + KubeletVersion +} + +// isSSHShadow reports whether shadows are rendered as SSH port-forwards. +func (p *Provider) isSSHShadow() bool { + return p.config.Network.ShadowMode == ShadowModeSSH +} + +// NormalizeShadowConfig fills in shadow defaults and rejects combinations that +// cannot work, so a misconfigured deployment fails at startup rather than when the +// first pod with an exposed port shows up. +func NormalizeShadowConfig(config *Config) error { + n := &config.Network + + n.ShadowMode = strings.ToLower(strings.TrimSpace(n.ShadowMode)) + switch n.ShadowMode { + case "": + n.ShadowMode = ShadowModeWstunnel + case ShadowModeWstunnel, ShadowModeSSH: + default: + return fmt.Errorf("unknown Network.ShadowMode %q: expected %q or %q", n.ShadowMode, ShadowModeWstunnel, ShadowModeSSH) + } + + if n.ShadowMode != ShadowModeSSH { + return nil + } + + // Mesh needs the compute node to dial out to the cluster, which is exactly what + // an SSH-only site cannot do. Combining the two is tracked separately. + if n.FullMesh { + return fmt.Errorf("Network.FullMesh is not supported with ShadowMode %q: the SSH shadow only exposes the offloaded pod's ports to the cluster, it does not give the pod access back into it (see interlink-hq/interLink#548)", ShadowModeSSH) + } + + s := &n.SSH + if strings.TrimSpace(s.LoginHost) == "" { + return fmt.Errorf("Network.SSH.LoginHost is required with ShadowMode %q", ShadowModeSSH) + } + if strings.TrimSpace(s.User) == "" { + return fmt.Errorf("Network.SSH.User is required with ShadowMode %q", ShadowModeSSH) + } + + if s.Port == 0 { + s.Port = DefaultSSHPort + } + if s.Image == "" { + s.Image = defaultSSHTunnelImage() + } + if s.NodeWaitTimeout == "" { + s.NodeWaitTimeout = DefaultSSHNodeWaitTimeout + } + if _, err := time.ParseDuration(s.NodeWaitTimeout); err != nil { + return fmt.Errorf("invalid Network.SSH.NodeWaitTimeout %q: %w", s.NodeWaitTimeout, err) + } + if s.ReplicateCredentials == nil { + replicate := true + s.ReplicateCredentials = &replicate + } + + s.Auth = strings.ToLower(strings.TrimSpace(s.Auth)) + switch s.Auth { + case "": + s.Auth = SSHAuthPublicKey + fallthrough + case SSHAuthPublicKey: + if strings.TrimSpace(s.KeySecret) == "" { + return fmt.Errorf("Network.SSH.KeySecret is required with Auth %q", SSHAuthPublicKey) + } + if s.KeySecretKey == "" { + s.KeySecretKey = DefaultSSHKeySecretKey + } + case SSHAuthKerberos: + if strings.TrimSpace(s.KeytabSecret) == "" { + return fmt.Errorf("Network.SSH.KeytabSecret is required with Auth %q", SSHAuthKerberos) + } + if strings.TrimSpace(s.Principal) == "" { + return fmt.Errorf("Network.SSH.Principal is required with Auth %q", SSHAuthKerberos) + } + if s.KeytabSecretKey == "" { + s.KeytabSecretKey = DefaultSSHKeytabSecretKey + } + default: + return fmt.Errorf("unknown Network.SSH.Auth %q: expected %q or %q", s.Auth, SSHAuthPublicKey, SSHAuthKerberos) + } + + return nil +} + +// sshNodeWaitSeconds converts the configured wait into whole seconds for the shell +// loop in the template. NormalizeShadowConfig has already validated the duration, +// so the fallback only guards against a Provider built without it. +func (p *Provider) sshNodeWaitSeconds() int { + d, err := time.ParseDuration(p.config.Network.SSH.NodeWaitTimeout) + if err != nil { + d = defaultSSHNodeWait + } + return int(d.Seconds()) +} + +// sshCredentialSecret returns the Secret name the configured auth method needs. +func sshCredentialSecret(s SSHTunnel) string { + if s.Auth == SSHAuthKerberos { + return s.KeytabSecret + } + return s.KeySecret +} + +// replicateShadowCredentials copies the SSH credential Secret and any supporting +// ConfigMaps from the virtual kubelet's own namespace into the shadow's namespace. +// +// Shadows follow the offloaded pod, which for multi-tenant setups means arbitrary +// per-user namespaces; without this every one of them would have to be seeded with +// the credential by hand before offloading could work. +func (p *Provider) replicateShadowCredentials(ctx context.Context, identity shadowResourceIdentity) error { + s := p.config.Network.SSH + if s.ReplicateCredentials == nil || !*s.ReplicateCredentials { + return nil + } + + source := p.config.Namespace + if source == "" || source == identity.Namespace { + return nil + } + + if name := sshCredentialSecret(s); name != "" { + if err := p.replicateSecret(ctx, source, identity.Namespace, name); err != nil { + return err + } + } + for _, name := range []string{s.Krb5ConfigMap, s.KnownHostsConfigMap} { + if name == "" { + continue + } + if err := p.replicateConfigMap(ctx, source, identity.Namespace, name); err != nil { + return err + } + } + return nil +} + +func (p *Provider) replicateSecret(ctx context.Context, source, target, name string) error { + src, err := p.clientSet.CoreV1().Secrets(source).Get(ctx, name, metav1.GetOptions{}) + if err != nil { + return fmt.Errorf("failed to read SSH credential secret %s/%s: %w", source, name, err) + } + copied := &v1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: target}, + Type: src.Type, + Data: src.Data, + } + if err := p.applyOrUpdateSecret(ctx, copied); err != nil { + return fmt.Errorf("failed to replicate secret %s into %s: %w", name, target, err) + } + log.G(ctx).Infof("Replicated SSH credential secret %s from %s to %s", name, source, target) + return nil +} + +func (p *Provider) replicateConfigMap(ctx context.Context, source, target, name string) error { + src, err := p.clientSet.CoreV1().ConfigMaps(source).Get(ctx, name, metav1.GetOptions{}) + if err != nil { + return fmt.Errorf("failed to read configmap %s/%s: %w", source, name, err) + } + copied := &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: target}, + Data: src.Data, + BinaryData: src.BinaryData, + } + if err := p.applyOrUpdateConfigMap(ctx, copied); err != nil { + return fmt.Errorf("failed to replicate configmap %s into %s: %w", name, target, err) + } + log.G(ctx).Infof("Replicated configmap %s from %s to %s", name, source, target) + return nil +} + +// warnOnUDPPorts reports ports the SSH shadow cannot carry. ssh -L forwards TCP +// only, so a UDP port is silently unreachable; say so rather than let it look like +// a broken tunnel. +func warnOnUDPPorts(ctx context.Context, ports []PortMapping) { + var udp []string + for _, port := range ports { + if strings.EqualFold(port.Protocol, "UDP") { + udp = append(udp, fmt.Sprintf("%d", port.Port)) + } + } + if len(udp) > 0 { + log.G(ctx).Warningf("SSH shadow cannot forward UDP ports %s: ssh -L is TCP only", strings.Join(udp, ", ")) + } +} diff --git a/pkg/virtualkubelet/shadow_ssh_test.go b/pkg/virtualkubelet/shadow_ssh_test.go new file mode 100644 index 00000000..9e59c062 --- /dev/null +++ b/pkg/virtualkubelet/shadow_ssh_test.go @@ -0,0 +1,337 @@ +package virtualkubelet + +import ( + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + appsv1 "k8s.io/api/apps/v1" + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/serializer" + "k8s.io/client-go/kubernetes/fake" + "k8s.io/client-go/kubernetes/scheme" +) + +const ( + testKeytabSecret = "hpc-keytab" + testPrincipal = "alice@EXAMPLE.ORG" +) + +func sshConfig() Config { + return Config{ + Namespace: "interlink", + Network: Network{ + EnableTunnel: true, + ShadowMode: ShadowModeSSH, + SSH: SSHTunnel{ + LoginHost: "login.hpc.example.org", + User: "alice", + KeySecret: "hpc-ssh-key", + }, + }, + } +} + +func normalized(t *testing.T, config Config) Config { + t.Helper() + assert.NoError(t, NormalizeShadowConfig(&config)) + return config +} + +func TestNormalizeShadowConfigDefaults(t *testing.T) { + t.Run("defaults to the wstunnel shadow", func(t *testing.T) { + config := Config{} + assert.NoError(t, NormalizeShadowConfig(&config)) + assert.Equal(t, ShadowModeWstunnel, config.Network.ShadowMode) + }) + + t.Run("fills in ssh defaults", func(t *testing.T) { + config := normalized(t, sshConfig()) + s := config.Network.SSH + + assert.Equal(t, DefaultSSHPort, s.Port) + assert.Equal(t, SSHAuthPublicKey, s.Auth) + assert.Equal(t, DefaultSSHKeySecretKey, s.KeySecretKey) + assert.Equal(t, DefaultSSHNodeWaitTimeout, s.NodeWaitTimeout) + assert.Contains(t, s.Image, "ssh-tunnel") + assert.NotNil(t, s.ReplicateCredentials) + assert.True(t, *s.ReplicateCredentials, "credentials replicate by default so per-user namespaces work") + }) + + t.Run("kerberos defaults", func(t *testing.T) { + config := sshConfig() + config.Network.SSH.KeySecret = "" + config.Network.SSH.Auth = "KERBEROS" + config.Network.SSH.KeytabSecret = testKeytabSecret + config.Network.SSH.Principal = testPrincipal + + config = normalized(t, config) + assert.Equal(t, SSHAuthKerberos, config.Network.SSH.Auth) + assert.Equal(t, DefaultSSHKeytabSecretKey, config.Network.SSH.KeytabSecretKey) + }) +} + +func TestNormalizeShadowConfigRejects(t *testing.T) { + tests := []struct { + name string + mutate func(*Config) + wantErr string + }{ + { + name: "unknown shadow mode", + mutate: func(c *Config) { c.Network.ShadowMode = "carrier-pigeon" }, + wantErr: "unknown Network.ShadowMode", + }, + { + // The mesh needs the compute node to dial the cluster, which is precisely + // what an ssh-only site cannot do. + name: "ssh together with full mesh", + mutate: func(c *Config) { c.Network.FullMesh = true }, + wantErr: "FullMesh is not supported", + }, + { + name: "missing login host", + mutate: func(c *Config) { c.Network.SSH.LoginHost = "" }, + wantErr: "LoginHost is required", + }, + { + name: "missing user", + mutate: func(c *Config) { c.Network.SSH.User = "" }, + wantErr: "User is required", + }, + { + name: "public key auth without a key secret", + mutate: func(c *Config) { c.Network.SSH.KeySecret = "" }, + wantErr: "KeySecret is required", + }, + { + name: "kerberos without a keytab", + mutate: func(c *Config) { + c.Network.SSH.Auth = SSHAuthKerberos + c.Network.SSH.Principal = testPrincipal + }, + wantErr: "KeytabSecret is required", + }, + { + name: "kerberos without a principal", + mutate: func(c *Config) { + c.Network.SSH.Auth = SSHAuthKerberos + c.Network.SSH.KeytabSecret = testKeytabSecret + }, + wantErr: "Principal is required", + }, + { + name: "unknown auth method", + mutate: func(c *Config) { c.Network.SSH.Auth = "password" }, + wantErr: "unknown Network.SSH.Auth", + }, + { + name: "unparseable node wait timeout", + mutate: func(c *Config) { c.Network.SSH.NodeWaitTimeout = "forever" }, + wantErr: "invalid Network.SSH.NodeWaitTimeout", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + config := sshConfig() + tt.mutate(&config) + + err := NormalizeShadowConfig(&config) + + assert.Error(t, err, "a misconfiguration must fail at startup, not at first offload") + assert.Contains(t, err.Error(), tt.wantErr) + }) + } +} + +// renderSSHShadow renders the ssh template and returns the manifest plus the +// Deployment decoded from it, using the same decoder applyShadowManifests uses. +func renderSSHShadow(t *testing.T, config Config, ports []PortMapping) (string, *appsv1.Deployment) { + t.Helper() + config = normalized(t, config) + p := &Provider{config: config} + + manifest, err := p.executeShadowTemplate(t.Context(), ShadowTemplateData{ + Name: "shadow-nb-user", + Namespace: "user", + ExposedPorts: ports, + NodeConfigMap: "shadow-nb-user-node", + NodeNameKey: shadowNodeNameKey, + SSH: config.Network.SSH, + SSHNodeWaitSeconds: p.sshNodeWaitSeconds(), + }) + assert.NoError(t, err) + + decoder := serializer.NewCodecFactory(scheme.Scheme).UniversalDeserializer() + var deployment *appsv1.Deployment + for _, doc := range strings.Split(manifest, "---") { + if strings.TrimSpace(doc) == "" { + continue + } + obj, _, err := decoder.Decode([]byte(doc), nil, nil) + assert.NoError(t, err, "every rendered document must decode:\n%s", doc) + if d, ok := obj.(*appsv1.Deployment); ok { + deployment = d + } + } + assert.NotNil(t, deployment, "the ssh template must render a Deployment") + return manifest, deployment +} + +func TestSSHShadowTemplate(t *testing.T) { + tcp := []PortMapping{{Port: 8888, Name: "notebook", Protocol: "TCP"}} + + t.Run("forwards each exposed port to the node the plugin reported", func(t *testing.T) { + manifest, deployment := renderSSHShadow(t, sshConfig(), tcp) + + assert.Contains(t, manifest, "-L 0.0.0.0:8888:$node:8888") + assert.Contains(t, manifest, "alice@login.hpc.example.org") + assert.Contains(t, manifest, "-i /interlink/ssh/"+DefaultSSHKeySecretKey) + // The node arrives through a mounted ConfigMap, not the pod spec: restarting + // the shadow would change the pod IP already reported for the offloaded pod. + assert.Contains(t, manifest, "shadow-nb-user-node") + + spec := deployment.Spec.Template.Spec + assert.Equal(t, "linux", spec.NodeSelector["kubernetes.io/os"], + "the shadow must not be scheduled onto the virtual node it shadows") + assert.Empty(t, spec.Tolerations) + }) + + t.Run("waits for the compute node in an init container", func(t *testing.T) { + _, deployment := renderSSHShadow(t, sshConfig(), tcp) + + names := make([]string, 0, len(deployment.Spec.Template.Spec.InitContainers)) + for _, c := range deployment.Spec.Template.Spec.InitContainers { + names = append(names, c.Name) + } + assert.Contains(t, names, "wait-for-node", + "a queued job is the normal case; the shadow shows Init rather than crash-looping") + }) + + t.Run("kerberos renders kinit and no ssh key", func(t *testing.T) { + config := sshConfig() + config.Network.SSH.KeySecret = "" + config.Network.SSH.Auth = SSHAuthKerberos + config.Network.SSH.KeytabSecret = testKeytabSecret + config.Network.SSH.Principal = testPrincipal + + manifest, deployment := renderSSHShadow(t, config, tcp) + + assert.Contains(t, manifest, "kinit -k -t /interlink/keytab/"+DefaultSSHKeytabSecretKey+" alice@EXAMPLE.ORG") + assert.Contains(t, manifest, "GSSAPIAuthentication=yes") + assert.NotContains(t, manifest, "/interlink/ssh/") + + var initNames, names []string + for _, c := range deployment.Spec.Template.Spec.InitContainers { + initNames = append(initNames, c.Name) + } + for _, c := range deployment.Spec.Template.Spec.Containers { + names = append(names, c.Name) + } + // kinit runs after the wait so the ticket is fresh however long the queue was, + // and the renewer keeps it alive for the life of the tunnel. + assert.Equal(t, []string{"wait-for-node", "kinit"}, initNames) + assert.Contains(t, names, "kinit-renew") + }) + + t.Run("skips UDP ports, which ssh -L cannot carry", func(t *testing.T) { + manifest, _ := renderSSHShadow(t, sshConfig(), []PortMapping{ + {Port: 8888, Name: "notebook", Protocol: "TCP"}, + {Port: 9999, Name: "telemetry", Protocol: "UDP"}, + }) + + assert.Contains(t, manifest, "-L 0.0.0.0:8888:$node:8888") + assert.NotContains(t, manifest, "9999") + }) + + t.Run("pins host keys when a known_hosts ConfigMap is configured", func(t *testing.T) { + config := sshConfig() + config.Network.SSH.KnownHostsConfigMap = "hpc-known-hosts" + + manifest, _ := renderSSHShadow(t, config, tcp) + + assert.Contains(t, manifest, "StrictHostKeyChecking=yes") + assert.Contains(t, manifest, "UserKnownHostsFile=/interlink/known-hosts/known_hosts") + assert.NotContains(t, manifest, "accept-new") + }) + + t.Run("falls back to accept-new without one", func(t *testing.T) { + manifest, _ := renderSSHShadow(t, sshConfig(), tcp) + assert.Contains(t, manifest, "StrictHostKeyChecking=accept-new") + }) +} + +func TestReplicateShadowCredentials(t *testing.T) { + const source = "interlink" + target := shadowResourceIdentity{Name: "shadow-nb-user", Namespace: "user"} + + newClient := func() *fake.Clientset { + return fake.NewSimpleClientset( + &v1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "hpc-ssh-key", Namespace: source}, + Data: map[string][]byte{DefaultSSHKeySecretKey: []byte("PRIVATE KEY")}, + }, + &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: "hpc-known-hosts", Namespace: source}, + Data: map[string]string{"known_hosts": "login.hpc.example.org ssh-ed25519 AAAA"}, + }, + ) + } + + t.Run("copies the credential into the shadow namespace", func(t *testing.T) { + config := sshConfig() + config.Network.SSH.KnownHostsConfigMap = "hpc-known-hosts" + client := newClient() + p := &Provider{clientSet: client, config: normalized(t, config)} + + assert.NoError(t, p.replicateShadowCredentials(t.Context(), target)) + + secret, err := client.CoreV1().Secrets(target.Namespace).Get(t.Context(), "hpc-ssh-key", metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, []byte("PRIVATE KEY"), secret.Data[DefaultSSHKeySecretKey]) + + cm, err := client.CoreV1().ConfigMaps(target.Namespace).Get(t.Context(), "hpc-known-hosts", metav1.GetOptions{}) + assert.NoError(t, err) + assert.Contains(t, cm.Data["known_hosts"], "ssh-ed25519") + }) + + t.Run("does nothing when replication is switched off", func(t *testing.T) { + config := normalized(t, sshConfig()) + off := false + config.Network.SSH.ReplicateCredentials = &off + client := newClient() + p := &Provider{clientSet: client, config: config} + + assert.NoError(t, p.replicateShadowCredentials(t.Context(), target)) + + _, err := client.CoreV1().Secrets(target.Namespace).Get(t.Context(), "hpc-ssh-key", metav1.GetOptions{}) + assert.Error(t, err, "the operator opted out; nothing should be copied") + }) + + t.Run("reports a missing source credential instead of rendering a broken shadow", func(t *testing.T) { + p := &Provider{clientSet: fake.NewSimpleClientset(), config: normalized(t, sshConfig())} + + err := p.replicateShadowCredentials(t.Context(), target) + + assert.Error(t, err) + assert.Contains(t, err.Error(), "hpc-ssh-key") + }) + + t.Run("is a no-op when the shadow already lives in the source namespace", func(t *testing.T) { + client := newClient() + p := &Provider{clientSet: client, config: normalized(t, sshConfig())} + + assert.NoError(t, p.replicateShadowCredentials(t.Context(), shadowResourceIdentity{Name: "s", Namespace: source})) + }) +} + +// TestDefaultSSHNodeWaitMatchesTimeout keeps the duration fallback in step with the +// string default that ends up in the rendered manifest. +func TestDefaultSSHNodeWaitMatchesTimeout(t *testing.T) { + parsed, err := time.ParseDuration(DefaultSSHNodeWaitTimeout) + assert.NoError(t, err) + assert.Equal(t, parsed, defaultSSHNodeWait) +} diff --git a/pkg/virtualkubelet/templates/shadow-ssh-template.yaml b/pkg/virtualkubelet/templates/shadow-ssh-template.yaml new file mode 100644 index 00000000..227a485a --- /dev/null +++ b/pkg/virtualkubelet/templates/shadow-ssh-template.yaml @@ -0,0 +1,248 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{.Name}} + namespace: {{.Namespace}} +spec: + replicas: 1 + selector: + matchLabels: + app.kubernetes.io/component: {{.Name}} + template: + metadata: + labels: + app.kubernetes.io/component: {{.Name}} + spec: + # Never schedule the shadow onto the virtual node it shadows: real nodes are + # kubernetes.io/os=linux, the virtual one is kubernetes.io/os=virtual-kubelet. + nodeSelector: + kubernetes.io/os: linux + tolerations: [] + initContainers: + # The remote job is usually still queued when the shadow starts, so the pod + # sits in Init until the plugin reports which node it landed on. That is the + # expected path, not a failure: it reads as "waiting for the allocation" + # instead of a crash-looping tunnel. + - name: wait-for-node + image: {{.SSH.Image}} + command: ["/bin/sh", "-c"] + args: + - | + set -eu + deadline=$(( $(date +%s) + {{.SSHNodeWaitSeconds}} )) + while :; do + node=$(tr -d '[:space:]' < /interlink/node/{{.NodeNameKey}} 2>/dev/null || true) + [ -n "$node" ] && break + if [ "$(date +%s)" -ge "$deadline" ]; then + echo "timed out after {{.SSH.NodeWaitTimeout}} waiting for the plugin to report a compute node." >&2 + echo "The plugin must return PodStatus.NodeName once the job runs; plugins that do not are unsupported by the ssh shadow." >&2 + exit 1 + fi + sleep 5 + done + echo "compute node: $node" + volumeMounts: + - name: shadow-node + mountPath: /interlink/node + readOnly: true + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: ["ALL"] + {{- if eq .SSH.Auth "kerberos" }} + # Runs after the node is known so the ticket is fresh when ssh starts, however + # long the job waited in the queue. + - name: kinit + image: {{.SSH.Image}} + command: ["/bin/sh", "-c"] + args: + - kinit -k -t /interlink/keytab/{{.SSH.KeytabSecretKey}} {{.SSH.Principal}} + env: + - name: KRB5CCNAME + value: FILE:/interlink/krb5cc/ccache + volumeMounts: + - name: shadow-keytab + mountPath: /interlink/keytab + readOnly: true + - name: shadow-krb5cc + mountPath: /interlink/krb5cc + {{- if .SSH.Krb5ConfigMap }} + - name: shadow-krb5-config + mountPath: /etc/krb5.conf + subPath: krb5.conf + readOnly: true + {{- end }} + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: ["ALL"] + {{- end }} + containers: + {{- if eq .SSH.Auth "kerberos" }} + - name: kinit-renew + image: {{.SSH.Image}} + command: ["/bin/sh", "-c"] + args: + - | + set -u + while :; do + sleep 21600 + kinit -k -t /interlink/keytab/{{.SSH.KeytabSecretKey}} {{.SSH.Principal}} || true + done + env: + - name: KRB5CCNAME + value: FILE:/interlink/krb5cc/ccache + volumeMounts: + - name: shadow-keytab + mountPath: /interlink/keytab + readOnly: true + - name: shadow-krb5cc + mountPath: /interlink/krb5cc + {{- if .SSH.Krb5ConfigMap }} + - name: shadow-krb5-config + mountPath: /etc/krb5.conf + subPath: krb5.conf + readOnly: true + {{- end }} + resources: + requests: + cpu: 10m + memory: 32Mi + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: ["ALL"] + {{- end }} + - name: ssh-forward + image: {{.SSH.Image}} + command: ["/bin/sh", "-c"] + args: + - | + set -eu + node=$(tr -d '[:space:]' < /interlink/node/{{.NodeNameKey}}) + [ -n "$node" ] || { echo "no compute node recorded" >&2; exit 1; } + echo "forwarding{{range .ExposedPorts}}{{if ne .Protocol "UDP"}} {{.Port}}{{end}}{{end}} to $node via {{.SSH.LoginHost}}" + exec ssh -N \ + -p {{.SSH.Port}} \ + -o BatchMode=yes \ + -o ExitOnForwardFailure=yes \ + -o ServerAliveInterval=30 \ + -o ServerAliveCountMax=3 \ + {{- if eq .SSH.Auth "kerberos" }} + -o GSSAPIAuthentication=yes \ + -o GSSAPIDelegateCredentials=yes \ + -o PreferredAuthentications=gssapi-with-mic \ + {{- else }} + -i /interlink/ssh/{{.SSH.KeySecretKey}} \ + -o IdentitiesOnly=yes \ + -o PreferredAuthentications=publickey \ + {{- end }} + {{- if .SSH.KnownHostsConfigMap }} + -o StrictHostKeyChecking=yes \ + -o UserKnownHostsFile=/interlink/known-hosts/known_hosts \ + {{- else }} + -o StrictHostKeyChecking=accept-new \ + {{- end }} + {{- range .SSH.ExtraOptions }} + -o {{.}} \ + {{- end }} + {{- range .ExposedPorts }} + {{- if ne .Protocol "UDP" }} + -L 0.0.0.0:{{.Port}}:$node:{{.Port}} \ + {{- end }} + {{- end }} + {{.SSH.User}}@{{.SSH.LoginHost}} + {{- if eq .SSH.Auth "kerberos" }} + env: + - name: KRB5CCNAME + value: FILE:/interlink/krb5cc/ccache + {{- end }} + ports: + {{- range .ExposedPorts }} + {{- if ne .Protocol "UDP" }} + - containerPort: {{.Port}} + name: {{if .Name}}{{.Name}}{{else}}port-{{.Port}}{{end}} + protocol: {{.Protocol}} + {{- end }} + {{- end }} + volumeMounts: + - name: shadow-node + mountPath: /interlink/node + readOnly: true + {{- if eq .SSH.Auth "kerberos" }} + - name: shadow-krb5cc + mountPath: /interlink/krb5cc + readOnly: true + {{- if .SSH.Krb5ConfigMap }} + - name: shadow-krb5-config + mountPath: /etc/krb5.conf + subPath: krb5.conf + readOnly: true + {{- end }} + {{- else }} + - name: shadow-ssh-key + mountPath: /interlink/ssh + readOnly: true + {{- end }} + {{- if .SSH.KnownHostsConfigMap }} + - name: shadow-known-hosts + mountPath: /interlink/known-hosts + readOnly: true + {{- end }} + resources: + requests: + cpu: 100m + memory: 64Mi + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: ["ALL"] + volumes: + # Mounted as a directory, not subPath: only directory mounts are refreshed in + # place when the VK writes the compute node into this ConfigMap. + - name: shadow-node + configMap: + name: {{.NodeConfigMap}} + {{- if eq .SSH.Auth "kerberos" }} + - name: shadow-keytab + secret: + secretName: {{.SSH.KeytabSecret}} + defaultMode: 0400 + - name: shadow-krb5cc + emptyDir: + medium: Memory + {{- if .SSH.Krb5ConfigMap }} + - name: shadow-krb5-config + configMap: + name: {{.SSH.Krb5ConfigMap}} + {{- end }} + {{- else }} + - name: shadow-ssh-key + secret: + secretName: {{.SSH.KeySecret}} + defaultMode: 0400 + {{- end }} + {{- if .SSH.KnownHostsConfigMap }} + - name: shadow-known-hosts + configMap: + name: {{.SSH.KnownHostsConfigMap}} + {{- end }} +--- +apiVersion: v1 +kind: Service +metadata: + name: {{.Name}} + namespace: {{.Namespace}} +spec: + type: ClusterIP + selector: + app.kubernetes.io/component: {{.Name}} + ports: + {{- range .ExposedPorts }} + {{- if ne .Protocol "UDP" }} + - port: {{.Port}} + targetPort: {{.Port}} + name: {{if .Name}}{{.Name}}{{else}}port-{{.Port}}{{end}} + protocol: {{.Protocol}} + {{- end }} + {{- end }} diff --git a/pkg/virtualkubelet/virtualkubelet.go b/pkg/virtualkubelet/virtualkubelet.go index d96668cf..9ba0591d 100644 --- a/pkg/virtualkubelet/virtualkubelet.go +++ b/pkg/virtualkubelet/virtualkubelet.go @@ -44,7 +44,7 @@ import ( k8stypes "k8s.io/apimachinery/pkg/types" ) -//go:embed templates/wstunnel-template.yaml templates/wstunnel-wireguard-template.yaml +//go:embed templates/wstunnel-template.yaml templates/wstunnel-wireguard-template.yaml templates/shadow-ssh-template.yaml var defaultShadowTemplates embed.FS //go:embed all:templates/mesh.sh @@ -111,6 +111,11 @@ type ShadowTemplateData struct { NodeConfigMap string // NodeNameKey is the key inside NodeConfigMap holding the node name. NodeNameKey string + // SSH carries the SSH port-forward settings, used by the ssh shadow template. + SSH SSHTunnel + // SSHNodeWaitSeconds is SSH.NodeWaitTimeout in whole seconds, for the shell loop + // that waits on NodeConfigMap. + SSHNodeWaitSeconds int } type PortMapping struct { @@ -589,6 +594,10 @@ func LoadConfig(ctx context.Context, providerConfig string) (config Config, err // config = configMap SetDefaultResource(&config) + if err = NormalizeShadowConfig(&config); err != nil { + return config, err + } + if _, err = resource.ParseQuantity(config.Resources.CPU); err != nil { return config, fmt.Errorf("invalid CPU value %v", config.Resources.CPU) } @@ -958,6 +967,12 @@ func (p *Provider) createShadowPod(ctx context.Context, originalPod *v1.Pod) (*v return nil, nil, err } + if p.isSSHShadow() { + if err := p.replicateShadowCredentials(ctx, identity); err != nil { + return nil, nil, err + } + } + localContainers := getLocalContainers(originalPod) localInitContainers := getLocalInitContainers(originalPod) @@ -981,6 +996,12 @@ func (p *Provider) createShadowPod(ctx context.Context, originalPod *v1.Pod) (*v FullMesh: fullMeshEnabledForPod, NodeConfigMap: shadowNodeConfigMapName(identity.Name), NodeNameKey: shadowNodeNameKey, + SSH: p.config.Network.SSH, + SSHNodeWaitSeconds: p.sshNodeWaitSeconds(), + } + + if p.isSSHShadow() { + warnOnUDPPorts(ctx, templateData.ExposedPorts) } log.G(ctx).Debugf("LocalInitContainers count: %d", len(templateData.LocalInitContainers)) @@ -1098,7 +1119,10 @@ func (p *Provider) executeShadowTemplate(ctx context.Context, data ShadowTemplat // Fall back to embedded template if templateContent == "" { templatePath := "templates/wstunnel-template.yaml" - if data.FullMesh { + switch { + case p.isSSHShadow(): + templatePath = "templates/shadow-ssh-template.yaml" + case data.FullMesh: templatePath = "templates/wstunnel-wireguard-template.yaml" } content, err := defaultShadowTemplates.ReadFile(templatePath) @@ -1676,11 +1700,15 @@ func (p *Provider) handleShadowCreation(ctx context.Context, pod *v1.Pod) (strin return "", err } - // Add wstunnel client command annotation to the original pod - if err := p.addWstunnelClientAnnotation(ctx, pod, templateData); err != nil { - log.G(ctx).Warningf("Failed to add wstunnel client annotation to pod %s/%s: %v", pod.Namespace, pod.Name, err) - // Note: We don't clean up here since the shadow infrastructure is working, - // just the annotation failed (non-critical) + // The SSH shadow dials in to the login node, so the workload has nothing to set + // up on its side: no wstunnel client to launch, no WireGuard config to apply. + if !p.isSSHShadow() { + // Add wstunnel client command annotation to the original pod + if err := p.addWstunnelClientAnnotation(ctx, pod, templateData); err != nil { + log.G(ctx).Warningf("Failed to add wstunnel client annotation to pod %s/%s: %v", pod.Namespace, pod.Name, err) + // Note: We don't clean up here since the shadow infrastructure is working, + // just the annotation failed (non-critical) + } } return podIP, nil From e5cfb668c2da201cb1cbbb255bbce226c9d119fb Mon Sep 17 00:00:00 2001 From: Matteo Bunino <48362942+matbun@users.noreply.github.com> Date: Fri, 31 Jul 2026 11:36:24 +0200 Subject: [PATCH 4/7] fix: rebuild the SSH shadow tunnel when the job moves to another node A requeued Slurm job comes back on a different host and the plugin reports the new one, which the virtual kubelet republishes into the shadow's node ConfigMap. The kubelet refreshes the mount, but ssh already had the old hostname baked into its -L arguments, so the tunnel kept pointing at the node the job had left. Supervise ssh instead of exec'ing it: poll the mounted file while it runs and restart it against the new node when the name changes. An ssh exit with the node unchanged still ends the container, so a dead tunnel is restarted by Kubernetes as before. Also name the login-node requirement in the startup line. A site with "AllowTcpForwarding no" only fails once traffic arrives, with a bare "channel N: open failed: administratively prohibited" and nothing pointing at the cause; now the requirement sits directly above it in the same log. Signed-off-by: Matteo Bunino <48362942+matbun@users.noreply.github.com> Co-Authored-By: Claude Opus 5 (1M context) --- .../templates/shadow-ssh-template.yaml | 95 ++++++++++++------- 1 file changed, 62 insertions(+), 33 deletions(-) diff --git a/pkg/virtualkubelet/templates/shadow-ssh-template.yaml b/pkg/virtualkubelet/templates/shadow-ssh-template.yaml index 227a485a..afb5fa8d 100644 --- a/pkg/virtualkubelet/templates/shadow-ssh-template.yaml +++ b/pkg/virtualkubelet/templates/shadow-ssh-template.yaml @@ -118,40 +118,69 @@ spec: command: ["/bin/sh", "-c"] args: - | - set -eu - node=$(tr -d '[:space:]' < /interlink/node/{{.NodeNameKey}}) + set -u + read_node() { tr -d '[:space:]' < /interlink/node/{{.NodeNameKey}} 2>/dev/null || true; } + + node=$(read_node) [ -n "$node" ] || { echo "no compute node recorded" >&2; exit 1; } - echo "forwarding{{range .ExposedPorts}}{{if ne .Protocol "UDP"}} {{.Port}}{{end}}{{end}} to $node via {{.SSH.LoginHost}}" - exec ssh -N \ - -p {{.SSH.Port}} \ - -o BatchMode=yes \ - -o ExitOnForwardFailure=yes \ - -o ServerAliveInterval=30 \ - -o ServerAliveCountMax=3 \ - {{- if eq .SSH.Auth "kerberos" }} - -o GSSAPIAuthentication=yes \ - -o GSSAPIDelegateCredentials=yes \ - -o PreferredAuthentications=gssapi-with-mic \ - {{- else }} - -i /interlink/ssh/{{.SSH.KeySecretKey}} \ - -o IdentitiesOnly=yes \ - -o PreferredAuthentications=publickey \ - {{- end }} - {{- if .SSH.KnownHostsConfigMap }} - -o StrictHostKeyChecking=yes \ - -o UserKnownHostsFile=/interlink/known-hosts/known_hosts \ - {{- else }} - -o StrictHostKeyChecking=accept-new \ - {{- end }} - {{- range .SSH.ExtraOptions }} - -o {{.}} \ - {{- end }} - {{- range .ExposedPorts }} - {{- if ne .Protocol "UDP" }} - -L 0.0.0.0:{{.Port}}:$node:{{.Port}} \ - {{- end }} - {{- end }} - {{.SSH.User}}@{{.SSH.LoginHost}} + + # A requeued job comes back on a different host, and the virtual kubelet + # republishes it into the mounted ConfigMap. ssh has the old name baked + # into its -L arguments, so watch for the change and rebuild the tunnel. + while :; do + echo "forwarding{{range .ExposedPorts}}{{if ne .Protocol "UDP"}} {{.Port}}{{end}}{{end}} to $node via {{.SSH.LoginHost}} (the login node must allow TCP forwarding)" + ssh -N \ + -p {{.SSH.Port}} \ + -o BatchMode=yes \ + -o ExitOnForwardFailure=yes \ + -o ServerAliveInterval=30 \ + -o ServerAliveCountMax=3 \ + {{- if eq .SSH.Auth "kerberos" }} + -o GSSAPIAuthentication=yes \ + -o GSSAPIDelegateCredentials=yes \ + -o PreferredAuthentications=gssapi-with-mic \ + {{- else }} + -i /interlink/ssh/{{.SSH.KeySecretKey}} \ + -o IdentitiesOnly=yes \ + -o PreferredAuthentications=publickey \ + {{- end }} + {{- if .SSH.KnownHostsConfigMap }} + -o StrictHostKeyChecking=yes \ + -o UserKnownHostsFile=/interlink/known-hosts/known_hosts \ + {{- else }} + -o StrictHostKeyChecking=accept-new \ + {{- end }} + {{- range .SSH.ExtraOptions }} + -o {{.}} \ + {{- end }} + {{- range .ExposedPorts }} + {{- if ne .Protocol "UDP" }} + -L 0.0.0.0:{{.Port}}:$node:{{.Port}} \ + {{- end }} + {{- end }} + {{.SSH.User}}@{{.SSH.LoginHost}} & + sshpid=$! + + while kill -0 "$sshpid" 2>/dev/null; do + current=$(read_node) + if [ -n "$current" ] && [ "$current" != "$node" ]; then + echo "compute node changed from $node to $current, rebuilding the tunnel" + kill "$sshpid" 2>/dev/null || true + break + fi + sleep 10 + done + + wait "$sshpid" + rc=$? + current=$(read_node) + if [ -n "$current" ] && [ "$current" != "$node" ]; then + node=$current + continue + fi + echo "ssh exited with status $rc" >&2 + exit "$rc" + done {{- if eq .SSH.Auth "kerberos" }} env: - name: KRB5CCNAME From 7dc98917417b7f8d234ca83e4cf05a1c0242b4ae Mon Sep 17 00:00:00 2001 From: Matteo Bunino <48362942+matbun@users.noreply.github.com> Date: Fri, 31 Jul 2026 13:08:54 +0200 Subject: [PATCH 5/7] feat: add an exec forward mode for login nodes that refuse port forwarding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ssh -L needs AllowTcpForwarding on the login node, and a good number of HPC sites do not grant it: SURF's ETP testbed sets "AllowTcpForwarding no" globally and re-enables it only for the MFA group. The shadow there authenticates, starts, reports Ready, and then resets every connection with "administratively prohibited". Nothing about that is fixable from the cluster. Network.SSH.ForwardMode: exec drops the forwarding channel entirely. Each exposed port gets a local socat listener in the shadow, and every accepted connection is relayed by a command run on the login node (Network.SSH.ExecConnectCommand, "nc " by default). All it needs is the ability to run a command over SSH, which is the same privilege the shadow already uses to log in. Both modes are interchangeable above the socket: the Service still fronts the same containerPort, and the offloaded workload still just binds 0.0.0.0. Only the transport differs. The relays share one multiplexed SSH connection, established as an explicit control master before the listeners open. Without it each connection would pay a full handshake — ruinous for something like a notebook UI, which opens dozens — and the login node would see a session per connection instead of one. portforward stays the default; it is cheaper and has no dependency on the site having a netcat. Signed-off-by: Matteo Bunino <48362942+matbun@users.noreply.github.com> Co-Authored-By: Claude Opus 5 (1M context) --- docker/Dockerfile.ssh-tunnel | 7 +- pkg/virtualkubelet/config.go | 27 ++++ pkg/virtualkubelet/shadow_ssh.go | 13 ++ pkg/virtualkubelet/shadow_ssh_test.go | 37 +++++ .../templates/shadow-ssh-template.yaml | 149 +++++++++++++----- 5 files changed, 193 insertions(+), 40 deletions(-) diff --git a/docker/Dockerfile.ssh-tunnel b/docker/Dockerfile.ssh-tunnel index 9a93fb96..6b40f904 100644 --- a/docker/Dockerfile.ssh-tunnel +++ b/docker/Dockerfile.ssh-tunnel @@ -2,13 +2,14 @@ # # It runs `ssh -N -L ...` against an HPC login node, so it needs an ssh client and, # for GSSAPI sites, the Kerberos client tools to obtain and renew a ticket from a -# keytab. Nothing from this repository is installed: the shadow runs no interLink -# binary, only stock ssh. +# keytab. socat provides the local listeners used by Network.SSH.ForwardMode "exec", +# where the login node grants no forwarding privilege. Nothing from this repository +# is installed: the shadow runs no interLink binary, only stock ssh. FROM debian:bookworm-slim RUN apt-get update \ && apt-get install -y --no-install-recommends \ - openssh-client krb5-user ca-certificates \ + openssh-client krb5-user socat ca-certificates \ && rm -rf /var/lib/apt/lists/* CMD ["/bin/sh"] diff --git a/pkg/virtualkubelet/config.go b/pkg/virtualkubelet/config.go index 45be7a37..e0426d33 100644 --- a/pkg/virtualkubelet/config.go +++ b/pkg/virtualkubelet/config.go @@ -191,6 +191,19 @@ const ( ShadowModeSSH = "ssh" ) +// Traffic-forwarding strategies selectable through SSHTunnel.ForwardMode. +const ( + // SSHForwardModePortForward uses `ssh -L`, and needs AllowTcpForwarding on the + // login node. This is the default. + SSHForwardModePortForward = "portforward" + // SSHForwardModeExec pipes each connection through a command run on the login + // node, for sites that do not grant TCP forwarding. + SSHForwardModeExec = "exec" +) + +// DefaultSSHExecConnectCommand relays a connection on stdin/stdout in "exec" mode. +const DefaultSSHExecConnectCommand = "nc" + // SSH authentication methods selectable through SSHTunnel.Auth. const ( // SSHAuthPublicKey authenticates with a private key from KeySecret. @@ -218,6 +231,20 @@ type SSHTunnel struct { Image string `yaml:"Image,omitempty"` // Auth selects the authentication method: "publickey" (default) or "kerberos" Auth string `yaml:"Auth,omitempty"` + // ForwardMode selects how traffic reaches the compute node: + // + // "portforward" (default) — one `ssh -L` per exposed port. Cheapest and most + // direct, but the login node must set AllowTcpForwarding yes for this + // account. Sites that disable it refuse every channel with + // "administratively prohibited". + // "exec" — a local listener per exposed port, each connection piped through a + // command run on the login node (see ExecConnectCommand). Needs no + // forwarding privilege at all, at the cost of one ssh process per connection. + ForwardMode string `yaml:"ForwardMode,omitempty"` + // ExecConnectCommand is the command run on the login node in "exec" mode. It is + // invoked as ` ` and must relay the connection on + // its stdin and stdout. Defaults to "nc". + ExecConnectCommand string `yaml:"ExecConnectCommand,omitempty"` // KeySecret is the Secret holding the SSH private key ("publickey" auth) KeySecret string `yaml:"KeySecret,omitempty"` // KeySecretKey is the key inside KeySecret holding the private key (default "id_ed25519") diff --git a/pkg/virtualkubelet/shadow_ssh.go b/pkg/virtualkubelet/shadow_ssh.go index 524386ab..19a2d829 100644 --- a/pkg/virtualkubelet/shadow_ssh.go +++ b/pkg/virtualkubelet/shadow_ssh.go @@ -88,6 +88,19 @@ func NormalizeShadowConfig(config *Config) error { s.ReplicateCredentials = &replicate } + s.ForwardMode = strings.ToLower(strings.TrimSpace(s.ForwardMode)) + switch s.ForwardMode { + case "": + s.ForwardMode = SSHForwardModePortForward + case SSHForwardModePortForward: + case SSHForwardModeExec: + if strings.TrimSpace(s.ExecConnectCommand) == "" { + s.ExecConnectCommand = DefaultSSHExecConnectCommand + } + default: + return fmt.Errorf("unknown Network.SSH.ForwardMode %q: expected %q or %q", s.ForwardMode, SSHForwardModePortForward, SSHForwardModeExec) + } + s.Auth = strings.ToLower(strings.TrimSpace(s.Auth)) switch s.Auth { case "": diff --git a/pkg/virtualkubelet/shadow_ssh_test.go b/pkg/virtualkubelet/shadow_ssh_test.go index 9e59c062..47bf4d03 100644 --- a/pkg/virtualkubelet/shadow_ssh_test.go +++ b/pkg/virtualkubelet/shadow_ssh_test.go @@ -55,11 +55,21 @@ func TestNormalizeShadowConfigDefaults(t *testing.T) { assert.Equal(t, SSHAuthPublicKey, s.Auth) assert.Equal(t, DefaultSSHKeySecretKey, s.KeySecretKey) assert.Equal(t, DefaultSSHNodeWaitTimeout, s.NodeWaitTimeout) + assert.Equal(t, SSHForwardModePortForward, s.ForwardMode) assert.Contains(t, s.Image, "ssh-tunnel") assert.NotNil(t, s.ReplicateCredentials) assert.True(t, *s.ReplicateCredentials, "credentials replicate by default so per-user namespaces work") }) + t.Run("exec forward mode gets a relay command", func(t *testing.T) { + config := sshConfig() + config.Network.SSH.ForwardMode = "EXEC" + + config = normalized(t, config) + assert.Equal(t, SSHForwardModeExec, config.Network.SSH.ForwardMode) + assert.Equal(t, DefaultSSHExecConnectCommand, config.Network.SSH.ExecConnectCommand) + }) + t.Run("kerberos defaults", func(t *testing.T) { config := sshConfig() config.Network.SSH.KeySecret = "" @@ -132,6 +142,11 @@ func TestNormalizeShadowConfigRejects(t *testing.T) { mutate: func(c *Config) { c.Network.SSH.NodeWaitTimeout = "forever" }, wantErr: "invalid Network.SSH.NodeWaitTimeout", }, + { + name: "unknown forward mode", + mutate: func(c *Config) { c.Network.SSH.ForwardMode = "smoke-signal" }, + wantErr: "unknown Network.SSH.ForwardMode", + }, } for _, tt := range tests { @@ -237,6 +252,28 @@ func TestSSHShadowTemplate(t *testing.T) { assert.Contains(t, names, "kinit-renew") }) + t.Run("exec mode relays through the login node instead of forwarding", func(t *testing.T) { + config := sshConfig() + config.Network.SSH.ForwardMode = SSHForwardModeExec + + manifest, deployment := renderSSHShadow(t, config, tcp) + + // No -L anywhere: sites running the exec mode are exactly the ones whose sshd + // refuses to open a forwarded channel at all. + assert.NotContains(t, manifest, "-L 0.0.0.0:") + assert.Contains(t, manifest, "socat TCP-LISTEN:8888,fork,reuseaddr,bind=0.0.0.0") + assert.Contains(t, manifest, `"$connect_cmd" "$node"`) + // One multiplexed connection, or every request would pay an SSH handshake and + // the login node would see a session per connection. + assert.Contains(t, manifest, "ssh -M -N -o ControlMaster=yes") + assert.Contains(t, manifest, "ControlPath=%s") + + // The Service still fronts the same containerPort, so nothing above the socket + // can tell the two modes apart. + container := deployment.Spec.Template.Spec.Containers[0] + assert.Equal(t, int32(8888), container.Ports[0].ContainerPort) + }) + t.Run("skips UDP ports, which ssh -L cannot carry", func(t *testing.T) { manifest, _ := renderSSHShadow(t, sshConfig(), []PortMapping{ {Port: 8888, Name: "notebook", Protocol: "TCP"}, diff --git a/pkg/virtualkubelet/templates/shadow-ssh-template.yaml b/pkg/virtualkubelet/templates/shadow-ssh-template.yaml index afb5fa8d..c52c4d09 100644 --- a/pkg/virtualkubelet/templates/shadow-ssh-template.yaml +++ b/pkg/virtualkubelet/templates/shadow-ssh-template.yaml @@ -121,65 +121,140 @@ spec: set -u read_node() { tr -d '[:space:]' < /interlink/node/{{.NodeNameKey}} 2>/dev/null || true; } - node=$(read_node) - [ -n "$node" ] || { echo "no compute node recorded" >&2; exit 1; } + target={{.SSH.User}}@{{.SSH.LoginHost}} + ssh_opts="\ + -p {{.SSH.Port}} \ + -o BatchMode=yes \ + -o ServerAliveInterval=30 \ + -o ServerAliveCountMax=3 \ + {{- if eq .SSH.Auth "kerberos" }} + -o GSSAPIAuthentication=yes \ + -o GSSAPIDelegateCredentials=yes \ + -o PreferredAuthentications=gssapi-with-mic \ + {{- else }} + -i /interlink/ssh/{{.SSH.KeySecretKey}} \ + -o IdentitiesOnly=yes \ + -o PreferredAuthentications=publickey \ + {{- end }} + {{- if .SSH.KnownHostsConfigMap }} + -o StrictHostKeyChecking=yes \ + -o UserKnownHostsFile=/interlink/known-hosts/known_hosts \ + {{- else }} + -o StrictHostKeyChecking=accept-new \ + {{- end }} + {{- range .SSH.ExtraOptions }} + -o {{.}} \ + {{- end }} + " - # A requeued job comes back on a different host, and the virtual kubelet - # republishes it into the mounted ConfigMap. ssh has the old name baked - # into its -L arguments, so watch for the change and rebuild the tunnel. - while :; do - echo "forwarding{{range .ExposedPorts}}{{if ne .Protocol "UDP"}} {{.Port}}{{end}}{{end}} to $node via {{.SSH.LoginHost}} (the login node must allow TCP forwarding)" - ssh -N \ - -p {{.SSH.Port}} \ - -o BatchMode=yes \ - -o ExitOnForwardFailure=yes \ - -o ServerAliveInterval=30 \ - -o ServerAliveCountMax=3 \ - {{- if eq .SSH.Auth "kerberos" }} - -o GSSAPIAuthentication=yes \ - -o GSSAPIDelegateCredentials=yes \ - -o PreferredAuthentications=gssapi-with-mic \ - {{- else }} - -i /interlink/ssh/{{.SSH.KeySecretKey}} \ - -o IdentitiesOnly=yes \ - -o PreferredAuthentications=publickey \ - {{- end }} - {{- if .SSH.KnownHostsConfigMap }} - -o StrictHostKeyChecking=yes \ - -o UserKnownHostsFile=/interlink/known-hosts/known_hosts \ - {{- else }} - -o StrictHostKeyChecking=accept-new \ + pids= + + {{- if eq .SSH.ForwardMode "exec" }} + # One local listener per port, each accepted connection relayed by a + # command run on the login node. Nothing here asks sshd to forward + # anything, which is the point: it works where AllowTcpForwarding is off. + forward_dir=/tmp/interlink-forward + control="$forward_dir/master" + connect_cmd='{{.SSH.ExecConnectCommand}}' + + start_forwarders() { + mkdir -p "$forward_dir" + rm -f "$control" + + # Every relayed stream rides one multiplexed SSH connection, so opening a + # connection costs a channel rather than a full handshake, and the login + # node sees a single session instead of one per connection. + # shellcheck disable=SC2086 + ssh -M -N -o ControlMaster=yes -o ControlPath="$control" -o ControlPersist=no $ssh_opts "$target" & + pids=$! + + waited=0 + # shellcheck disable=SC2086 + until ssh -O check -o ControlPath="$control" $ssh_opts "$target" >/dev/null 2>&1; do + waited=$((waited + 1)) + if [ "$waited" -ge 30 ]; then + echo "the ssh control master did not come up" >&2 + return 1 + fi + sleep 1 + done + {{- range .ExposedPorts }} + {{- if ne .Protocol "UDP" }} + + printf '#!/bin/sh\nexec ssh -q -o ControlMaster=auto -o ControlPath=%s %s %s "%s %s {{.Port}}"\n' \ + "$control" "$ssh_opts" "$target" "$connect_cmd" "$node" > "$forward_dir/{{.Port}}" + chmod +x "$forward_dir/{{.Port}}" + socat TCP-LISTEN:{{.Port}},fork,reuseaddr,bind=0.0.0.0 EXEC:"$forward_dir/{{.Port}}" & + pids="$pids $!" {{- end }} - {{- range .SSH.ExtraOptions }} - -o {{.}} \ {{- end }} + } + {{- else }} + # One -L per port. The login node must set AllowTcpForwarding yes for this + # account, or every channel is refused with "administratively prohibited". + start_forwarders() { + # shellcheck disable=SC2086 + ssh -N -o ExitOnForwardFailure=yes $ssh_opts \ {{- range .ExposedPorts }} {{- if ne .Protocol "UDP" }} -L 0.0.0.0:{{.Port}}:$node:{{.Port}} \ {{- end }} {{- end }} - {{.SSH.User}}@{{.SSH.LoginHost}} & - sshpid=$! + "$target" & + pids=$! + } + {{- end }} - while kill -0 "$sshpid" 2>/dev/null; do + all_alive() { + for pid in $pids; do + kill -0 "$pid" 2>/dev/null || return 1 + done + return 0 + } + + stop_forwarders() { + for pid in $pids; do + kill "$pid" 2>/dev/null || true + done + wait 2>/dev/null || true + pids= + } + + node=$(read_node) + [ -n "$node" ] || { echo "no compute node recorded" >&2; exit 1; } + + # A requeued job comes back on a different host and the virtual kubelet + # republishes it into the mounted ConfigMap, but the running forwarders + # still hold the old name. Watch for the change and rebuild them. + while :; do + echo "forwarding{{range .ExposedPorts}}{{if ne .Protocol "UDP"}} {{.Port}}{{end}}{{end}} to $node via {{.SSH.LoginHost}} ({{.SSH.ForwardMode}} mode)" + if ! start_forwarders; then + stop_forwarders + exit 1 + fi + + changed=false + while all_alive; do current=$(read_node) if [ -n "$current" ] && [ "$current" != "$node" ]; then echo "compute node changed from $node to $current, rebuilding the tunnel" - kill "$sshpid" 2>/dev/null || true + node=$current + changed=true break fi sleep 10 done - wait "$sshpid" - rc=$? + stop_forwarders + $changed && continue + current=$(read_node) if [ -n "$current" ] && [ "$current" != "$node" ]; then node=$current continue fi - echo "ssh exited with status $rc" >&2 - exit "$rc" + echo "the tunnel exited; letting Kubernetes restart the container" >&2 + exit 1 done {{- if eq .SSH.Auth "kerberos" }} env: From aa8081d904b256bb3f5dc57badae2287b1abd972 Mon Sep 17 00:00:00 2001 From: Matteo Bunino <48362942+matbun@users.noreply.github.com> Date: Fri, 31 Jul 2026 15:49:06 +0200 Subject: [PATCH 6/7] fix: refuse a compute node name that is not a hostname The node a plugin reports goes straight into a shell command, twice: into the `-L` argument in portforward mode, and into the relay command that exec mode sends to the login node. Neither was quoted, and neither was checked. A name containing whitespace splits into an extra ssh argument, and ssh takes options anywhere: -L 0.0.0.0:8888:h:1 -oProxyCommand=touch:8888 user@login ProxyCommand runs a command in the shadow container. In exec mode a name containing a quote or `$(` breaks out of the relay command instead, which the login node's shell then re-parses. For the Slurm plugin this value is `hostname -f` written to a file in the pod's job directory on a shared filesystem, so it is only as trustworthy as the permissions on that directory, and a plugin that computes it some other way could be wrong in more interesting ways. Refuse anything that is not an RFC 1123 hostname or an IP literal, before it reaches the ConfigMap the shadow mounts, and log why. Escaping was the other option, but there are two shells and two modes to get right and no legitimate node name needs it. Quote the node in both modes as well, so a future path that skips the check cannot reintroduce the same bug quietly. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Matteo Bunino <48362942+matbun@users.noreply.github.com> --- pkg/virtualkubelet/shadow.go | 25 ++++++++ pkg/virtualkubelet/shadow_ssh_test.go | 9 +-- pkg/virtualkubelet/shadow_test.go | 63 +++++++++++++++++++ .../templates/shadow-ssh-template.yaml | 9 ++- 4 files changed, 99 insertions(+), 7 deletions(-) diff --git a/pkg/virtualkubelet/shadow.go b/pkg/virtualkubelet/shadow.go index e5762497..02427048 100644 --- a/pkg/virtualkubelet/shadow.go +++ b/pkg/virtualkubelet/shadow.go @@ -6,6 +6,7 @@ import ( "encoding/hex" "encoding/json" "fmt" + "regexp" "strings" "github.com/containerd/containerd/log" @@ -30,6 +31,24 @@ const shadowNodeConfigMapSuffix = "-node" // shadowNodeNameKey is the key inside that ConfigMap holding the node name. const shadowNodeNameKey = "compute-node" +// maxComputeNodeNameLen is the longest DNS name, and so the longest thing a plugin +// can legitimately report as a compute node. +const maxComputeNodeNameLen = 253 + +// computeNodeNamePattern matches a hostname or an IP literal and nothing else. The +// shadow interpolates the reported node into a shell command, both locally and on +// the login node, so anything outside this set has to be refused rather than +// escaped: a name containing a space injects an extra ssh argument +// (`-oProxyCommand=...` runs a command in the shadow), and one containing a quote +// or `$(` breaks out of the relay command in exec mode. +var computeNodeNamePattern = regexp.MustCompile(`^[A-Za-z0-9]([A-Za-z0-9._:-]*[A-Za-z0-9])?$`) + +// isValidComputeNodeName reports whether a plugin-supplied node name is safe to put +// in front of ssh. +func isValidComputeNodeName(nodeName string) bool { + return len(nodeName) <= maxComputeNodeNameLen && computeNodeNamePattern.MatchString(nodeName) +} + func sanitizeDNSName(name string) string { // Convert to lowercase name = strings.ToLower(name) @@ -292,6 +311,12 @@ func (p *Provider) publishShadowNodeName(ctx context.Context, pod *v1.Pod, nodeN if nodeName == "" || !p.hasShadow(pod) { return } + if !isValidComputeNodeName(nodeName) { + log.G(ctx).Errorf( + "Refusing to publish compute node %q for %s/%s: not a hostname or IP address. The shadow passes this to ssh, so it is rejected rather than escaped.", + nodeName, pod.Namespace, pod.Name) + return + } if last, ok := p.shadowNodeNames.Load(string(pod.UID)); ok && last == nodeName { return } diff --git a/pkg/virtualkubelet/shadow_ssh_test.go b/pkg/virtualkubelet/shadow_ssh_test.go index 47bf4d03..3abb0564 100644 --- a/pkg/virtualkubelet/shadow_ssh_test.go +++ b/pkg/virtualkubelet/shadow_ssh_test.go @@ -202,7 +202,7 @@ func TestSSHShadowTemplate(t *testing.T) { t.Run("forwards each exposed port to the node the plugin reported", func(t *testing.T) { manifest, deployment := renderSSHShadow(t, sshConfig(), tcp) - assert.Contains(t, manifest, "-L 0.0.0.0:8888:$node:8888") + assert.Contains(t, manifest, `-L "0.0.0.0:8888:$node:8888"`) assert.Contains(t, manifest, "alice@login.hpc.example.org") assert.Contains(t, manifest, "-i /interlink/ssh/"+DefaultSSHKeySecretKey) // The node arrives through a mounted ConfigMap, not the pod spec: restarting @@ -260,9 +260,10 @@ func TestSSHShadowTemplate(t *testing.T) { // No -L anywhere: sites running the exec mode are exactly the ones whose sshd // refuses to open a forwarded channel at all. - assert.NotContains(t, manifest, "-L 0.0.0.0:") + assert.NotContains(t, manifest, "-L \"0.0.0.0:") assert.Contains(t, manifest, "socat TCP-LISTEN:8888,fork,reuseaddr,bind=0.0.0.0") - assert.Contains(t, manifest, `"$connect_cmd" "$node"`) + // the node is single-quoted for the login node's shell, which re-parses it + assert.Contains(t, manifest, `remote_cmd="$connect_cmd '$node'"`) // One multiplexed connection, or every request would pay an SSH handshake and // the login node would see a session per connection. assert.Contains(t, manifest, "ssh -M -N -o ControlMaster=yes") @@ -280,7 +281,7 @@ func TestSSHShadowTemplate(t *testing.T) { {Port: 9999, Name: "telemetry", Protocol: "UDP"}, }) - assert.Contains(t, manifest, "-L 0.0.0.0:8888:$node:8888") + assert.Contains(t, manifest, `-L "0.0.0.0:8888:$node:8888"`) assert.NotContains(t, manifest, "9999") }) diff --git a/pkg/virtualkubelet/shadow_test.go b/pkg/virtualkubelet/shadow_test.go index 4cb4dd73..bd573db1 100644 --- a/pkg/virtualkubelet/shadow_test.go +++ b/pkg/virtualkubelet/shadow_test.go @@ -1,6 +1,7 @@ package virtualkubelet import ( + "strings" "testing" "github.com/stretchr/testify/assert" @@ -154,3 +155,65 @@ func TestCleanupShadowResourcesRemovesNodeConfigMap(t *testing.T) { _, err := client.CoreV1().ConfigMaps(ns).Get(t.Context(), shadowNodeConfigMapName(name), metav1.GetOptions{}) assert.True(t, apierrors.IsNotFound(err), "the node configmap should be deleted with the rest of the shadow") } + +// TestIsValidComputeNodeName pins what a plugin is allowed to report. The shadow +// puts this in front of ssh, so the interesting cases are the ones that would +// change the meaning of a command rather than merely look odd. +func TestIsValidComputeNodeName(t *testing.T) { + valid := []string{ + "j18n3", + "j18n3.liza.surf.nl", + "hpc-cloud-test001.cern.ch", + "as01r2b14", + "10.0.0.7", + "fe80::1", + "node_7", + } + for _, name := range valid { + assert.True(t, isValidComputeNodeName(name), "expected %q to be accepted", name) + } + + hostile := map[string]string{ + "whitespace injects an ssh option": "h -oProxyCommand=touch", + "tab does the same": "h\t-oProxyCommand=touch", + "newline": "h\nsecond", + "quote breaks the relay command": `x" ; touch /tmp/pwned ; echo "`, + "command substitution": "x$(id)y", + "backticks": "x`id`y", + "semicolon": "h;id", + "glob": "h*", + "leading hyphen reads as a flag": "-oProxyCommand=touch", + "empty": "", + } + for why, name := range hostile { + assert.False(t, isValidComputeNodeName(name), "expected %q to be refused (%s)", name, why) + } + + assert.False(t, isValidComputeNodeName(strings.Repeat("a", maxComputeNodeNameLen+1)), + "a name longer than a DNS name is not one") +} + +// TestPublishShadowNodeNameRefusesHostileNames makes sure a bad value never reaches +// the ConfigMap the shadow mounts, so neither forward mode can be made to run +// something other than ssh. +func TestPublishShadowNodeNameRefusesHostileNames(t *testing.T) { + pod := podWithPort("nb", testNamespaceDefault, "uid-hostile") + identity, err := computeShadowResourceIdentity(pod) + assert.NoError(t, err) + cmName := shadowNodeConfigMapName(identity.Name) + + for _, nodeName := range []string{"h -oProxyCommand=touch", `x" ; id ; echo "`, "x$(id)y"} { + client := fake.NewSimpleClientset(&v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: cmName, Namespace: identity.Namespace}, + Data: map[string]string{shadowNodeNameKey: ""}, + }) + p := tunnelProvider(client) + + p.publishShadowNodeName(t.Context(), pod, nodeName) + + assert.Empty(t, configMapPatches(client), "must not patch for %q", nodeName) + cm, err := client.CoreV1().ConfigMaps(identity.Namespace).Get(t.Context(), cmName, metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, "", cm.Data[shadowNodeNameKey]) + } +} diff --git a/pkg/virtualkubelet/templates/shadow-ssh-template.yaml b/pkg/virtualkubelet/templates/shadow-ssh-template.yaml index c52c4d09..159a016a 100644 --- a/pkg/virtualkubelet/templates/shadow-ssh-template.yaml +++ b/pkg/virtualkubelet/templates/shadow-ssh-template.yaml @@ -168,6 +168,9 @@ spec: ssh -M -N -o ControlMaster=yes -o ControlPath="$control" -o ControlPersist=no $ssh_opts "$target" & pids=$! + # Quoted for the login node's shell, which re-parses whatever ssh sends it. + remote_cmd="$connect_cmd '$node'" + waited=0 # shellcheck disable=SC2086 until ssh -O check -o ControlPath="$control" $ssh_opts "$target" >/dev/null 2>&1; do @@ -181,8 +184,8 @@ spec: {{- range .ExposedPorts }} {{- if ne .Protocol "UDP" }} - printf '#!/bin/sh\nexec ssh -q -o ControlMaster=auto -o ControlPath=%s %s %s "%s %s {{.Port}}"\n' \ - "$control" "$ssh_opts" "$target" "$connect_cmd" "$node" > "$forward_dir/{{.Port}}" + printf '#!/bin/sh\nexec ssh -q -o ControlMaster=auto -o ControlPath=%s %s %s "%s {{.Port}}"\n' \ + "$control" "$ssh_opts" "$target" "$remote_cmd" > "$forward_dir/{{.Port}}" chmod +x "$forward_dir/{{.Port}}" socat TCP-LISTEN:{{.Port}},fork,reuseaddr,bind=0.0.0.0 EXEC:"$forward_dir/{{.Port}}" & pids="$pids $!" @@ -197,7 +200,7 @@ spec: ssh -N -o ExitOnForwardFailure=yes $ssh_opts \ {{- range .ExposedPorts }} {{- if ne .Protocol "UDP" }} - -L 0.0.0.0:{{.Port}}:$node:{{.Port}} \ + -L "0.0.0.0:{{.Port}}:$node:{{.Port}}" \ {{- end }} {{- end }} "$target" & From 04020c11bd00fa7227764a5148281a38339efb40 Mon Sep 17 00:00:00 2001 From: Matteo Bunino <48362942+matbun@users.noreply.github.com> Date: Fri, 31 Jul 2026 15:50:49 +0200 Subject: [PATCH 7/7] fix: do not overwrite a credential the shadow namespace already had Credential replication created-or-updated unconditionally. With interlink.eu/shadow-same-ns the shadow lands in the offloaded pod's own namespace, which on a multi-tenant cluster is somebody's personal namespace, so a Secret that merely shares the configured name was replaced with the HPC key and its contents lost. Nothing said so. Mark replicated objects with interlink.eu/replicated-from and only overwrite ones carrying it. A pre-existing object without it fails pod creation with an error naming the collision and the two ways out, which is recoverable; silently destroying a user's Secret is not. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Matteo Bunino <48362942+matbun@users.noreply.github.com> --- pkg/virtualkubelet/shadow_ssh.go | 73 ++++++++++++++++++++++++++- pkg/virtualkubelet/shadow_ssh_test.go | 63 +++++++++++++++++++++++ 2 files changed, 134 insertions(+), 2 deletions(-) diff --git a/pkg/virtualkubelet/shadow_ssh.go b/pkg/virtualkubelet/shadow_ssh.go index 19a2d829..c76c33bf 100644 --- a/pkg/virtualkubelet/shadow_ssh.go +++ b/pkg/virtualkubelet/shadow_ssh.go @@ -3,11 +3,13 @@ package virtualkubelet import ( "context" "fmt" + "reflect" "strings" "time" "github.com/containerd/containerd/log" v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -182,13 +184,65 @@ func (p *Provider) replicateShadowCredentials(ctx context.Context, identity shad return nil } +// shadowReplicatedFromAnnotation marks an object as a copy this provider made into +// a shadow namespace. Replication only overwrites objects carrying it, so a Secret +// the namespace's owner already had under the same name is never destroyed. +const shadowReplicatedFromAnnotation = "interlink.eu/replicated-from" + +// replicationTarget describes an object already present in the target namespace. +type replicationTarget struct { + exists bool + annotations map[string]string + sameContent bool +} + +// checkReplicationTarget refuses to overwrite an object that interLink did not put +// there. With interlink.eu/shadow-same-ns the target is the offloaded pod's own +// namespace, which for a multi-tenant cluster is somebody's personal namespace, so +// a name collision would otherwise silently replace their data. +// +// An unmarked object whose content already matches is adopted rather than refused: +// that is what a copy made by a version predating the marker looks like, and +// rewriting it with identical content loses nothing. +func checkReplicationTarget(kind, target, name string, found replicationTarget) error { + if !found.exists || found.annotations[shadowReplicatedFromAnnotation] != "" || found.sameContent { + return nil + } + return fmt.Errorf( + "refusing to overwrite %s %s/%s: it already exists, holds different content and was not created by interLink. "+ + "Rename the credential, or set Network.SSH.ReplicateCredentials to false and provision it yourself", + kind, target, name) +} + +func replicatedMeta(name, target, source string) metav1.ObjectMeta { + return metav1.ObjectMeta{ + Name: name, + Namespace: target, + Annotations: map[string]string{shadowReplicatedFromAnnotation: source}, + } +} + func (p *Provider) replicateSecret(ctx context.Context, source, target, name string) error { src, err := p.clientSet.CoreV1().Secrets(source).Get(ctx, name, metav1.GetOptions{}) if err != nil { return fmt.Errorf("failed to read SSH credential secret %s/%s: %w", source, name, err) } + + existing, err := p.clientSet.CoreV1().Secrets(target).Get(ctx, name, metav1.GetOptions{}) + if err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("failed to inspect secret %s/%s: %w", target, name, err) + } + found := replicationTarget{exists: err == nil} + if found.exists { + found.annotations = existing.Annotations + found.sameContent = existing.Type == src.Type && reflect.DeepEqual(existing.Data, src.Data) + } + if err := checkReplicationTarget("secret", target, name, found); err != nil { + return err + } + copied := &v1.Secret{ - ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: target}, + ObjectMeta: replicatedMeta(name, target, source), Type: src.Type, Data: src.Data, } @@ -204,8 +258,23 @@ func (p *Provider) replicateConfigMap(ctx context.Context, source, target, name if err != nil { return fmt.Errorf("failed to read configmap %s/%s: %w", source, name, err) } + + existing, err := p.clientSet.CoreV1().ConfigMaps(target).Get(ctx, name, metav1.GetOptions{}) + if err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("failed to inspect configmap %s/%s: %w", target, name, err) + } + found := replicationTarget{exists: err == nil} + if found.exists { + found.annotations = existing.Annotations + found.sameContent = reflect.DeepEqual(existing.Data, src.Data) && + reflect.DeepEqual(existing.BinaryData, src.BinaryData) + } + if err := checkReplicationTarget("configmap", target, name, found); err != nil { + return err + } + copied := &v1.ConfigMap{ - ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: target}, + ObjectMeta: replicatedMeta(name, target, source), Data: src.Data, BinaryData: src.BinaryData, } diff --git a/pkg/virtualkubelet/shadow_ssh_test.go b/pkg/virtualkubelet/shadow_ssh_test.go index 3abb0564..86c50de4 100644 --- a/pkg/virtualkubelet/shadow_ssh_test.go +++ b/pkg/virtualkubelet/shadow_ssh_test.go @@ -373,3 +373,66 @@ func TestDefaultSSHNodeWaitMatchesTimeout(t *testing.T) { assert.NoError(t, err) assert.Equal(t, parsed, defaultSSHNodeWait) } + +// TestReplicateCredentialsRefusesToClobber covers the multi-tenant case: with +// interlink.eu/shadow-same-ns the shadow lands in the offloaded pod's own +// namespace, so a Secret the owner already has under the credential's name must +// not be silently replaced with the HPC key. +func TestReplicateCredentialsRefusesToClobber(t *testing.T) { + const source = "interlink" + const target = "alice" + + config := normalized(t, sshConfig()) + credential := &v1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "hpc-ssh-key", Namespace: source}, + Data: map[string][]byte{DefaultSSHKeySecretKey: []byte("hpc-private-key")}, + } + + t.Run("a secret the user already owns is left alone", func(t *testing.T) { + theirs := &v1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "hpc-ssh-key", Namespace: target}, + Data: map[string][]byte{"theirs": []byte("do not lose me")}, + } + client := fake.NewSimpleClientset(credential, theirs) + p := &Provider{clientSet: client, config: config} + + err := p.replicateShadowCredentials(t.Context(), shadowResourceIdentity{Name: "shadow-nb", Namespace: target}) + + assert.ErrorContains(t, err, "refusing to overwrite") + kept, getErr := client.CoreV1().Secrets(target).Get(t.Context(), "hpc-ssh-key", metav1.GetOptions{}) + assert.NoError(t, getErr) + assert.Equal(t, []byte("do not lose me"), kept.Data["theirs"]) + }) + + // A copy made by a version that predates the marker has no annotation, but its + // content matches, so upgrading must not start failing every offloaded pod. + t.Run("an unmarked copy with identical content is adopted", func(t *testing.T) { + unmarked := &v1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "hpc-ssh-key", Namespace: target}, + Data: map[string][]byte{DefaultSSHKeySecretKey: []byte("hpc-private-key")}, + } + client := fake.NewSimpleClientset(credential, unmarked) + p := &Provider{clientSet: client, config: config} + + err := p.replicateShadowCredentials(t.Context(), shadowResourceIdentity{Name: "shadow-nb", Namespace: target}) + + assert.NoError(t, err, "an identical copy loses nothing by being rewritten") + adopted, getErr := client.CoreV1().Secrets(target).Get(t.Context(), "hpc-ssh-key", metav1.GetOptions{}) + assert.NoError(t, getErr) + assert.Equal(t, source, adopted.Annotations[shadowReplicatedFromAnnotation]) + }) + + t.Run("a copy interLink made earlier is refreshed", func(t *testing.T) { + client := fake.NewSimpleClientset(credential) + p := &Provider{clientSet: client, config: config} + identity := shadowResourceIdentity{Name: "shadow-nb", Namespace: target} + + assert.NoError(t, p.replicateShadowCredentials(t.Context(), identity)) + first, err := client.CoreV1().Secrets(target).Get(t.Context(), "hpc-ssh-key", metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, source, first.Annotations[shadowReplicatedFromAnnotation]) + + // a second offloaded pod in the same namespace must not trip the guard + assert.NoError(t, p.replicateShadowCredentials(t.Context(), identity)) + }) +}