"crypto/hmac"
"crypto/sha256"
"fmt"
+ "io/ioutil"
+ "net"
+ "net/http"
"net/url"
"os"
"os/exec"
+ "strings"
+ "sync"
+ "syscall"
+ "time"
"git.arvados.org/arvados.git/lib/controller/rpc"
"git.arvados.org/arvados.git/lib/crunchrun"
"git.arvados.org/arvados.git/sdk/go/arvados"
"git.arvados.org/arvados.git/sdk/go/arvadostest"
+ "git.arvados.org/arvados.git/sdk/go/httpserver"
check "gopkg.in/check.v1"
)
ContainerUUID: uuid,
Address: "0.0.0.0:0",
AuthSecret: authSecret,
+ // Just forward connections to localhost instead of a
+ // container, so we can test without running a
+ // container.
+ ContainerIPAddress: func() (string, error) { return "0.0.0.0", nil },
}
err := gw.Start()
c.Assert(err, check.IsNil)
c.Check(cmd.Run(), check.NotNil)
c.Log(stderr.String())
c.Check(stderr.String(), check.Matches, `(?ms).*(No such container: theperthcountyconspiracy|exec: \"docker\": executable file not found in \$PATH).*`)
+
+ // Set up an http server, and try using "arvados-client shell"
+ // to forward traffic to it.
+ httpTarget := &httpserver.Server{}
+ httpTarget.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ c.Logf("httpTarget.Handler: incoming request: %s %s", r.Method, r.URL)
+ if r.URL.Path == "/foo" {
+ fmt.Fprintln(w, "bar baz")
+ } else {
+ w.WriteHeader(http.StatusNotFound)
+ }
+ })
+ err = httpTarget.Start()
+ c.Assert(err, check.IsNil)
+
+ ln, err := net.Listen("tcp", ":0")
+ c.Assert(err, check.IsNil)
+ _, forwardedPort, _ := net.SplitHostPort(ln.Addr().String())
+ ln.Close()
+
+ stdout.Reset()
+ stderr.Reset()
+ ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(10*time.Second))
+ defer cancel()
+ cmd = exec.CommandContext(ctx,
+ "go", "run", ".", "shell", uuid,
+ "-L", forwardedPort+":"+httpTarget.Addr,
+ "-o", "controlpath=none",
+ "-o", "userknownhostsfile="+c.MkDir()+"/known_hosts",
+ "-N",
+ )
+ c.Logf("cmd.Args: %s", cmd.Args)
+ cmd.Env = append(cmd.Env, os.Environ()...)
+ cmd.Env = append(cmd.Env, "ARVADOS_API_TOKEN="+arvadostest.ActiveTokenV2)
+ cmd.Stdout = &stdout
+ cmd.Stderr = &stderr
+ cmd.Start()
+
+ forwardedURL := fmt.Sprintf("http://localhost:%s/foo", forwardedPort)
+
+ for range time.NewTicker(time.Second / 20).C {
+ resp, err := http.Get(forwardedURL)
+ if err != nil {
+ if !strings.Contains(err.Error(), "connect") {
+ c.Fatal(err)
+ } else if ctx.Err() != nil {
+ if cmd.Process.Signal(syscall.Signal(0)) != nil {
+ c.Error("OpenSSH exited")
+ } else {
+ c.Errorf("timed out trying to connect: %s", err)
+ }
+ c.Logf("OpenSSH stdout:\n%s", stdout.String())
+ c.Logf("OpenSSH stderr:\n%s", stderr.String())
+ c.FailNow()
+ }
+ // Retry until OpenSSH starts listening
+ continue
+ }
+ c.Check(resp.StatusCode, check.Equals, http.StatusOK)
+ body, err := ioutil.ReadAll(resp.Body)
+ c.Check(err, check.IsNil)
+ c.Check(string(body), check.Equals, "bar baz\n")
+ break
+ }
+
+ var wg sync.WaitGroup
+ for i := 0; i < 10; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ resp, err := http.Get(forwardedURL)
+ if !c.Check(err, check.IsNil) {
+ return
+ }
+ body, err := ioutil.ReadAll(resp.Body)
+ c.Check(err, check.IsNil)
+ c.Check(string(body), check.Equals, "bar baz\n")
+ }()
+ }
+ wg.Wait()
}
"crypto/sha256"
"fmt"
"io"
+ "io/ioutil"
+ "net"
"time"
"git.arvados.org/arvados.git/lib/config"
"git.arvados.org/arvados.git/sdk/go/arvadostest"
"git.arvados.org/arvados.git/sdk/go/auth"
"git.arvados.org/arvados.git/sdk/go/ctxlog"
+ "golang.org/x/crypto/ssh"
check "gopkg.in/check.v1"
)
authKey := fmt.Sprintf("%x", h.Sum(nil))
s.gw = &crunchrun.Gateway{
- DockerContainerID: new(string),
- ContainerUUID: s.ctrUUID,
- AuthSecret: authKey,
- Address: "localhost:0",
- Log: ctxlog.TestLogger(c),
+ DockerContainerID: new(string),
+ ContainerUUID: s.ctrUUID,
+ AuthSecret: authKey,
+ Address: "localhost:0",
+ Log: ctxlog.TestLogger(c),
+ ContainerIPAddress: func() (string, error) { return "localhost", nil },
}
c.Assert(s.gw.Start(), check.IsNil)
rootctx := auth.NewContext(context.Background(), &auth.Credentials{Tokens: []string{s.cluster.SystemRootToken}})
}
}
+func (s *ContainerGatewaySuite) TestDirectTCP(c *check.C) {
+ // Set up servers on a few TCP ports
+ var addrs []string
+ for i := 0; i < 3; i++ {
+ ln, err := net.Listen("tcp", ":0")
+ c.Assert(err, check.IsNil)
+ defer ln.Close()
+ addrs = append(addrs, ln.Addr().String())
+ go func() {
+ for {
+ conn, err := ln.Accept()
+ if err != nil {
+ return
+ }
+ var gotAddr string
+ fmt.Fscanf(conn, "%s\n", &gotAddr)
+ c.Logf("stub server listening at %s received string %q from remote %s", ln.Addr().String(), gotAddr, conn.RemoteAddr())
+ if gotAddr == ln.Addr().String() {
+ fmt.Fprintf(conn, "%s\n", ln.Addr().String())
+ }
+ conn.Close()
+ }
+ }()
+ }
+
+ c.Logf("connecting to %s", s.gw.Address)
+ sshconn, err := s.localdb.ContainerSSH(s.ctx, arvados.ContainerSSHOptions{UUID: s.ctrUUID})
+ c.Assert(err, check.IsNil)
+ c.Assert(sshconn.Conn, check.NotNil)
+ defer sshconn.Conn.Close()
+ conn, chans, reqs, err := ssh.NewClientConn(sshconn.Conn, "zzzz-dz642-abcdeabcdeabcde", &ssh.ClientConfig{
+ HostKeyCallback: func(hostname string, remote net.Addr, key ssh.PublicKey) error { return nil },
+ })
+ c.Assert(err, check.IsNil)
+ client := ssh.NewClient(conn, chans, reqs)
+ for _, expectAddr := range addrs {
+ _, port, err := net.SplitHostPort(expectAddr)
+ c.Assert(err, check.IsNil)
+
+ c.Logf("trying foo:%s", port)
+ {
+ conn, err := client.Dial("tcp", "foo:"+port)
+ c.Assert(err, check.IsNil)
+ conn.SetDeadline(time.Now().Add(time.Second))
+ buf, err := ioutil.ReadAll(conn)
+ c.Check(err, check.IsNil)
+ c.Check(string(buf), check.Equals, "")
+ }
+
+ c.Logf("trying localhost:%s", port)
+ {
+ conn, err := client.Dial("tcp", "localhost:"+port)
+ c.Assert(err, check.IsNil)
+ conn.SetDeadline(time.Now().Add(time.Second))
+ conn.Write([]byte(expectAddr + "\n"))
+ var gotAddr string
+ fmt.Fscanf(conn, "%s\n", &gotAddr)
+ c.Check(gotAddr, check.Equals, expectAddr)
+ }
+ }
+}
+
func (s *ContainerGatewaySuite) TestConnect(c *check.C) {
c.Logf("connecting to %s", s.gw.Address)
sshconn, err := s.localdb.ContainerSSH(s.ctx, arvados.ContainerSSHOptions{UUID: s.ctrUUID})
"os"
"os/exec"
"sync"
+ "sync/atomic"
"syscall"
+ "time"
"git.arvados.org/arvados.git/lib/selfsigned"
+ "git.arvados.org/arvados.git/sdk/go/ctxlog"
"git.arvados.org/arvados.git/sdk/go/httpserver"
"github.com/creack/pty"
+ dockerclient "github.com/docker/docker/client"
"github.com/google/shlex"
"golang.org/x/crypto/ssh"
+ "golang.org/x/net/context"
)
type Gateway struct {
Log interface {
Printf(fmt string, args ...interface{})
}
+ // return local ip address of running container, or "" if not available
+ ContainerIPAddress func() (string, error)
sshConfig ssh.ServerConfig
requestAuth string
defer conn.Close()
go ssh.DiscardRequests(reqs)
for newch := range newchans {
- if newch.ChannelType() != "session" {
- newch.Reject(ssh.UnknownChannelType, fmt.Sprintf("unsupported channel type %q", newch.ChannelType()))
- continue
+ switch newch.ChannelType() {
+ case "direct-tcpip":
+ go gw.handleDirectTCPIP(ctx, newch)
+ case "session":
+ go gw.handleSession(ctx, newch, detachKeys, username)
+ default:
+ go newch.Reject(ssh.UnknownChannelType, fmt.Sprintf("unsupported channel type %q", newch.ChannelType()))
}
- ch, reqs, err := newch.Accept()
+ }
+}
+
+func (gw *Gateway) handleDirectTCPIP(ctx context.Context, newch ssh.NewChannel) {
+ ch, reqs, err := newch.Accept()
+ if err != nil {
+ gw.Log.Printf("accept direct-tcpip channel: %s", err)
+ return
+ }
+ defer ch.Close()
+ go ssh.DiscardRequests(reqs)
+
+ // RFC 4254 7.2 (copy of channelOpenDirectMsg in
+ // golang.org/x/crypto/ssh)
+ var msg struct {
+ Raddr string
+ Rport uint32
+ Laddr string
+ Lport uint32
+ }
+ err = ssh.Unmarshal(newch.ExtraData(), &msg)
+ if err != nil {
+ fmt.Fprintf(ch.Stderr(), "unmarshal direct-tcpip extradata: %s\n", err)
+ return
+ }
+ switch msg.Raddr {
+ case "localhost", "0.0.0.0", "127.0.0.1", "::1", "::":
+ default:
+ fmt.Fprintf(ch.Stderr(), "cannot forward to ports on %q, only localhost\n", msg.Raddr)
+ return
+ }
+
+ var dstaddr string
+ if gw.ContainerIPAddress != nil {
+ dstaddr, err = gw.ContainerIPAddress()
if err != nil {
- gw.Log.Printf("accept channel: %s", err)
+ fmt.Fprintf(ch.Stderr(), "container has no IP address: %s\n", err)
return
}
- var pty0, tty0 *os.File
- go func() {
- // Where to send errors/messages for the
- // client to see
- logw := io.Writer(ch.Stderr())
- // How to end lines when sending
- // errors/messages to the client (changes to
- // \r\n when using a pty)
- eol := "\n"
- // Env vars to add to child process
- termEnv := []string(nil)
- for req := range reqs {
- ok := false
- switch req.Type {
- case "shell", "exec":
- ok = true
- var payload struct {
- Command string
- }
- ssh.Unmarshal(req.Payload, &payload)
- execargs, err := shlex.Split(payload.Command)
- if err != nil {
- fmt.Fprintf(logw, "error parsing supplied command: %s"+eol, err)
- return
- }
- if len(execargs) == 0 {
- execargs = []string{"/bin/bash", "-login"}
- }
- go func() {
- cmd := exec.CommandContext(ctx, "docker", "exec", "-i", "--detach-keys="+detachKeys, "--user="+username)
- cmd.Stdin = ch
- cmd.Stdout = ch
- cmd.Stderr = ch.Stderr()
- if tty0 != nil {
- cmd.Args = append(cmd.Args, "-t")
- cmd.Stdin = tty0
- cmd.Stdout = tty0
- cmd.Stderr = tty0
- var wg sync.WaitGroup
- defer wg.Wait()
- wg.Add(2)
- go func() { io.Copy(ch, pty0); wg.Done() }()
- go func() { io.Copy(pty0, ch); wg.Done() }()
- // Send our own debug messages to tty as well.
- logw = tty0
- }
- cmd.Args = append(cmd.Args, *gw.DockerContainerID)
- cmd.Args = append(cmd.Args, execargs...)
- cmd.SysProcAttr = &syscall.SysProcAttr{
- Setctty: tty0 != nil,
- Setsid: true,
- }
- cmd.Env = append(os.Environ(), termEnv...)
- err := cmd.Run()
- var resp struct {
- Status uint32
- }
- if exiterr, ok := err.(*exec.ExitError); ok {
- if status, ok := exiterr.Sys().(syscall.WaitStatus); ok {
- resp.Status = uint32(status.ExitStatus())
- }
- } else if err != nil {
- // Propagate errors like `exec: "docker": executable file not found in $PATH`
- fmt.Fprintln(ch.Stderr(), err)
- }
- errClose := ch.CloseWrite()
- if resp.Status == 0 && (err != nil || errClose != nil) {
- resp.Status = 1
- }
- ch.SendRequest("exit-status", false, ssh.Marshal(&resp))
- ch.Close()
- }()
- case "pty-req":
- eol = "\r\n"
- p, t, err := pty.Open()
- if err != nil {
- fmt.Fprintf(ch.Stderr(), "pty failed: %s"+eol, err)
- break
- }
- defer p.Close()
- defer t.Close()
- pty0, tty0 = p, t
- ok = true
- var payload struct {
- Term string
- Cols uint32
- Rows uint32
- X uint32
- Y uint32
- }
- ssh.Unmarshal(req.Payload, &payload)
- termEnv = []string{"TERM=" + payload.Term, "USE_TTY=1"}
- err = pty.Setsize(pty0, &pty.Winsize{Rows: uint16(payload.Rows), Cols: uint16(payload.Cols), X: uint16(payload.X), Y: uint16(payload.Y)})
- if err != nil {
- fmt.Fprintf(logw, "pty-req: setsize failed: %s"+eol, err)
- }
- case "window-change":
- var payload struct {
- Cols uint32
- Rows uint32
- X uint32
- Y uint32
- }
- ssh.Unmarshal(req.Payload, &payload)
- err := pty.Setsize(pty0, &pty.Winsize{Rows: uint16(payload.Rows), Cols: uint16(payload.Cols), X: uint16(payload.X), Y: uint16(payload.Y)})
- if err != nil {
- fmt.Fprintf(logw, "window-change: setsize failed: %s"+eol, err)
- break
+ }
+ if dstaddr == "" {
+ fmt.Fprintf(ch.Stderr(), "container has no IP address\n")
+ return
+ }
+
+ dst := net.JoinHostPort(dstaddr, fmt.Sprintf("%d", msg.Rport))
+ tcpconn, err := net.Dial("tcp", dst)
+ if err != nil {
+ fmt.Fprintf(ch.Stderr(), "%s: %s\n", dst, err)
+ return
+ }
+ go func() {
+ n, _ := io.Copy(ch, tcpconn)
+ ctxlog.FromContext(ctx).Debugf("tcpip: sent %d bytes\n", n)
+ ch.CloseWrite()
+ }()
+ n, _ := io.Copy(tcpconn, ch)
+ ctxlog.FromContext(ctx).Debugf("tcpip: received %d bytes\n", n)
+}
+
+func (gw *Gateway) handleSession(ctx context.Context, newch ssh.NewChannel, detachKeys, username string) {
+ ch, reqs, err := newch.Accept()
+ if err != nil {
+ gw.Log.Printf("accept session channel: %s", err)
+ return
+ }
+ var pty0, tty0 *os.File
+ // Where to send errors/messages for the client to see
+ logw := io.Writer(ch.Stderr())
+ // How to end lines when sending errors/messages to the client
+ // (changes to \r\n when using a pty)
+ eol := "\n"
+ // Env vars to add to child process
+ termEnv := []string(nil)
+ for req := range reqs {
+ ok := false
+ switch req.Type {
+ case "shell", "exec":
+ ok = true
+ var payload struct {
+ Command string
+ }
+ ssh.Unmarshal(req.Payload, &payload)
+ execargs, err := shlex.Split(payload.Command)
+ if err != nil {
+ fmt.Fprintf(logw, "error parsing supplied command: %s"+eol, err)
+ return
+ }
+ if len(execargs) == 0 {
+ execargs = []string{"/bin/bash", "-login"}
+ }
+ go func() {
+ cmd := exec.CommandContext(ctx, "docker", "exec", "-i", "--detach-keys="+detachKeys, "--user="+username)
+ cmd.Stdin = ch
+ cmd.Stdout = ch
+ cmd.Stderr = ch.Stderr()
+ if tty0 != nil {
+ cmd.Args = append(cmd.Args, "-t")
+ cmd.Stdin = tty0
+ cmd.Stdout = tty0
+ cmd.Stderr = tty0
+ var wg sync.WaitGroup
+ defer wg.Wait()
+ wg.Add(2)
+ go func() { io.Copy(ch, pty0); wg.Done() }()
+ go func() { io.Copy(pty0, ch); wg.Done() }()
+ // Send our own debug messages to tty as well.
+ logw = tty0
+ }
+ cmd.Args = append(cmd.Args, *gw.DockerContainerID)
+ cmd.Args = append(cmd.Args, execargs...)
+ cmd.SysProcAttr = &syscall.SysProcAttr{
+ Setctty: tty0 != nil,
+ Setsid: true,
+ }
+ cmd.Env = append(os.Environ(), termEnv...)
+ err := cmd.Run()
+ var resp struct {
+ Status uint32
+ }
+ if exiterr, ok := err.(*exec.ExitError); ok {
+ if status, ok := exiterr.Sys().(syscall.WaitStatus); ok {
+ resp.Status = uint32(status.ExitStatus())
}
- ok = true
- case "env":
- // TODO: implement "env"
- // requests by setting env
- // vars in the docker-exec
- // command (not docker-exec's
- // own environment, which
- // would be a gaping security
- // hole).
- default:
- // fmt.Fprintf(logw, "declining %q req"+eol, req.Type)
+ } else if err != nil {
+ // Propagate errors like `exec: "docker": executable file not found in $PATH`
+ fmt.Fprintln(ch.Stderr(), err)
}
- if req.WantReply {
- req.Reply(ok, nil)
+ errClose := ch.CloseWrite()
+ if resp.Status == 0 && (err != nil || errClose != nil) {
+ resp.Status = 1
}
+ ch.SendRequest("exit-status", false, ssh.Marshal(&resp))
+ ch.Close()
+ }()
+ case "pty-req":
+ eol = "\r\n"
+ p, t, err := pty.Open()
+ if err != nil {
+ fmt.Fprintf(ch.Stderr(), "pty failed: %s"+eol, err)
+ break
+ }
+ defer p.Close()
+ defer t.Close()
+ pty0, tty0 = p, t
+ ok = true
+ var payload struct {
+ Term string
+ Cols uint32
+ Rows uint32
+ X uint32
+ Y uint32
+ }
+ ssh.Unmarshal(req.Payload, &payload)
+ termEnv = []string{"TERM=" + payload.Term, "USE_TTY=1"}
+ err = pty.Setsize(pty0, &pty.Winsize{Rows: uint16(payload.Rows), Cols: uint16(payload.Cols), X: uint16(payload.X), Y: uint16(payload.Y)})
+ if err != nil {
+ fmt.Fprintf(logw, "pty-req: setsize failed: %s"+eol, err)
}
- }()
+ case "window-change":
+ var payload struct {
+ Cols uint32
+ Rows uint32
+ X uint32
+ Y uint32
+ }
+ ssh.Unmarshal(req.Payload, &payload)
+ err := pty.Setsize(pty0, &pty.Winsize{Rows: uint16(payload.Rows), Cols: uint16(payload.Cols), X: uint16(payload.X), Y: uint16(payload.Y)})
+ if err != nil {
+ fmt.Fprintf(logw, "window-change: setsize failed: %s"+eol, err)
+ break
+ }
+ ok = true
+ case "env":
+ // TODO: implement "env"
+ // requests by setting env
+ // vars in the docker-exec
+ // command (not docker-exec's
+ // own environment, which
+ // would be a gaping security
+ // hole).
+ default:
+ // fmt.Fprintf(logw, "declining %q req"+eol, req.Type)
+ }
+ if req.WantReply {
+ req.Reply(ok, nil)
+ }
+ }
+}
+
+func dockerContainerIPAddress(containerID *string) func() (string, error) {
+ var saved atomic.Value
+ return func() (string, error) {
+ if ip, ok := saved.Load().(*string); ok {
+ return *ip, nil
+ }
+ docker, err := dockerclient.NewClient(dockerclient.DefaultDockerHost, "1.21", nil, nil)
+ if err != nil {
+ return "", fmt.Errorf("cannot create docker client: %s", err)
+ }
+ ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Minute))
+ defer cancel()
+ ctr, err := docker.ContainerInspect(ctx, *containerID)
+ if err != nil {
+ return "", fmt.Errorf("cannot get docker container info: %s", err)
+ }
+ ip := ctr.NetworkSettings.IPAddress
+ if ip == "" {
+ // TODO: try to enable networking if it wasn't
+ // already enabled when the container was
+ // created.
+ return "", fmt.Errorf("container has no IP address")
+ }
+ saved.Store(&ip)
+ return ip, nil
}
}
projects / arvados.git / commitdiff