diff --git a/cmd/virtual-kubelet/main.go b/cmd/virtual-kubelet/main.go index 24eb99f8..56d6caea 100644 --- a/cmd/virtual-kubelet/main.go +++ b/cmd/virtual-kubelet/main.go @@ -61,6 +61,7 @@ import ( "k8s.io/apimachinery/pkg/fields" "k8s.io/client-go/kubernetes/scheme" lease "k8s.io/client-go/kubernetes/typed/coordination/v1" + typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" "k8s.io/client-go/tools/cache" "k8s.io/client-go/tools/clientcmd" "k8s.io/client-go/tools/record" @@ -541,7 +542,10 @@ func main() { }() eb := record.NewBroadcaster() + eb.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: localClient.CoreV1().Events("")}) + defer eb.Shutdown() EventRecorder := eb.NewRecorder(scheme.Scheme, v1.EventSource{Component: path.Join(cfg.NodeName, "pod-controller")}) + nodeProvider.SetEventRecorder(EventRecorder) podInformerFactory, scmInformerFactory, stopper := setupInformers(ctx, localClient, cfg.NodeName) defer close(stopper) diff --git a/pkg/virtualkubelet/config.go b/pkg/virtualkubelet/config.go index a0b357ac..658ea6e7 100644 --- a/pkg/virtualkubelet/config.go +++ b/pkg/virtualkubelet/config.go @@ -254,8 +254,7 @@ const ( // ForwardTarget selects where the forwarding session ends (the login node, or the // compute node's own sshd) and ForwardMode whether sshd forwards or session // commands relay. 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). +// direction; Reverse gives the offloaded pod access to listed cluster endpoints. type SSHTunnel struct { // LoginHost is the SSH login node to forward through (required) LoginHost string `yaml:"LoginHost,omitempty"` @@ -300,7 +299,8 @@ type SSHTunnel struct { // ExecConnectCommand is the relay command of the "exec" mode, run on the login // node. It must relay a connection on its stdin and stdout, and defaults to // "nc": ` ` with ForwardTarget "login", - // ` 22` with "node". + // ` 22` with "node". Reverse.NodeExec "srun" also runs + // ` 127.0.0.1 22` inside the job's allocation. ExecConnectCommand string `yaml:"ExecConnectCommand,omitempty"` // NodeHostKeys selects how compute node host keys are checked on every session // that ends on a compute node: "verify" (default) checks them against @@ -347,4 +347,42 @@ type SSHTunnel struct { // exception: GSSAPIDelegateCredentials=yes restores Kerberos delegation on the // login hop. ExtraOptions []string `yaml:"ExtraOptions,omitempty"` + // Reverse lets offloaded pods open connections back into the cluster through + // the same login node, see the interlink.eu/reverse-* annotations. + Reverse SSHReverse `yaml:"Reverse,omitempty"` +} + +// SSHReverse configures connections opened by offloaded pods towards cluster +// endpoints. Each forward is made by the compute node's own sshd, or by a +// user-space sshd as a fallback, with `ssh -R` to a unix socket in the pod's socket +// directory; a relay the job starts puts the pod's loopback address in front of +// those sockets and accepts only the job's user. Nothing on the compute node has +// to reach the cluster. +type SSHReverse struct { + // Enabled lets pods ask for it (default false). A pod that asks while this is + // false is refused at creation rather than run without its tunnel. + Enabled bool `yaml:"Enabled,omitempty"` + // Sshd selects which sshd makes the forwards on the compute node: "node" + // (default) is the node's own sshd, reached like ForwardTarget "node" and + // checked against NodeHostKeys; "userspace" is an sshd the shadow starts in the + // job as the job's user, for sites whose node sshd refuses forwarding. + Sshd string `yaml:"Sshd,omitempty"` + // AllowSocks lets pods ask for a SOCKS5 proxy with + // interlink.eu/reverse-socks-port (default false). The proxy listens on + // 127.0.0.1 of the compute node, where every user of that node can reach + // through it any endpoint the shadow reaches in the cluster. + AllowSocks bool `yaml:"AllowSocks,omitempty"` + // NodeExec is how the shadow reaches the compute node: "ssh" (default) opens + // node:22 through the login node as ForwardMode says, which the site must + // allow into a node where the user has a job; "srun" runs `srun + // --jobid= --overlap` on the login node, inside the job's own allocation. + // srun is a last resort: lower throughput, and with Kerberos it delegates the + // ticket to the login node. Set here only, never per pod. + NodeExec string `yaml:"NodeExec,omitempty"` + // SshdPath is the sshd binary on the compute node, for Sshd "userspace" only + // (default "/usr/sbin/sshd") + SshdPath string `yaml:"SshdPath,omitempty"` + // PythonPath is the python3 (3.6 or later) on the compute node that runs the + // job's owner-only relay, isolated with -I (default "python3") + PythonPath string `yaml:"PythonPath,omitempty"` } diff --git a/pkg/virtualkubelet/shadow_ssh.go b/pkg/virtualkubelet/shadow_ssh.go index c68e7c9a..ee56f898 100644 --- a/pkg/virtualkubelet/shadow_ssh.go +++ b/pkg/virtualkubelet/shadow_ssh.go @@ -62,7 +62,7 @@ func NormalizeShadowConfig(config *Config) error { // 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) + return fmt.Errorf("Network.FullMesh is not supported with ShadowMode %q: the SSH shadow has no mesh; pods reach cluster endpoints with Network.SSH.Reverse (see interlink-hq/interLink#548)", ShadowModeSSH) } s := &n.SSH @@ -90,6 +90,10 @@ func NormalizeShadowConfig(config *Config) error { s.ReplicateCredentials = &replicate } + if err := normalizeReverseConfig(s); err != nil { + return err + } + s.ForwardMode = strings.ToLower(strings.TrimSpace(s.ForwardMode)) switch s.ForwardMode { case "": @@ -160,10 +164,44 @@ func NormalizeShadowConfig(config *Config) error { return nil } +// normalizeReverseConfig defaults and validates Network.SSH.Reverse. +func normalizeReverseConfig(s *SSHTunnel) error { + r := &s.Reverse + r.NodeExec = strings.ToLower(strings.TrimSpace(r.NodeExec)) + switch r.NodeExec { + case "": + r.NodeExec = SSHReverseNodeExecSSH + case SSHReverseNodeExecSSH, SSHReverseNodeExecSrun: + default: + return fmt.Errorf("unknown Network.SSH.Reverse.NodeExec %q: expected %q or %q", r.NodeExec, SSHReverseNodeExecSSH, SSHReverseNodeExecSrun) + } + if r.NodeExec == SSHReverseNodeExecSrun && strings.TrimSpace(s.ExecConnectCommand) == "" { + s.ExecConnectCommand = DefaultSSHExecConnectCommand + } + r.Sshd = strings.ToLower(strings.TrimSpace(r.Sshd)) + switch r.Sshd { + case "": + r.Sshd = SSHReverseSshdNode + case SSHReverseSshdNode, SSHReverseSshdUserspace: + default: + return fmt.Errorf("unknown Network.SSH.Reverse.Sshd %q: expected %q or %q", r.Sshd, SSHReverseSshdNode, SSHReverseSshdUserspace) + } + if r.SshdPath == "" { + r.SshdPath = DefaultSSHReverseSshdPath + } + if r.PythonPath == "" { + r.PythonPath = DefaultSSHReversePythonPath + } + return nil +} + // usesNodeHops reports whether the shadow opens ssh sessions on compute nodes, -// which is what NodeHostKeys governs. +// which is what NodeHostKeys governs. Only a user-space sshd started through srun +// avoids them: its host key is one the shadow generated. func usesNodeHops(s SSHTunnel) bool { - return s.ForwardTarget == SSHForwardTargetNode + r := s.Reverse + return s.ForwardTarget == SSHForwardTargetNode || + (r.Enabled && (r.Sshd == SSHReverseSshdNode || r.NodeExec == SSHReverseNodeExecSSH)) } // sshNodeWaitSeconds converts the configured wait into whole seconds for the shell diff --git a/pkg/virtualkubelet/shadow_ssh_reverse.go b/pkg/virtualkubelet/shadow_ssh_reverse.go new file mode 100644 index 00000000..d6cfc1ec --- /dev/null +++ b/pkg/virtualkubelet/shadow_ssh_reverse.go @@ -0,0 +1,371 @@ +package virtualkubelet + +import ( + "context" + _ "embed" + "encoding/json" + "fmt" + "hash/fnv" + "net" + "regexp" + "strconv" + "strings" + + "github.com/containerd/containerd/log" + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + k8stypes "k8s.io/apimachinery/pkg/types" +) + +// Pod annotations asking the ssh shadow for connections back into the cluster. +const ( + // annReverseForwards lists the cluster endpoints the pod may open connections + // to, space separated, each "host:port" or "listen-port:host:port". + annReverseForwards = "interlink.eu/reverse-forwards" + // annReverseSocksPort asks for a SOCKS5 proxy on the given port for endpoints + // not listed in annReverseForwards. + annReverseSocksPort = "interlink.eu/reverse-socks-port" + // annReverseBind overrides the 127.x.y.z address of the owner-only relay + // inside the job, normally derived from the pod UID. + annReverseBind = "interlink.eu/reverse-bind" +) + +// How the shadow reaches the compute node. +const ( + // SSHReverseNodeExecSSH opens node:22 through the login node, as + // Network.SSH.ForwardMode says. Needs the site to allow ssh into a node where + // the user has a running job. + SSHReverseNodeExecSSH = "ssh" + // SSHReverseNodeExecSrun runs `srun --jobid= --overlap` on the login node, + // inside the job's own allocation. A last resort: lower throughput, and the + // login node needs the user's Kerberos ticket where there is one. + SSHReverseNodeExecSrun = "srun" +) + +// Which sshd makes the forwards on the compute node. +const ( + // SSHReverseSshdNode is the compute node's own sshd. + SSHReverseSshdNode = "node" + // SSHReverseSshdUserspace is an sshd the shadow starts in the job, as the + // job's user, with its own config, for sites whose node sshd refuses + // forwarding. + SSHReverseSshdUserspace = "userspace" +) + +const ( + // DefaultSSHReverseSshdPath is where sshd is looked for on the compute node. + DefaultSSHReverseSshdPath = "/usr/sbin/sshd" + // DefaultSSHReversePythonPath runs the job's owner-only relay. + DefaultSSHReversePythonPath = "python3" + // reverseTunnelWaitSeconds bounds how long the job waits for its tunnel before + // giving up. The tunnel follows the compute node name, which the shadow learns a + // status poll or two after the job starts. + reverseTunnelWaitSeconds = 600 + // reversePrivilegedPortLimit: an unprivileged sshd cannot bind below it. + reversePrivilegedPortLimit = 1024 +) + +// ReverseForward is one cluster endpoint the job may connect to, as the shadow +// template sees it: the job dials ListenPort on its loopback address, the shadow +// dials Host:Port. +type ReverseForward struct { + ListenPort int32 + Host string + Port int32 +} + +// ReverseTunnel carries the pod's reverse settings into the ssh shadow template. +// Everything in it has been validated: the template passes it to ssh unquoted. +type ReverseTunnel struct { + Forwards []ReverseForward + // SocksPort is 0 when no SOCKS proxy was asked for. + SocksPort int32 + // Bind is the 127.x.y.z address the job's relay listens on. + Bind string + // NodeExec is SSHReverseNodeExecSSH or SSHReverseNodeExecSrun. + NodeExec string + // Sshd is SSHReverseSshdNode or SSHReverseSshdUserspace. + Sshd string + SshdPath string + PythonPath string + // JobName is how srun mode finds the allocation: the plugin names the batch + // job after the pod UID. + JobName string +} + +var reverseHostPattern = regexp.MustCompile(`^[A-Za-z0-9]([A-Za-z0-9.-]*[A-Za-z0-9])?$`) + +// hasReverseAnnotations reports whether the pod asked for any reverse forward. +func hasReverseAnnotations(pod *v1.Pod) bool { + if pod == nil || pod.Annotations == nil { + return false + } + return strings.TrimSpace(pod.Annotations[annReverseForwards]) != "" || + strings.TrimSpace(pod.Annotations[annReverseSocksPort]) != "" +} + +// reverseRequested reports whether the pod gets a reverse tunnel: it asked for +// one and this deployment can provide it. +func (p *Provider) reverseRequested(pod *v1.Pod) bool { + return p.isSSHShadow() && p.config.Network.SSH.Reverse.Enabled && hasReverseAnnotations(pod) +} + +// checkReverseRequest refuses, at creation, a pod that asks for the reverse +// tunnel on a deployment that cannot provide it. Silently running it without the +// tunnel would only move the failure to the first connection the job opens. +func (p *Provider) checkReverseRequest(pod *v1.Pod) error { + if !hasReverseAnnotations(pod) { + return nil + } + switch { + case !p.isSSHShadow(): + return fmt.Errorf("pod %s/%s sets %s, which needs Network.ShadowMode %q", pod.Namespace, pod.Name, annReverseForwards, ShadowModeSSH) + case !p.config.Network.EnableTunnel: + return fmt.Errorf("pod %s/%s sets %s, but Network.EnableTunnel is false", pod.Namespace, pod.Name, annReverseForwards) + case !p.config.Network.SSH.Reverse.Enabled: + return fmt.Errorf("pod %s/%s sets %s, but this virtual kubelet does not allow it (Network.SSH.Reverse.Enabled)", pod.Namespace, pod.Name, annReverseForwards) + case pod.Annotations["interlink.eu/pod-vpn"] != "": + return fmt.Errorf("pod %s/%s sets %s, but interlink.eu/pod-vpn disables its shadow tunnel", pod.Namespace, pod.Name, annReverseForwards) + case strings.TrimSpace(pod.Annotations[annReverseSocksPort]) != "" && !p.config.Network.SSH.Reverse.AllowSocks: + return fmt.Errorf("pod %s/%s sets %s, but this virtual kubelet does not allow a SOCKS proxy (Network.SSH.Reverse.AllowSocks)", pod.Namespace, pod.Name, annReverseSocksPort) + } + return nil +} + +// warnSocksProxy tells the pod's owner, in the VK log and as an Event on the pod, +// that its SOCKS proxy is open to every user of the compute node. +func (p *Provider) warnSocksProxy(ctx context.Context, pod *v1.Pod, port int32) { + msg := fmt.Sprintf("SOCKS proxy on 127.0.0.1:%d on the compute node: every user of that node can reach through it any endpoint the shadow reaches in the cluster", port) + log.G(ctx).Warningf("pod %s/%s: %s", pod.Namespace, pod.Name, msg) + if p.eventRecorder != nil { + p.eventRecorder.Event(pod, v1.EventTypeWarning, "ReverseSocksProxy", msg) + } +} + +// parseReverseTunnel turns the pod's reverse annotations into template data, +// rejecting anything that could not work or that the template could not pass to +// ssh safely. +func parseReverseTunnel(pod *v1.Pod, cfg SSHReverse) (*ReverseTunnel, error) { + ann := pod.Annotations + rt := &ReverseTunnel{ + NodeExec: cfg.NodeExec, + Sshd: cfg.Sshd, + SshdPath: cfg.SshdPath, + PythonPath: cfg.PythonPath, + JobName: string(pod.UID), + Bind: reverseBindAddress(string(pod.UID)), + } + + listening := map[int32]string{} + for _, spec := range strings.Fields(ann[annReverseForwards]) { + f, err := parseReverseForward(spec) + if err != nil { + return nil, fmt.Errorf("%s: %w", annReverseForwards, err) + } + if other, taken := listening[f.ListenPort]; taken { + return nil, fmt.Errorf("%s: %q and %q both listen on port %d", annReverseForwards, other, spec, f.ListenPort) + } + listening[f.ListenPort] = spec + rt.Forwards = append(rt.Forwards, f) + } + + if v := strings.TrimSpace(ann[annReverseSocksPort]); v != "" { + port, err := parseReversePort(v) + if err != nil { + return nil, fmt.Errorf("%s: %w", annReverseSocksPort, err) + } + if port < reversePrivilegedPortLimit { + return nil, fmt.Errorf("%s: port %d is below %d, which an unprivileged sshd cannot bind on the compute node", annReverseSocksPort, port, reversePrivilegedPortLimit) + } + if other, taken := listening[port]; taken { + return nil, fmt.Errorf("%s: port %d is already used by forward %q", annReverseSocksPort, port, other) + } + rt.SocksPort = port + } + + if v := strings.TrimSpace(ann[annReverseBind]); v != "" { + ip := net.ParseIP(v).To4() + if ip == nil || ip[0] != 127 || v == "127.0.0.1" { + return nil, fmt.Errorf("%s: %q is not a 127.x.y.z address other than 127.0.0.1", annReverseBind, v) + } + rt.Bind = ip.String() + } + + return rt, nil +} + +// parseReverseForward parses "host:port" or "listen-port:host:port". The listen +// port defaults to the destination port, which the job then uses unchanged; a +// destination port below 1024 needs an explicit listen port. +func parseReverseForward(spec string) (ReverseForward, error) { + parts := strings.Split(spec, ":") + var f ReverseForward + var err error + switch len(parts) { + case 2: + f.Host = parts[0] + if f.Port, err = parseReversePort(parts[1]); err != nil { + return f, fmt.Errorf("%q: %w", spec, err) + } + f.ListenPort = f.Port + case 3: + if f.ListenPort, err = parseReversePort(parts[0]); err != nil { + return f, fmt.Errorf("%q: listen %w", spec, err) + } + f.Host = parts[1] + if f.Port, err = parseReversePort(parts[2]); err != nil { + return f, fmt.Errorf("%q: %w", spec, err) + } + default: + return f, fmt.Errorf("%q: expected host:port or listen-port:host:port", spec) + } + if len(f.Host) > 253 || !reverseHostPattern.MatchString(f.Host) { + return f, fmt.Errorf("%q: %q is not a hostname or IP address", spec, f.Host) + } + if f.ListenPort < reversePrivilegedPortLimit { + return f, fmt.Errorf("%q: port %d is below %d, which an unprivileged sshd cannot bind on the compute node; use listen-port:host:port with a higher listen port", spec, f.ListenPort, reversePrivilegedPortLimit) + } + return f, nil +} + +func parseReversePort(s string) (int32, error) { + port, err := strconv.ParseInt(s, 10, 32) + if err != nil || port < 1 || port > 65535 { + return 0, fmt.Errorf("port %q is not in 1..65535", s) + } + return int32(port), nil +} + +// reverseBindAddress picks the loopback address a pod's forwards listen on inside +// the job. Two pods of the same user on one compute node must not share it, so it +// is derived from the pod UID rather than fixed; 127.0.0.0/24 is left alone. +func reverseBindAddress(podUID string) string { + h := fnv.New32a() + h.Write([]byte(podUID)) + s := h.Sum32() + return fmt.Sprintf("127.%d.%d.%d", 1+(s>>16)%254, (s>>8)&0xff, 1+(s&0xff)%254) +} + +// reverseHostAliases lists the names a forward's host resolves to inside the job: +// the host as given plus, for a Service FQDN, the shorter forms cluster DNS would +// also answer. The bare Service name only holds in the pod's own namespace. +func reverseHostAliases(host, podNamespace string) []string { + names := []string{host} + parts := strings.Split(host, ".") + if len(parts) >= 4 && parts[2] == "svc" { + name, ns := parts[0], parts[1] + names = append(names, name+"."+ns+".svc", name+"."+ns) + if ns == podNamespace { + names = append(names, name) + } + } + return names +} + +// reverseRelay is the owner-only relay the job runs in front of the tunnel's unix +// sockets. +// +//go:embed templates/reverse-relay.py +var reverseRelay string + +// reversePreExec is the shell the job runs before its workload. It starts the +// relay that holds the pod's loopback address for the whole job, maps the +// forwarded names to that address in the container's /etc/hosts, and waits until +// the shadow has bound every forward in the socket directory, which happens only +// once the shadow has learned which node the job landed on. A SOCKS proxy is +// waited for on 127.0.0.1 and must belong to the job's user. It assumes a bash +// batch script and a Singularity or Apptainer runtime, as the mesh pre-exec does. +// +// The hosts file stays in node-local temporary space and never goes into +// $INTERLINK_SOCKET_DIR: only the batch node runs the relay, so ranks on other +// nodes must not resolve the names to an address nobody holds there. +// reverseHostsBind mounts the hosts file for Singularity and Apptainer. Apptainer +// reads SINGULARITY_BIND only while APPTAINER_BIND is unset, so the job's own +// binds from the legacy variable are carried into APPTAINER_BIND. +const reverseHostsBind = `il_rev_ab=${APPTAINER_BIND:-${SINGULARITY_BIND:-}} +export SINGULARITY_BIND="${SINGULARITY_BIND:+$SINGULARITY_BIND,}$il_rev_hosts:/etc/hosts" +export APPTAINER_BIND="${il_rev_ab:+$il_rev_ab,}$il_rev_hosts:/etc/hosts" +` + +func reversePreExec(rt *ReverseTunnel, podNamespace string) string { + seen := map[string]bool{} + var hosts []string + for _, f := range rt.Forwards { + for _, name := range reverseHostAliases(f.Host, podNamespace) { + if !seen[name] { + seen[name] = true + hosts = append(hosts, name) + } + } + } + timeout := fmt.Sprintf("[ $(( $(date +%%s) - il_rev_t0 )) -ge %d ] && { echo 'interLink: the reverse tunnel did not come up within %ds' >&2; exit 1; }\n", + reverseTunnelWaitSeconds, reverseTunnelWaitSeconds) + + var b strings.Builder + b.WriteString("# interLink reverse tunnel: cluster endpoints reachable from this job\n") + b.WriteString("il_rev_d=${INTERLINK_SOCKET_DIR:-}\n") + b.WriteString("[ -n \"$il_rev_d\" ] || { echo 'interLink: the reverse tunnel needs a plugin with SocketDirRoot set' >&2; exit 1; }\n") + b.WriteString("il_rev_t0=$(date +%s)\n") + if len(rt.Forwards) > 0 { + fmt.Fprintf(&b, "# Node-local on purpose: only the batch node has a relay behind %s.\n", rt.Bind) + b.WriteString("il_rev_hosts=$(mktemp \"${TMPDIR:-/tmp}/interlink-hosts.XXXXXX\") || exit 1\n") + fmt.Fprintf(&b, "{ cat /etc/hosts; echo '%s %s'; } > \"$il_rev_hosts\"\n", rt.Bind, strings.Join(hosts, " ")) + b.WriteString(reverseHostsBind) + b.WriteString("cat > \"$il_rev_d/rev-relay.py\" <<'EORELAY'\n") + b.WriteString(reverseRelay) + if !strings.HasSuffix(reverseRelay, "\n") { + b.WriteString("\n") + } + b.WriteString("EORELAY\n") + // -I: a module loaded by the batch script must not reach the relay through + // PYTHONPATH or PYTHONHOME. + fmt.Fprintf(&b, "%s -I \"$il_rev_d/rev-relay.py\" %s", rt.PythonPath, rt.Bind) + var ready []string + for _, f := range rt.Forwards { + sock := fmt.Sprintf("\"$il_rev_d/rev-%d.sock\"", f.ListenPort) + fmt.Fprintf(&b, " %d=%s", f.ListenPort, sock) + ready = append(ready, "[ -S "+sock+" ]") + } + b.WriteString(" &\n") + b.WriteString("il_rev_relay=$!\n") + fmt.Fprintf(&b, "until [ -e \"$il_rev_d/rev-relay.ready\" ] && %s; do\n", strings.Join(ready, " && ")) + b.WriteString(" kill -0 \"$il_rev_relay\" 2>/dev/null || { echo 'interLink: the reverse relay stopped, see above' >&2; exit 1; }\n") + b.WriteString(" " + timeout) + b.WriteString(" sleep 2\n") + b.WriteString("done\n") + } + if rt.SocksPort != 0 { + // Anyone on the node may bind 127.0.0.1 first: wait for the listener to be + // the job user's own, and fail if it is somebody else's. + port := fmt.Sprintf("%04X", rt.SocksPort) + b.WriteString("il_rev_uid=$(id -u)\n") + b.WriteString(`il_rev_owner() { awk -v p=":$1" -v u="$il_rev_uid" 'NR>1 && $4=="0A" && ($2=="0100007F"p || $2=="00000000"p) {print ($8==u ? "own" : "other"); exit}' /proc/net/tcp; }` + "\n") + fmt.Fprintf(&b, "until [ \"$(il_rev_owner %s)\" = own ]; do\n", port) + fmt.Fprintf(&b, " [ \"$(il_rev_owner %s)\" = other ] && { echo 'interLink: another user listens on the SOCKS port' >&2; exit 1; }\n", port) + b.WriteString(" " + timeout) + b.WriteString(" sleep 2\n") + b.WriteString("done\n") + } + b.WriteString("echo \"interLink: reverse tunnel ready after $(( $(date +%s) - il_rev_t0 ))s\"\n") + return b.String() +} + +// addReversePreExec prepends the tunnel set-up to the pod's pre-exec, the way the +// mesh does, and records it on the Kubernetes object. +func (p *Provider) addReversePreExec(ctx context.Context, pod *v1.Pod, rt *ReverseTunnel) error { + if pod.Annotations == nil { + pod.Annotations = make(map[string]string) + } + pod.Annotations["slurm-job.vk.io/pre-exec"] = reversePreExec(rt, pod.Namespace) + pod.Annotations["slurm-job.vk.io/pre-exec"] + log.G(ctx).Infof("Added reverse tunnel pre-exec to pod %s/%s (%d forwards, socks port %d, bind %s)", + pod.Namespace, pod.Name, len(rt.Forwards), rt.SocksPort, rt.Bind) + + patch, err := json.Marshal(map[string]interface{}{ + "metadata": map[string]interface{}{"annotations": pod.Annotations}, + }) + if err != nil { + return err + } + _, err = p.clientSet.CoreV1().Pods(pod.Namespace).Patch(ctx, pod.Name, k8stypes.StrategicMergePatchType, patch, metav1.PatchOptions{}) + return err +} diff --git a/pkg/virtualkubelet/shadow_ssh_reverse_test.go b/pkg/virtualkubelet/shadow_ssh_reverse_test.go new file mode 100644 index 00000000..c7f0304e --- /dev/null +++ b/pkg/virtualkubelet/shadow_ssh_reverse_test.go @@ -0,0 +1,826 @@ +package virtualkubelet + +import ( + "bytes" + "errors" + "io" + "net" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + 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" + "k8s.io/client-go/tools/record" +) + +const ( + reverseTestNamespace = "user" + reverseTestService = "minio.user.svc.cluster.local" + reverseTestForward = "minio.user.svc:9000" + reverseTestPodVPN = "interlink.eu/pod-vpn" + reverseTestBind = "127.77.1.2" + reverseTestUID = "0d9b2d4a-6f6e-4a5f-9a66-2c1c5a4e9f10" + reverseTestSocks = "1080" + reverseTestContainer = "ssh-reverse" + reverseTestReady = "rev-relay.ready" + reverseTestUserspace = `"%s -i -e -f %s/rev-sshd_config"` +) + +func reverseConfig() Config { + config := sshConfig() + config.Network.SSH.Reverse.Enabled = true + return config +} + +func reversePod(annotations map[string]string) *v1.Pod { + return &v1.Pod{ObjectMeta: metav1.ObjectMeta{ + Name: "trainer", + Namespace: reverseTestNamespace, + UID: reverseTestUID, + Annotations: annotations, + }} +} + +func TestParseReverseTunnel(t *testing.T) { + cfg := normalized(t, reverseConfig()).Network.SSH.Reverse + + t.Run("defaults", func(t *testing.T) { + pod := reversePod(map[string]string{annReverseForwards: reverseTestService + ":9000"}) + rt, err := parseReverseTunnel(pod, cfg) + assert.NoError(t, err) + assert.Equal(t, []ReverseForward{{ListenPort: 9000, Host: reverseTestService, Port: 9000}}, rt.Forwards) + assert.Zero(t, rt.SocksPort, "SOCKS is off unless asked for") + assert.Equal(t, SSHReverseNodeExecSSH, rt.NodeExec) + assert.Equal(t, SSHReverseSshdNode, rt.Sshd) + assert.Equal(t, DefaultSSHReverseSshdPath, rt.SshdPath) + assert.Equal(t, string(pod.UID), rt.JobName) + assert.Equal(t, reverseBindAddress(string(pod.UID)), rt.Bind) + }) + + t.Run("remapped listen port, socks and bind", func(t *testing.T) { + pod := reversePod(map[string]string{ + annReverseForwards: "6443:kubernetes.default.svc.cluster.local:443 redis.user.svc:6379", + annReverseSocksPort: reverseTestSocks, + annReverseBind: reverseTestBind, + }) + rt, err := parseReverseTunnel(pod, cfg) + assert.NoError(t, err) + assert.Equal(t, []ReverseForward{ + {ListenPort: 6443, Host: "kubernetes.default.svc.cluster.local", Port: 443}, + {ListenPort: 6379, Host: "redis.user.svc", Port: 6379}, + }, rt.Forwards) + assert.Equal(t, int32(1080), rt.SocksPort) + assert.Equal(t, reverseTestBind, rt.Bind) + }) + + t.Run("the removed owner-only and node-exec annotations are ignored", func(t *testing.T) { + pod := reversePod(map[string]string{ + annReverseForwards: reverseTestService + ":9000", + "interlink.eu/reverse-owner-only": "false", + "interlink.eu/reverse-node-exec": "srun", + }) + rt, err := parseReverseTunnel(pod, cfg) + assert.NoError(t, err, "pods written for the draft must still start") + assert.Equal(t, SSHReverseNodeExecSSH, rt.NodeExec, "the carrier is the site's decision") + }) + + rejected := []struct { + name string + annotations map[string]string + wantErr string + }{ + {"malformed forward", map[string]string{annReverseForwards: "minio"}, "expected host:port"}, + {"privileged port without a listen port", map[string]string{annReverseForwards: "kubernetes.default.svc:443"}, "below 1024"}, + {"host that is not a name", map[string]string{annReverseForwards: "minio;rm:9000"}, "not a hostname"}, + {"two forwards on one listen port", map[string]string{annReverseForwards: "a.user.svc:9000 9000:b.user.svc:9001"}, "both listen on port 9000"}, + {"socks on a forward's port", map[string]string{annReverseForwards: "a.user.svc:9000", annReverseSocksPort: "9000"}, "already used"}, + {"socks on a privileged port", map[string]string{annReverseSocksPort: "80"}, "below 1024"}, + {"bind outside loopback", map[string]string{annReverseSocksPort: reverseTestSocks, annReverseBind: "10.0.0.1"}, "127.x.y.z"}, + {"bind on the shared loopback address", map[string]string{annReverseSocksPort: reverseTestSocks, annReverseBind: "127.0.0.1"}, "127.x.y.z"}, + } + for _, tt := range rejected { + t.Run("rejects "+tt.name, func(t *testing.T) { + _, err := parseReverseTunnel(reversePod(tt.annotations), cfg) + assert.Error(t, err) + assert.Contains(t, err.Error(), tt.wantErr) + }) + } +} + +func TestReverseBindAddress(t *testing.T) { + a := reverseBindAddress(reverseTestUID) + b := reverseBindAddress("6c0f1e3a-1b2c-4d5e-8f90-a1b2c3d4e5f6") + assert.Equal(t, a, reverseBindAddress(reverseTestUID), "the address must not change across restarts") + assert.NotEqual(t, a, b, "two pods on one node need distinct addresses") + for _, addr := range []string{a, b} { + assert.Regexp(t, `^127\.([1-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-4])\.[0-9]{1,3}\.([1-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-4])$`, addr, + "stay out of 127.0.0.0/24 and off .0 and .255") + } +} + +// assertBashParses runs script through `bash -n`, the shell the batch script uses. +func assertBashParses(t *testing.T, script string) { + t.Helper() + path, err := exec.LookPath("bash") + if err != nil { + t.Log("bash not found, not checking the pre-exec") + return + } + cmd := exec.Command(path, "-n") + cmd.Stdin = strings.NewReader(script) + var stderr bytes.Buffer + cmd.Stderr = &stderr + assert.NoError(t, cmd.Run(), "bash -n on the pre-exec: %s", stderr.String()) +} + +func TestReversePreExec(t *testing.T) { + rt := &ReverseTunnel{ + Forwards: []ReverseForward{ + {ListenPort: 9000, Host: reverseTestService, Port: 9000}, + {ListenPort: 6379, Host: "redis.other.svc.cluster.local", Port: 6379}, + }, + Bind: reverseTestBind, + PythonPath: DefaultSSHReversePythonPath, + } + script := reversePreExec(rt, reverseTestNamespace) + + assert.Contains(t, script, "echo '127.77.1.2 minio.user.svc.cluster.local minio.user.svc minio.user minio redis.other.svc.cluster.local redis.other.svc redis.other'", + "the bare Service name resolves only in the pod's own namespace") + assert.Contains(t, script, "SINGULARITY_BIND") + assert.Contains(t, script, "APPTAINER_BIND") + assert.Contains(t, script, `mktemp "${TMPDIR:-/tmp}/interlink-hosts.XXXXXX"`, + "the hosts file is node-local and unpredictable, never in the socket directory") + assert.Contains(t, script, "needs a plugin with SocketDirRoot set") + assert.Contains(t, script, `python3 -I "$il_rev_d/rev-relay.py" 127.77.1.2 9000="$il_rev_d/rev-9000.sock" 6379="$il_rev_d/rev-6379.sock" &`, + "the job owns the relay for its whole life") + assert.Contains(t, script, reverseTestReady) + assert.Contains(t, script, `[ -S "$il_rev_d/rev-9000.sock" ]`) + assert.Contains(t, script, `[ -S "$il_rev_d/rev-6379.sock" ]`) + assert.Contains(t, script, "the reverse relay stopped") + assert.Contains(t, script, "peer_uid", "the relay is embedded whole") + assert.NotContains(t, script, "/dev/tcp", "a probe would also succeed against a squatter") + assert.NotContains(t, script, "trap", "the plugin's EXIT trap owns the directory") + assert.NotContains(t, script, "il_rev_owner", "no SOCKS wait unless asked") + assertBashParses(t, script) + + socksOnly := reversePreExec(&ReverseTunnel{SocksPort: 1080, Bind: reverseTestBind, PythonPath: DefaultSSHReversePythonPath}, reverseTestNamespace) + assert.Contains(t, socksOnly, `until [ "$(il_rev_owner 0438)" = own ]; do`, "wait for the job user's own listener on 1080") + assert.Contains(t, socksOnly, "another user listens on the SOCKS port") + assert.NotContains(t, socksOnly, "/etc/hosts", "nothing to name without forwards") + assert.NotContains(t, socksOnly, "rev-relay.py", "no relay without forwards") + assertBashParses(t, socksOnly) +} + +func TestReverseHostsBind(t *testing.T) { + path, err := exec.LookPath("bash") + if err != nil { + t.Skip("bash not found") + } + binds := func(env ...string) string { + cmd := exec.Command(path, "-c", "il_rev_hosts=/tmp/h\n"+reverseHostsBind+`echo "$SINGULARITY_BIND|$APPTAINER_BIND"`) + cmd.Env = append([]string{}, env...) + out, err := cmd.Output() + require.NoError(t, err) + return strings.TrimSpace(string(out)) + } + + assert.Equal(t, "/tmp/h:/etc/hosts|/tmp/h:/etc/hosts", binds()) + assert.Equal(t, "/data:/data,/tmp/h:/etc/hosts|/data:/data,/tmp/h:/etc/hosts", binds("SINGULARITY_BIND=/data:/data"), + "Apptainer ignores SINGULARITY_BIND once APPTAINER_BIND is set, so the job's binds move over") + assert.Equal(t, "/tmp/h:/etc/hosts|/data:/data,/tmp/h:/etc/hosts", binds("APPTAINER_BIND=/data:/data")) + assert.Equal(t, "/a:/a,/tmp/h:/etc/hosts|/b:/b,/tmp/h:/etc/hosts", binds("SINGULARITY_BIND=/a:/a", "APPTAINER_BIND=/b:/b")) +} + +func TestShouldCreateShadowForReverse(t *testing.T) { + pod := reversePod(map[string]string{annReverseForwards: reverseTestForward}) + + t.Run("a reverse-only pod gets a shadow", func(t *testing.T) { + p := &Provider{config: normalized(t, reverseConfig())} + assert.True(t, p.shouldCreateShadow(pod)) + assert.NoError(t, p.checkReverseRequest(pod)) + }) + + t.Run("not when the site has not enabled it", func(t *testing.T) { + p := &Provider{config: normalized(t, sshConfig())} + assert.False(t, p.shouldCreateShadow(pod)) + err := p.checkReverseRequest(pod) + assert.Error(t, err, "refused at creation rather than run without its tunnel") + assert.Contains(t, err.Error(), "Network.SSH.Reverse.Enabled") + }) + + t.Run("not with the wstunnel shadow", func(t *testing.T) { + config := Config{Network: Network{EnableTunnel: true}} + assert.NoError(t, NormalizeShadowConfig(&config)) + p := &Provider{config: config} + assert.False(t, p.shouldCreateShadow(pod)) + assert.Error(t, p.checkReverseRequest(pod)) + }) + + t.Run("not with interlink.eu/pod-vpn, which disables the shadow", func(t *testing.T) { + p := &Provider{config: normalized(t, reverseConfig())} + vpn := reversePod(map[string]string{annReverseForwards: reverseTestForward, reverseTestPodVPN: "true"}) + assert.False(t, p.shouldCreateShadow(vpn)) + err := p.checkReverseRequest(vpn) + assert.Error(t, err, "refused at creation rather than run without its tunnel") + assert.Contains(t, err.Error(), reverseTestPodVPN) + }) + + t.Run("a UDP-only pod gets a shadow only for its reverse forwards", func(t *testing.T) { + p := &Provider{config: normalized(t, reverseConfig())} + udp := reversePod(nil) + udp.Spec.Containers = []v1.Container{{Name: "dns", Ports: []v1.ContainerPort{{ContainerPort: 5353, Protocol: v1.ProtocolUDP}}}} + assert.False(t, p.shouldCreateShadow(udp), "nothing for ssh to carry") + + udp.Annotations = map[string]string{annReverseForwards: reverseTestForward} + assert.True(t, p.shouldCreateShadow(udp)) + }) + + t.Run("a pod without the annotations is untouched", func(t *testing.T) { + p := &Provider{config: normalized(t, reverseConfig())} + plain := reversePod(nil) + assert.False(t, p.shouldCreateShadow(plain)) + assert.NoError(t, p.checkReverseRequest(plain)) + }) +} + +func TestCheckReverseRequestSocks(t *testing.T) { + pod := reversePod(map[string]string{annReverseSocksPort: reverseTestSocks}) + + p := &Provider{config: normalized(t, reverseConfig())} + err := p.checkReverseRequest(pod) + assert.Error(t, err, "a SOCKS proxy is open to the whole node: off unless the site allows it") + assert.Contains(t, err.Error(), "Network.SSH.Reverse.AllowSocks") + + config := reverseConfig() + config.Network.SSH.Reverse.AllowSocks = true + p = &Provider{config: normalized(t, config)} + assert.NoError(t, p.checkReverseRequest(pod)) +} + +func TestNormalizeShadowConfigReverse(t *testing.T) { + config := normalized(t, reverseConfig()) + r := config.Network.SSH.Reverse + assert.Equal(t, SSHReverseNodeExecSSH, r.NodeExec) + assert.Equal(t, SSHReverseSshdNode, r.Sshd) + assert.Equal(t, DefaultSSHReverseSshdPath, r.SshdPath) + assert.Equal(t, DefaultSSHReversePythonPath, r.PythonPath) + assert.Empty(t, config.Network.SSH.ExecConnectCommand, "nothing runs a relay command in portforward mode with ssh") + + config = reverseConfig() + config.Network.SSH.Reverse.Sshd = " UserSpace " + config.Network.SSH.Reverse.NodeExec = "SRUN" + config = normalized(t, config) + assert.Equal(t, SSHReverseSshdUserspace, config.Network.SSH.Reverse.Sshd) + assert.Equal(t, SSHReverseNodeExecSrun, config.Network.SSH.Reverse.NodeExec) + assert.Equal(t, DefaultSSHExecConnectCommand, config.Network.SSH.ExecConnectCommand, + "srun reaches node:22 with the relay command") + + for field, mutate := range map[string]func(*SSHReverse){ + "Network.SSH.Reverse.NodeExec": func(r *SSHReverse) { r.NodeExec = "telnet" }, + "Network.SSH.Reverse.Sshd": func(r *SSHReverse) { r.Sshd = "dropbear" }, + } { + config = reverseConfig() + mutate(&config.Network.SSH.Reverse) + err := NormalizeShadowConfig(&config) + assert.Error(t, err) + assert.Contains(t, err.Error(), field) + } +} + +func TestUsesNodeHopsWithReverse(t *testing.T) { + tests := []struct { + sshd, nodeExec string + want bool + }{ + {SSHReverseSshdNode, SSHReverseNodeExecSSH, true}, + {SSHReverseSshdNode, SSHReverseNodeExecSrun, true}, + {SSHReverseSshdUserspace, SSHReverseNodeExecSSH, true}, + // srun starts the user-space sshd, whose host key the shadow generated. + {SSHReverseSshdUserspace, SSHReverseNodeExecSrun, false}, + } + for _, tt := range tests { + config := reverseConfig() + config.Network.SSH.Reverse.Sshd = tt.sshd + config.Network.SSH.Reverse.NodeExec = tt.nodeExec + config = normalized(t, config) + assert.Equal(t, tt.want, usesNodeHops(config.Network.SSH), "%s sshd, %s carrier", tt.sshd, tt.nodeExec) + } + + config := normalized(t, sshConfig()) + assert.False(t, usesNodeHops(config.Network.SSH), "the login target without the reverse tunnel stays off the nodes") +} + +// reverseTunnelFor is what parseReverseTunnel yields for a pod with two forwards +// under config. +func reverseTunnelFor(config Config) *ReverseTunnel { + r := config.Network.SSH.Reverse + return &ReverseTunnel{ + Forwards: []ReverseForward{{ListenPort: 9000, Host: reverseTestService, Port: 9000}, {ListenPort: 6443, Host: "kubernetes.default.svc.cluster.local", Port: 443}}, + Bind: reverseTestBind, + NodeExec: r.NodeExec, + Sshd: r.Sshd, + SshdPath: r.SshdPath, + PythonPath: r.PythonPath, + JobName: reverseTestUID, + } +} + +// renderReverseShadow renders the ssh template for a pod with the given reverse +// tunnel and exposed ports, and returns the manifest, the Deployment and whether a +// Service was rendered. Every container script must parse as shell. +func renderReverseShadow(t *testing.T, config Config, rt *ReverseTunnel, ports []PortMapping) (string, *appsv1.Deployment, bool) { + t.Helper() + config = normalized(t, config) + p := &Provider{config: config} + manifest, err := p.executeShadowTemplate(t.Context(), ShadowTemplateData{ + Name: "shadow-trainer-user", + Namespace: reverseTestNamespace, + ExposedPorts: ports, + Reverse: rt, + NodeConfigMap: "shadow-trainer-user-node", + NodeNameKey: shadowNodeNameKey, + SocketDirKey: shadowSocketDirKey, + NeedsSocketDir: rt != nil || hasSocketPorts(ports), + SSH: config.Network.SSH, + SSHNodeWaitSeconds: p.sshNodeWaitSeconds(), + }) + require.NoError(t, err) + + decoder := serializer.NewCodecFactory(scheme.Scheme).UniversalDeserializer() + var deployment *appsv1.Deployment + service := false + for _, doc := range strings.Split(manifest, "---") { + if strings.TrimSpace(doc) == "" { + continue + } + obj, _, err := decoder.Decode([]byte(doc), nil, nil) + require.NoError(t, err, "every rendered document must decode:\n%s", doc) + switch o := obj.(type) { + case *v1.Service: + service = true + assert.NotEmpty(t, o.Spec.Ports, "a Service without ports is rejected by the API server") + case *appsv1.Deployment: + deployment = o + } + } + require.NotNil(t, deployment) + spec := deployment.Spec.Template.Spec + for _, c := range append(spec.InitContainers, spec.Containers...) { + assertShellParses(t, c) + } + return manifest, deployment, service +} + +func containerNames(deployment *appsv1.Deployment) []string { + var names []string + for _, c := range deployment.Spec.Template.Spec.Containers { + names = append(names, c.Name) + } + return names +} + +// reverseScript returns the script of the ssh-reverse container. +// TestReverseHoldEndsWithItsSockets runs build_rflags from the rendered script: a +// requeue on the same node recreates the socket directory under the same path, so +// the session holding the forwards must end once their sockets are gone. +func TestReverseHoldEndsWithItsSockets(t *testing.T) { + sh, err := exec.LookPath("sh") + if err != nil { + t.Skip("sh not found") + } + config := normalized(t, withReverse(sshConfig(), SSHReverseSshdNode, SSHReverseNodeExecSSH)) + _, deployment, _ := renderReverseShadow(t, config, reverseTunnelFor(config), nil) + script := reverseScript(t, deployment) + start := strings.Index(script, "build_rflags() {\n") + require.GreaterOrEqual(t, start, 0) + end := strings.Index(script[start:], "\n}\n") + require.Positive(t, end) + buildRflags := script[start : start+end+3] + + dir, err := os.MkdirTemp("", "rh") + require.NoError(t, err) + defer os.RemoveAll(dir) + for _, port := range []string{"9000", "6379"} { + l, err := net.Listen("unix", filepath.Join(dir, "rev-"+port+".sock")) + require.NoError(t, err) + l.(*net.UnixListener).SetUnlinkOnClose(false) + defer l.Close() + } + held := func() (bool, bool) { + script := buildRflags + `build_rflags +# the loop condition of each hold command +c=${hold#while } && eval "${c%%; do*}" && echo hold +c=${hold_dir#while } && eval "${c%%; do*}" && echo dir +exit 0` + cmd := exec.Command(sh, "-c", script) // #nosec G204 -- the script this test rendered + cmd.Env = []string{"REV_FORWARDS=9000:minio.user.svc:9000 6379:redis.user.svc:6379", "sd=" + dir} + out, err := cmd.CombinedOutput() + require.NoError(t, err, "%s", out) + return strings.Contains(string(out), "hold"), strings.Contains(string(out), "dir") + } + + forwards, socks := held() + assert.True(t, forwards) + assert.True(t, socks) + + require.NoError(t, os.Remove(filepath.Join(dir, "rev-6379.sock"))) + forwards, socks = held() + assert.False(t, forwards, "a forward whose socket is gone must end its session") + assert.True(t, socks, "the SOCKS session holds no socket in the directory") +} + +func reverseScript(t *testing.T, deployment *appsv1.Deployment) string { + t.Helper() + for _, c := range deployment.Spec.Template.Spec.Containers { + if c.Name == reverseTestContainer && len(c.Args) > 0 { + return c.Args[0] + } + } + t.Fatal("no ssh-reverse container") + return "" +} + +func withReverse(config Config, sshd, nodeExec string) Config { + config.Network.SSH.Reverse.Enabled = true + config.Network.SSH.Reverse.Sshd = sshd + config.Network.SSH.Reverse.NodeExec = nodeExec + return config +} + +func TestSSHShadowTemplateReverse(t *testing.T) { + config := normalized(t, reverseConfig()) + rt := reverseTunnelFor(config) + tcp := []PortMapping{{Port: 8888, Name: "jupyter", Protocol: DefaultProtocol}} + + t.Run("forwards and reverse side by side", func(t *testing.T) { + manifest, deployment, service := renderReverseShadow(t, config, rt, tcp) + assert.Contains(t, containerNames(deployment), "ssh-forward") + assert.Contains(t, containerNames(deployment), reverseTestContainer) + assert.True(t, service) + assert.Contains(t, manifest, `value: "9000:minio.user.svc.cluster.local:9000 6443:kubernetes.default.svc.cluster.local:443"`) + assert.Contains(t, manifest, `-L "0.0.0.0:8888:$node:8888"`) + assert.Contains(t, manifest, `name: REV_SOCKS_PORT + value: ""`, "no SOCKS listener unless the pod asked") + }) + + t.Run("a reverse-only pod renders no forward and no Service, and waits for the socket directory", func(t *testing.T) { + _, deployment, service := renderReverseShadow(t, config, rt, nil) + assert.NotContains(t, containerNames(deployment), "ssh-forward") + assert.Contains(t, containerNames(deployment), reverseTestContainer) + assert.False(t, service) + assert.Contains(t, initScript(t, deployment), "/interlink/node/"+shadowSocketDirKey) + }) + + t.Run("a forward-only pod is unchanged", func(t *testing.T) { + manifest, deployment, service := renderReverseShadow(t, config, nil, tcp) + assert.Contains(t, containerNames(deployment), "ssh-forward") + assert.NotContains(t, containerNames(deployment), reverseTestContainer) + assert.True(t, service) + assert.NotContains(t, manifest, reverseTestUserspace) + assert.NotContains(t, manifest, "job_id") + }) + + t.Run("node sshd forwards to the job's sockets and holds while the job runs", func(t *testing.T) { + manifest, deployment, _ := renderReverseShadow(t, config, rt, nil) + script := reverseScript(t, deployment) + + for _, gone := range []string{reverseTestUserspace, "ssh-keygen -q", "ForceCommand", "EORELAY", "relay.py", "/tmp/il-rev-"} { + assert.NotContains(t, script, gone) + } + assert.NotContains(t, manifest, "REV_OWNER_ONLY") + assert.NotContains(t, manifest, "REV_SSHD", "the node's own sshd needs no path") + assert.Contains(t, script, `"rm -f $sd/rev-*.sock"`, "a stale socket would fail the -R") + assert.Contains(t, script, `rflags="$rflags -R $sd/rev-$lport.sock:$dest"`) + assert.Contains(t, script, `hold="while $held; do sleep 10; echo; done"`, + "the hold loop must write to its session, or it outlives a dropped one") + assert.Contains(t, script, `node_hop "$REV_NODE_EXEC"`) + assert.Contains(t, script, `-o ExitOnForwardFailure=yes $node_opts "$@" "alice@$node" "$h" >/dev/null &`) + assert.Contains(t, script, `session "$hold" $rflags`+"\n") + assert.Contains(t, script, "retrying in 30s") + assert.NotContains(t, script, testRestartHint) + assert.NotContains(t, script, "StrictHostKeyChecking=accept-new $node", "no unchecked hop from the login node") + }) + + t.Run("userspace sshd goes through the checked node hop", func(t *testing.T) { + userspace := normalized(t, withReverse(sshConfig(), SSHReverseSshdUserspace, SSHReverseNodeExecSSH)) + manifest, deployment, _ := renderReverseShadow(t, userspace, reverseTunnelFor(userspace), nil) + script := reverseScript(t, deployment) + + assert.Contains(t, script, reverseTestUserspace) + assert.Contains(t, script, "node_hop ssh || return 1") + assert.Contains(t, script, `run="ssh $node_opts alice@$node"`) + assert.Contains(t, optionSet(t, script, "node_opts"), "ProxyCommand=$w/node-proxy") + assert.Contains(t, script, "cd $sd || exit 1") + assert.Contains(t, script, "StreamLocalBindMask 0177") + assert.NotContains(t, script, "GatewayPorts") + assert.NotContains(t, script, "ForceCommand") + assert.NotContains(t, script, "rm -rf $sd", "the job owns its directory") + assert.Contains(t, script, `"$@" alice@interlink-job "$h" >/dev/null &`) + assert.Contains(t, manifest, `name: REV_SSHD + value: "/usr/sbin/sshd"`) + }) + + t.Run("srun reaches node:22 inside the allocation", func(t *testing.T) { + srun := normalized(t, withReverse(sshConfig(), SSHReverseSshdNode, SSHReverseNodeExecSrun)) + _, deployment, _ := renderReverseShadow(t, srun, reverseTunnelFor(srun), nil) + script := reverseScript(t, deployment) + + assert.Contains(t, script, `exec ssh %s %s "srun --jobid=%s --overlap -u -n1 -w %s --cpus-per-task=1 %s 127.0.0.1 22"`, + "pinned to the batch host, where the relay runs") + assert.Contains(t, script, `squeue -h --me -t R -n $REV_JOB_NAME -o '%i %B'`) + assert.Contains(t, script, "connect_cmd='nc'") + }) + + t.Run("socks warns on every start", func(t *testing.T) { + socks := *rt + socks.SocksPort = 1080 + manifest, deployment, _ := renderReverseShadow(t, config, &socks, nil) + assert.Contains(t, manifest, `name: REV_SOCKS_PORT + value: "1080"`) + script := reverseScript(t, deployment) + assert.Contains(t, script, `session "$hold_dir" -R "127.0.0.1:$REV_SOCKS_PORT"`, + "the SOCKS listener is not in the socket directory") + assert.NotContains(t, script, `rflags="$rflags -R 127.0.0.1`, + "a port another user takes must not fail the session holding the forwards") + assert.Contains(t, script, "keep_socks\n") + assert.Contains(t, script, "WARNING: SOCKS proxy on 127.0.0.1:$REV_SOCKS_PORT") + }) + + t.Run("kerberos delegates on the login hop only for srun", func(t *testing.T) { + for nodeExec, login := range map[string]string{SSHReverseNodeExecSrun: "yes", SSHReverseNodeExecSSH: "no"} { + krb := normalized(t, withReverse(kerberosConfig(), SSHReverseSshdNode, nodeExec)) + _, deployment, _ := renderReverseShadow(t, krb, reverseTunnelFor(krb), nil) + script := reverseScript(t, deployment) + + assert.Contains(t, optionSet(t, script, "ssh_opts"), "-o GSSAPIDelegateCredentials="+login, nodeExec) + assert.Contains(t, optionSet(t, script, "node_opts"), "-o GSSAPIDelegateCredentials=no", nodeExec) + assert.NotContains(t, optionSet(t, script, "node_opts"), "GSSAPIDelegateCredentials=yes", nodeExec) + } + }) +} + +// TestSSHShadowReverseScriptsParse renders the reverse container for every sshd, +// carrier, forward mode and auth method; renderReverseShadow runs each container +// script through the shell parser. +func TestSSHShadowReverseScriptsParse(t *testing.T) { + for _, sshd := range []string{SSHReverseSshdNode, SSHReverseSshdUserspace} { + for _, nodeExec := range []string{SSHReverseNodeExecSSH, SSHReverseNodeExecSrun} { + for _, mode := range []string{SSHForwardModePortForward, SSHForwardModeExec} { + for _, auth := range []string{SSHAuthPublicKey, SSHAuthKerberos} { + t.Run(strings.Join([]string{sshd, nodeExec, mode, auth}, "/"), func(t *testing.T) { + config := sshConfig() + if auth == SSHAuthKerberos { + config = kerberosConfig() + } + config = withForward(config, SSHForwardTargetNode, mode) + config = normalized(t, withReverse(config, sshd, nodeExec)) + rt := reverseTunnelFor(config) + rt.SocksPort = 1080 + renderReverseShadow(t, config, rt, socketAndTCPPorts()) + }) + } + } + } + } +} + +// TestCreateShadowPodRecordsSocksEvent runs createShadowPod against a fake API +// server and checks that a SOCKS proxy, and only a SOCKS proxy, leaves a Warning +// Event on the pod. +func TestCreateShadowPodRecordsSocksEvent(t *testing.T) { + config := reverseConfig() + config.Network.SSH.Reverse.AllowSocks = true + config = normalized(t, config) + + for _, socks := range []bool{true, false} { + annotations := map[string]string{annReverseForwards: reverseTestService + ":9000"} + if socks { + annotations[annReverseSocksPort] = reverseTestSocks + } + pod := reversePod(annotations) + identity, err := computeShadowResourceIdentity(pod) + require.NoError(t, err) + key := &v1.Secret{ObjectMeta: metav1.ObjectMeta{Name: testSSHKeySecret, Namespace: config.Namespace}} + running := &v1.Pod{ObjectMeta: metav1.ObjectMeta{ + Name: identity.Name + "-0", + Namespace: identity.Namespace, + Labels: map[string]string{shadowComponentLabel: identity.Name}, + }} + recorder := record.NewFakeRecorder(10) + p := &Provider{clientSet: fake.NewSimpleClientset(key, running), config: config} + p.SetEventRecorder(recorder) + + _, data, err := p.createShadowPod(t.Context(), pod) + require.NoError(t, err) + assert.True(t, data.NeedsSocketDir, "the reverse tunnel lives in the socket directory") + + select { + case event := <-recorder.Events: + assert.True(t, socks, "unexpected event %q", event) + assert.Contains(t, event, "Warning ReverseSocksProxy SOCKS proxy on 127.0.0.1:1080") + default: + assert.False(t, socks, "no ReverseSocksProxy event") + } + } +} + +// startRelay writes the embedded relay into dir and runs it with python3 -I, +// skipping where python3 or /proc/net/tcp is missing. +func startRelay(t *testing.T, dir string, args ...string) (*exec.Cmd, *bytes.Buffer) { + t.Helper() + python, err := exec.LookPath("python3") + if err != nil { + t.Skip("python3 not found") + } + if _, err := os.Stat("/proc/net/tcp"); err != nil { + t.Skip("no /proc/net/tcp") + } + script := filepath.Join(dir, "rev-relay.py") + require.NoError(t, os.WriteFile(script, []byte(reverseRelay), 0o600)) + cmd := exec.Command(python, append([]string{"-I", script}, args...)...) + var stderr bytes.Buffer + cmd.Stderr = &stderr + require.NoError(t, cmd.Start()) + t.Cleanup(func() { + if err := cmd.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) { + t.Logf("stopping the relay: %v", err) + } + }) + return cmd, &stderr +} + +// waitExit waits up to timeout for cmd and returns its exit code, or -1. +func waitExit(cmd *exec.Cmd, timeout time.Duration) int { + done := make(chan error, 1) + go func() { done <- cmd.Wait() }() + select { + case err := <-done: + var exit *exec.ExitError + if errors.As(err, &exit) { + return exit.ExitCode() + } + if err == nil { + return 0 + } + return -1 + case <-time.After(timeout): + return -1 + } +} + +func freeLoopbackPort(t *testing.T) string { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + port := strconv.Itoa(l.Addr().(*net.TCPAddr).Port) + require.NoError(t, l.Close()) + return port +} + +func TestReverseRelay(t *testing.T) { + t.Run("refuses to start when its address is taken", func(t *testing.T) { + squatter, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer squatter.Close() + port := strconv.Itoa(squatter.Addr().(*net.TCPAddr).Port) + dir := t.TempDir() + + cmd, stderr := startRelay(t, dir, "127.0.0.1", port+"="+filepath.Join(dir, "rev-"+port+".sock")) + + assert.Equal(t, 1, waitExit(cmd, 15*time.Second), "a squatter must fail the job, not share the port") + assert.Contains(t, stderr.String(), "relay: cannot listen on 127.0.0.1:"+port) + assert.NoFileExists(t, filepath.Join(dir, reverseTestReady)) + }) + + t.Run("signals ready, relays to the socket and exits with its directory", func(t *testing.T) { + port := freeLoopbackPort(t) + dir, err := os.MkdirTemp("", "rr") + require.NoError(t, err) + defer os.RemoveAll(dir) + sock := filepath.Join(dir, "rev-"+port+".sock") + upstream, err := net.Listen("unix", sock) + if err != nil { + t.Skipf("cannot listen on %s: %v", sock, err) + } + defer upstream.Close() + go func() { + conn, err := upstream.Accept() + if err != nil { + return + } + defer conn.Close() + buf := make([]byte, 4) + if _, err := io.ReadFull(conn, buf); err != nil { + return + } + if _, err := conn.Write(bytes.ToUpper(buf)); err != nil { + return + } + }() + + cmd, stderr := startRelay(t, dir, "127.0.0.1", port+"="+sock) + require.Eventually(t, func() bool { + _, err := os.Stat(filepath.Join(dir, reverseTestReady)) + return err == nil + }, 15*time.Second, 100*time.Millisecond, "no ready file: %s", stderr) + + conn, err := net.Dial("tcp", "127.0.0.1:"+port) + require.NoError(t, err) + _, err = conn.Write([]byte("ping")) + require.NoError(t, err) + reply := make([]byte, 4) + require.NoError(t, conn.SetReadDeadline(time.Now().Add(10*time.Second))) + _, err = io.ReadFull(conn, reply) + require.NoError(t, err, "the job user's own connection must pass: %s", stderr) + assert.Equal(t, "PING", string(reply)) + conn.Close() + + require.NoError(t, os.RemoveAll(dir)) + assert.Equal(t, 0, waitExit(cmd, 15*time.Second), "the relay must not outlive the job's directory") + }) + + // A client on an IPv6 socket, as a JVM opens by default, reaches the IPv4 + // listener through a v4-mapped address and is listed in /proc/net/tcp6. + t.Run("accepts the job user's own IPv6 socket on a v4-mapped address", func(t *testing.T) { + if _, err := os.Stat("/proc/net/tcp6"); err != nil { + t.Skip("no /proc/net/tcp6") + } + port := freeLoopbackPort(t) + dir, err := os.MkdirTemp("", "rr") + require.NoError(t, err) + defer os.RemoveAll(dir) + sock := filepath.Join(dir, "rev-"+port+".sock") + upstream, err := net.Listen("unix", sock) + if err != nil { + t.Skipf("cannot listen on %s: %v", sock, err) + } + defer upstream.Close() + go func() { + conn, err := upstream.Accept() + if err != nil { + return + } + defer conn.Close() + if _, err := io.Copy(conn, conn); err != nil { + return + } + }() + + cmd, stderr := startRelay(t, dir, "127.0.0.1", port+"="+sock) + require.Eventually(t, func() bool { + _, err := os.Stat(filepath.Join(dir, reverseTestReady)) + return err == nil + }, 15*time.Second, 100*time.Millisecond, "no ready file: %s", stderr) + + // Go dials a v4-mapped address on an IPv4 socket, so the client is python's + probe := `import socket, sys +s = socket.socket(socket.AF_INET6) +s.settimeout(10) +s.connect(("::ffff:127.0.0.1", int(sys.argv[1]))) +s.sendall(b"ping") +print(s.recv(4).decode())` + client := exec.Command(cmd.Path, "-I", "-c", probe, port) // #nosec G204 -- python3 found by startRelay + out, err := client.CombinedOutput() + require.NoError(t, err, "%s", out) + assert.Equal(t, "ping", strings.TrimSpace(string(out)), "relay log: %s", stderr) + }) + + t.Run("keeps serving while the tunnel is down and logs that once", func(t *testing.T) { + port := freeLoopbackPort(t) + dir, err := os.MkdirTemp("", "rr") + require.NoError(t, err) + defer os.RemoveAll(dir) + sock := filepath.Join(dir, "rev-"+port+".sock") + + cmd, stderr := startRelay(t, dir, "127.0.0.1", port+"="+sock) + require.Eventually(t, func() bool { + _, err := os.Stat(filepath.Join(dir, reverseTestReady)) + return err == nil + }, 15*time.Second, 100*time.Millisecond, "no ready file") + + for range 3 { + conn, err := net.Dial("tcp", "127.0.0.1:"+port) + require.NoError(t, err) + require.NoError(t, conn.SetReadDeadline(time.Now().Add(10*time.Second))) + _, err = conn.Read(make([]byte, 1)) + assert.ErrorIs(t, err, io.EOF, "a connection with no tunnel behind it is closed at once") + conn.Close() + } + + require.NoError(t, os.RemoveAll(dir)) + require.Equal(t, 0, waitExit(cmd, 15*time.Second)) + assert.Equal(t, 1, strings.Count(stderr.String(), sock+":"), + "the log is the job's output: one line per outage, not per connection:\n%s", stderr) + }) +} diff --git a/pkg/virtualkubelet/templates/reverse-relay.py b/pkg/virtualkubelet/templates/reverse-relay.py new file mode 100644 index 00000000..6d41e0c5 --- /dev/null +++ b/pkg/virtualkubelet/templates/reverse-relay.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +"""Owner-only TCP front for the reverse tunnel's unix sockets. + +A TCP listener on a loopback address can be reached by every user of the compute +node. This relay accepts a connection only when the connecting socket belongs to the +user running it, read from /proc/net/tcp or tcp6, and then pipes it to a unix +socket that only that user can open. + +The job starts it before its workload and it holds the addresses for the whole job, +so nobody else can take them while the tunnel is down. It refuses to start when one +is already taken, writes rev-relay.ready next to itself once all are bound, and +exits when the job's shell or its own directory goes away. + + rev-relay.py = [...] +""" +import os, socket, sys, threading, time + +ALLOW = os.getuid() +lock = threading.Lock() +counts = {} + + +def note(key, msg): + """Logs the first of a kind of failure and every 1000th after it: the log is + the job's output, and anyone on the node can connect in a loop.""" + with lock: + n = counts[key] = counts.get(key, 0) + 1 + if n == 1 or n % 1000 == 0: + print(f"relay: {msg} ({n} so far)", file=sys.stderr, flush=True) + + +def hexip(ip): + return "".join("%02X" % b for b in reversed(socket.inet_aton(ip))) + + +def peer_uid(conn): + (pip, pport), (lip, lport) = conn.getpeername(), conn.getsockname() + # A client on an IPv6 socket reaches this IPv4 listener through a v4-mapped + # address and is listed in tcp6, as ::ffff: in the same word order. + for table, prefix in (("/proc/net/tcp", ""), ("/proc/net/tcp6", "0000000000000000FFFF0000")): + want_local = "%s%s:%04X" % (prefix, hexip(pip), pport) + want_remote = "%s%s:%04X" % (prefix, hexip(lip), lport) + try: + f = open(table) + except FileNotFoundError: + continue + with f: + next(f) + for line in f: + c = line.split() + if c[1] == want_local and c[2] == want_remote: + return int(c[7]) + return None + + +def pump(src, dst): + try: + while True: + data = src.recv(1 << 16) + if not data: + break + dst.sendall(data) + except OSError: + pass + finally: + try: + dst.shutdown(socket.SHUT_WR) + except OSError: + pass + + +def handle(conn, path): + try: + uid = peer_uid(conn) + if uid != ALLOW: + note(("uid", uid), f"refused connections from uid {uid}") + return + with socket.socket(socket.AF_UNIX) as up: + try: + up.connect(path) + except OSError as e: + note(path, f"{path}: {e}") + return + with lock: + counts.pop(path, None) + t = threading.Thread(target=pump, args=(up, conn), daemon=True) + t.start() + pump(conn, up) + t.join() + except (OSError, RuntimeError) as e: + # Out of descriptors or threads, or a peer gone before it was looked up. + note("drop", f"dropped a connection: {e}") + finally: + conn.close() + + +def serve(s, path): + # A listener whose thread died would stay bound and leave connections hanging + # in its backlog for the rest of the job, so no error may end this loop. + while True: + try: + conn, _ = s.accept() + except OSError as e: + note("accept", f"accept: {e}") + time.sleep(1) + continue + try: + conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + threading.Thread(target=handle, args=(conn, path), daemon=True).start() + except (OSError, RuntimeError) as e: + note("drop", f"dropped a connection: {e}") + conn.close() + + +here = os.path.dirname(os.path.abspath(__file__)) +parent = os.getppid() +bind = sys.argv[1] +listeners = [] +for spec in sys.argv[2:]: + port, path = spec.split("=", 1) + s = socket.socket() + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + s.bind((bind, int(port))) + s.listen(128) + except OSError as e: + print(f"relay: cannot listen on {bind}:{port}: {e}", file=sys.stderr, flush=True) + sys.exit(1) + listeners.append((s, path)) +for s, path in listeners: + threading.Thread(target=serve, args=(s, path), daemon=True).start() +open(os.path.join(here, "rev-relay.ready"), "w").close() +print(f"relay up on {bind}, allowing uid {ALLOW}", file=sys.stderr, flush=True) +while os.getppid() == parent and os.path.isdir(here): + time.sleep(5) diff --git a/pkg/virtualkubelet/templates/shadow-ssh-template.yaml b/pkg/virtualkubelet/templates/shadow-ssh-template.yaml index 4b3a631f..c9a19902 100644 --- a/pkg/virtualkubelet/templates/shadow-ssh-template.yaml +++ b/pkg/virtualkubelet/templates/shadow-ssh-template.yaml @@ -1,6 +1,7 @@ {{- /* ssh client options for the login hop, shared by every ssh the shadow runs against the login node. Delegation comes after ExtraOptions, so that -GSSAPIDelegateCredentials=yes there turns it back on. */ -}} +GSSAPIDelegateCredentials=yes there turns it back on. It is on only for the +reverse tunnel's srun carrier, because srun on the login node needs the ticket. */ -}} {{- define "sshOptions" -}} ssh_opts="\ -p {{.SSH.Port}} \ @@ -25,7 +26,7 @@ ssh_opts="\ -o {{.}} \ {{- end }} {{- if eq .SSH.Auth "kerberos" }} - -o GSSAPIDelegateCredentials=no \ + -o GSSAPIDelegateCredentials={{if and .Reverse (eq .Reverse.NodeExec "srun")}}yes{{else}}no{{end}} \ {{- end }} " {{- end -}} @@ -55,6 +56,19 @@ node_opts="\ " {{- end -}} {{- define "nodeHop" -}} +{{- $srun := and .Reverse (eq .Reverse.NodeExec "srun") -}} +{{- if $srun -}} +# The plugin names the batch job after the pod UID. Steps are pinned to the + # batch host, where the job's relay and socket directory are. + job_id() { + # shellcheck disable=SC2086 + out=$(ssh $ssh_opts "$target" "squeue -h --me -t R -n $REV_JOB_NAME -o '%i %B'" 2>/dev/null | head -n 1) + jid=${out%% *} bhost=${out#"$jid"} bhost=${bhost# } + echo "$jid" | grep -Eq '^[0-9]+$' || { echo "no running job named $REV_JOB_NAME" >&2; return 1; } + echo "$bhost" | grep -Eq '^[A-Za-z0-9][A-Za-z0-9._-]*$' || { echo "no batch host for job $jid" >&2; return 1; } + } + + {{ end -}} # $w/node-proxy carries a session to the compute node's own sshd through the login # node; $w/node_known_hosts is what that session is checked against. node_hop() { @@ -76,6 +90,15 @@ node_opts="\ {{- else }} echo "WARNING: the host key of $node is not verified (NodeHostKeys: accept-new)" >&2 {{- end }} + {{- if $srun }} + if [ "${1:-ssh}" = srun ]; then + job_id || return 1 + printf '#!/bin/sh\nexec ssh %s %s "srun --jobid=%s --overlap -u -n1 -w %s --cpus-per-task=1 %s 127.0.0.1 22"\n' \ + "$ssh_opts" "$target" "$jid" "$bhost" "$connect_cmd" > "$w/node-proxy" + chmod +x "$w/node-proxy" + return + fi + {{- end }} {{- if eq .SSH.ForwardMode "exec" }} printf '#!/bin/sh\nexec ssh %s %s "%s %s 22"\n' "$ssh_opts" "$target" "$connect_cmd" "'$node'" > "$w/node-proxy" {{- else }} @@ -86,6 +109,9 @@ node_opts="\ chmod +x "$w/node-proxy" } {{- end -}} +{{- /* nothing to forward without a TCP port: no forward container, no Service */ -}} +{{- $hasTCP := false }} +{{- range .ExposedPorts }}{{ if ne .Protocol "UDP" }}{{ $hasTCP = true }}{{ end }}{{ end }} apiVersion: apps/v1 kind: Deployment metadata: @@ -213,6 +239,7 @@ spec: capabilities: drop: ["ALL"] {{- end }} + {{- if $hasTCP }} - name: ssh-forward image: {{.SSH.Image}} command: ["/bin/sh", "-c"] @@ -473,6 +500,275 @@ spec: allowPrivilegeEscalation: false capabilities: drop: ["ALL"] + {{- end }} + {{- if .Reverse }} + # Connections opened BY the offloaded pod towards cluster endpoints, with no + # WireGuard and no network namespace (interlink-hq/interLink#548). + # + # The job runs a relay on the pod's loopback address that accepts only the + # job's user and, for each connection, dials a unix socket in the pod's socket + # directory. This container binds those sockets with `ssh -R` on an sshd of + # the compute node, so every connection made there is dialled from here, + # inside the cluster. That sshd is the node's own (Reverse.Sshd "node"), or a + # `sshd -i` started in the job as the job's user, over the stdio of a command + # run through the login node ("userspace"). + - name: ssh-reverse + image: {{.SSH.Image}} + command: ["/bin/sh", "-c"] + args: + - | + set -u + read_node() { tr -d '[:space:]' < /interlink/node/{{.NodeNameKey}} 2>/dev/null || true; } + read_sockdir() { tr -d '[:space:]' < /interlink/node/{{.SocketDirKey}} 2>/dev/null || true; } + + target={{.SSH.User}}@{{.SSH.LoginHost}} + {{- if or (eq .SSH.ForwardMode "exec") (eq .Reverse.NodeExec "srun") }} + connect_cmd='{{.SSH.ExecConnectCommand}}' + {{- end }} + w=/tmp/interlink-reverse + {{ template "sshOptions" . }} + ssh_opts="$ssh_opts -o LogLevel=ERROR" + + {{ template "nodeHop" . }} + + pid= + socks_pid= + socks_t=0 + + # One -R per forward, to the socket the job's relay dials. + build_rflags() { + rflags= + held="[ -d $sd ]" + for f in $REV_FORWARDS; do + lport=${f%%:*} + dest=${f#*:} + rflags="$rflags -R $sd/rev-$lport.sock:$dest" + held="$held && [ -S $sd/rev-$lport.sock ]" + done + # A session holds its forwards until the job removes its directory; the + # one holding the sockets also ends when one is gone, as after a requeue + # on the same node recreated the directory, so that they are bound again. + # The echo ends it with SIGPIPE once its sshd is gone, so a dropped + # session leaves nothing behind on the node. + hold="while $held; do sleep 10; echo; done" + hold_dir="while [ -d $sd ]; do sleep 10; echo; done" + } + {{- if eq .Reverse.Sshd "node" }} + + # Starts a session running the hold command $1 with the -R flags that + # follow, in the background. + session() { + h=$1 + shift + # shellcheck disable=SC2086 + ssh -n -T -o ExitOnForwardFailure=yes $node_opts "$@" "{{.SSH.User}}@$node" "$h" >/dev/null & + } + + start_reverse() { + rm -rf "$w" + mkdir -p "$w" + {{ template "nodeOptions" . }} + node_hop "$REV_NODE_EXEC" || return 1 + build_rflags + + # The node's sshd keeps a stale socket (StreamLocalBindUnlink no) and -R is + # requested before any command runs: clear them in a session of their own. + # shellcheck disable=SC2086 + ssh -n $node_opts "{{.SSH.User}}@$node" "rm -f $sd/rev-*.sock" || { echo "could not clear $sd on $node" >&2; return 1; } + # shellcheck disable=SC2086 + session "$hold" $rflags + pid=$! + } + {{- else }} + + # Starts a session running the hold command $1 with the -R flags that + # follow, in the background, on an sshd -i of its own. + session() { + h=$1 + shift + ssh -n -T \ + -o ProxyCommand="$w/proxy.sh" \ + -o ExitOnForwardFailure=yes \ + -o BatchMode=yes \ + -o ServerAliveInterval=30 -o ServerAliveCountMax=3 \ + -o UserKnownHostsFile="$w/known_hosts" -o StrictHostKeyChecking=yes -o HostKeyAlias=interlink-job \ + -o GSSAPIAuthentication=no -o PreferredAuthentications=publickey \ + -i "$w/client" -o IdentitiesOnly=yes \ + "$@" {{.SSH.User}}@interlink-job "$h" >/dev/null & + } + + start_reverse() { + rm -rf "$w" + mkdir -p "$w" + {{ template "nodeOptions" . }} + + # A fresh key pair for each side on every start: nothing here outlives + # the session, and nothing is taken from the site credential. + ssh-keygen -q -t ed25519 -N '' -C interlink-reverse -f "$w/client" + ssh-keygen -q -t ed25519 -N '' -C interlink-reverse -f "$w/host" + echo "interlink-job $(cat "$w/host.pub")" > "$w/known_hosts" + + { + echo "umask 077; cd $sd || exit 1; rm -f rev-*.sock rev-host rev-authorized_keys rev-sshd_config" + echo "cat > rev-host <<'EOK'"; cat "$w/host"; echo EOK + echo "cat > rev-authorized_keys <<'EOK'"; cat "$w/client.pub"; echo EOK + echo "cat > rev-sshd_config <<'EOK'" + echo "HostKey $sd/rev-host" + echo "AuthorizedKeysFile $sd/rev-authorized_keys" + echo "PidFile none" + echo "UsePAM no" + echo "StrictModes no" + echo "PasswordAuthentication no" + echo "KbdInteractiveAuthentication no" + echo "PubkeyAuthentication yes" + echo "AllowTcpForwarding remote" + echo "AllowStreamLocalForwarding remote" + echo "StreamLocalBindUnlink yes" + echo "StreamLocalBindMask 0177" + echo "AllowAgentForwarding no" + echo "PermitTTY no" + echo "X11Forwarding no" + echo EOK + } > "$w/bootstrap.sh" + + # srun goes through Slurm; ssh goes through the same checked hop as the + # node's own sshd would. + if [ "$REV_NODE_EXEC" = srun ]; then + job_id || return 1 + run="ssh $ssh_opts $target srun --jobid=$jid --overlap -u -n1 -w $bhost --cpus-per-task=1" + else + node_hop ssh || return 1 + run="ssh $node_opts {{.SSH.User}}@$node" + fi + # shellcheck disable=SC2086 + $run "sh -s" < "$w/bootstrap.sh" || { echo "could not prepare $sd on $node" >&2; return 1; } + printf '#!/bin/sh\nexec %s "%s -i -e -f %s/rev-sshd_config"\n' "$run" "$REV_SSHD" "$sd" > "$w/proxy.sh" + chmod +x "$w/proxy.sh" + build_rflags + + # shellcheck disable=SC2086 + session "$hold" $rflags + pid=$! + } + {{- end }} + + # SOCKS has no unix-socket form, so it listens on the node's 127.0.0.1, + # where another user can take the port while the tunnel is down. It runs in + # a session of its own, retried every 30s, so that a failed SOCKS forward + # never takes the listed forwards down with it. + keep_socks() { + [ -n "$REV_SOCKS_PORT" ] || return 0 + if [ -n "$socks_pid" ] && kill -0 "$socks_pid" 2>/dev/null; then + return 0 + fi + now=$(date +%s) + [ $((now - socks_t)) -ge 30 ] || return 0 + if [ -n "$socks_pid" ]; then + echo "the SOCKS forward on $node is down (is 127.0.0.1:$REV_SOCKS_PORT taken?); retrying" >&2 + fi + socks_t=$now + session "$hold_dir" -R "127.0.0.1:$REV_SOCKS_PORT" + socks_pid=$! + } + + stop_reverse() { + for p in $pid $socks_pid; do + kill "$p" 2>/dev/null || true + done + wait 2>/dev/null || true + pid= + socks_pid= + socks_t=0 + } + + # A requeued job comes back on a different host, and the virtual kubelet + # republishes the node and the socket directory into the mounted ConfigMap. + moved() { + cur=$(read_node) + cur_sd=$(read_sockdir) + [ -n "$cur" ] || return 1 + [ "$cur" != "$node" ] || { [ -n "$cur_sd" ] && [ "$cur_sd" != "$sd" ]; } + } + + node=$(read_node) + sd=$(read_sockdir) + [ -n "$node" ] || { echo "no compute node recorded" >&2; exit 1; } + [ -n "$sd" ] || { echo "no socket directory reported: set SocketDirRoot in the plugin" >&2; exit 1; } + + while :; do + echo "reverse tunnel on $node through {{.SSH.LoginHost}} ({{.Reverse.Sshd}} sshd, $REV_NODE_EXEC carrier): $REV_FORWARDS" + if [ -n "$REV_SOCKS_PORT" ]; then + echo "WARNING: SOCKS proxy on 127.0.0.1:$REV_SOCKS_PORT on $node, open to every user of that node" >&2 + fi + if start_reverse; then + while kill -0 "$pid" 2>/dev/null && ! moved; do + keep_socks + sleep 10 + done + fi + stop_reverse + + if moved; then + echo "compute node or socket directory changed, rebuilding the reverse tunnel" + node=$cur + [ -n "$cur_sd" ] && sd=$cur_sd + continue + fi + # The session ends with the job, and a requeued job does not run on $node + # yet: retry here rather than crash-loop into minutes of back-off. + echo "the reverse tunnel to $node is down; retrying in 30s" >&2 + sleep 30 + done + env: + - name: REV_FORWARDS + value: "{{range $i, $f := .Reverse.Forwards}}{{if $i}} {{end}}{{$f.ListenPort}}:{{$f.Host}}:{{$f.Port}}{{end}}" + - name: REV_SOCKS_PORT + value: "{{if .Reverse.SocksPort}}{{.Reverse.SocksPort}}{{end}}" + - name: REV_NODE_EXEC + value: "{{.Reverse.NodeExec}}" + - name: REV_JOB_NAME + value: "{{.Reverse.JobName}}" + {{- if eq .Reverse.Sshd "userspace" }} + - name: REV_SSHD + value: "{{.Reverse.SshdPath}}" + {{- end }} + {{- if eq .SSH.Auth "kerberos" }} + - name: KRB5CCNAME + value: FILE:/interlink/krb5cc/ccache + {{- 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"] + {{- end }} volumes: # Mounted as a directory, not subPath: only directory mounts are refreshed in # place when the VK writes the compute node into this ConfigMap. @@ -503,6 +799,7 @@ spec: configMap: name: {{.SSH.KnownHostsConfigMap}} {{- end }} +{{- if $hasTCP }} --- apiVersion: v1 kind: Service @@ -522,3 +819,4 @@ spec: protocol: {{.Protocol}} {{- end }} {{- end }} +{{- end }} diff --git a/pkg/virtualkubelet/virtualkubelet.go b/pkg/virtualkubelet/virtualkubelet.go index d0327c47..76357c03 100644 --- a/pkg/virtualkubelet/virtualkubelet.go +++ b/pkg/virtualkubelet/virtualkubelet.go @@ -40,6 +40,7 @@ import ( "k8s.io/client-go/kubernetes" "k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/tools/clientcmd" + "k8s.io/client-go/tools/record" "k8s.io/client-go/util/retry" stats "k8s.io/kubelet/pkg/apis/stats/v1alpha1" @@ -148,7 +149,8 @@ type ShadowTemplateData struct { // job created on the compute node. SocketDirKey string // NeedsSocketDir is true when the shadow cannot start without a socket - // directory, because at least one exposed port is a unix socket. + // directory, because at least one exposed port is a unix socket or the pod + // has a reverse tunnel. NeedsSocketDir bool // SSH carries the SSH port-forward settings, used by the ssh shadow template. SSH SSHTunnel @@ -158,6 +160,8 @@ type ShadowTemplateData struct { // PodUID is the UID of the pod the shadow stands for, put on the shadow's pods as // shadowOfLabel. PodUID string + // Reverse is the pod's tunnel back into the cluster, nil unless it asked for one. + Reverse *ReverseTunnel } type PortMapping struct { @@ -219,6 +223,9 @@ type Provider struct { pendingDeletes map[string]*pendingDelete pendingDeletesMu sync.Mutex deleteLoopOnce sync.Once + // eventRecorder records Events on the pods this node runs; nil until + // SetEventRecorder is called. + eventRecorder record.EventRecorder } // pendingDelete tracks a pod whose remote deletion has not been confirmed by the @@ -1097,6 +1104,17 @@ func (p *Provider) createShadowPod(ctx context.Context, originalPod *v1.Pod) (*v if p.isSSHShadow() { warnOnUDPPorts(ctx, templateData.ExposedPorts) + if p.reverseRequested(originalPod) { + reverse, err := parseReverseTunnel(originalPod, p.config.Network.SSH.Reverse) + if err != nil { + return nil, nil, err + } + templateData.Reverse = reverse + templateData.NeedsSocketDir = true + if reverse.SocksPort != 0 { + p.warnSocksProxy(ctx, originalPod, reverse.SocksPort) + } + } } log.G(ctx).Debugf("LocalInitContainers count: %d", len(templateData.LocalInitContainers)) @@ -1945,8 +1963,20 @@ func hasSocketPorts(ports []PortMapping) 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"] == "" + if !p.config.Network.EnableTunnel || pod.Annotations["interlink.eu/pod-vpn"] != "" { + return false + } + if p.reverseRequested(pod) { + return true + } + if !hasExposedPorts(pod) && !hasExtraPortsAnnotation(pod) { + return false + } + // ssh carries TCP only: a pod exposing nothing else would get a shadow with no + // container to run (see warnOnUDPPorts) + return !p.isSSHShadow() || slices.ContainsFunc(extractPortMappings(pod), func(m PortMapping) bool { + return !strings.EqualFold(m.Protocol, "UDP") + }) } // isMeshNetworkingDisabled returns true when the pod has opted out of mesh networking @@ -2000,6 +2030,12 @@ func (p *Provider) handleShadowCreation(ctx context.Context, pod *v1.Pod) (strin // Note: We don't clean up here since the shadow infrastructure is working, // just the annotation failed (non-critical) } + } else if templateData.Reverse != nil { + // The reverse tunnel is the one ssh case where the workload has something to + // do: name the forwarded endpoints and wait for the tunnel before starting. + if err := p.addReversePreExec(ctx, pod, templateData.Reverse); err != nil { + log.G(ctx).Warningf("Failed to record the reverse tunnel pre-exec on pod %s/%s: %v", pod.Namespace, pod.Name, err) + } } return podIP, nil @@ -2165,6 +2201,9 @@ func (p *Provider) CreatePod(ctx context.Context, pod *v1.Pod) error { if err := p.checkSocketPortsRequest(pod); err != nil { return err } + if err := p.checkReverseRequest(pod); err != nil { + return err + } var state v1.ContainerState @@ -2586,6 +2625,12 @@ func (p *Provider) NotifyPods(_ context.Context, f func(*v1.Pod)) { // statusLoopInterval is how often statusLoop polls the plugin for the pods' status. var statusLoopInterval = 5 * time.Second +// SetEventRecorder sets where the provider records Events on pods, normally the +// recorder the pod controller uses. +func (p *Provider) SetEventRecorder(r record.EventRecorder) { + p.eventRecorder = r +} + // statusLoop preiodically monitoring the status of all the pods in p.pods func (p *Provider) statusLoop(ctx context.Context) { t := time.NewTimer(statusLoopInterval)