19166: Set up tunnel for container gateway requests
[arvados.git] / lib / crunchrun / container_gateway.go
index 01457015e16f1870bf4adf4785b8f9c08cec10d5..49eb68c0f50ba263667ba49429dee83a91bd6c5b 100644 (file)
@@ -14,16 +14,22 @@ import (
        "io"
        "net"
        "net/http"
+       "net/url"
        "os"
        "os/exec"
        "sync"
        "syscall"
+       "time"
 
+       "git.arvados.org/arvados.git/lib/controller/rpc"
        "git.arvados.org/arvados.git/lib/selfsigned"
+       "git.arvados.org/arvados.git/sdk/go/arvados"
+       "git.arvados.org/arvados.git/sdk/go/auth"
        "git.arvados.org/arvados.git/sdk/go/ctxlog"
        "git.arvados.org/arvados.git/sdk/go/httpserver"
        "github.com/creack/pty"
        "github.com/google/shlex"
+       "github.com/hashicorp/yamux"
        "golang.org/x/crypto/ssh"
        "golang.org/x/net/context"
 )
@@ -45,12 +51,27 @@ func (GatewayTargetStub) InjectCommand(ctx context.Context, detachKeys, username
 
 type Gateway struct {
        ContainerUUID string
-       Address       string // listen host:port; if port=0, Start() will change it to the selected port
-       AuthSecret    string
-       Target        GatewayTarget
-       Log           interface {
+       // Caller should set Address to "", or "host:0" or "host:port"
+       // where host is a known external IP address; port is a
+       // desired port number to listen on; and ":0" chooses an
+       // available dynamic port.
+       //
+       // If Address is "", Start() listens only on the loopback
+       // interface (and changes Address to "127.0.0.1:port").
+       // Otherwise it listens on all interfaces.
+       //
+       // If Address is "host:0", Start() updates Address to
+       // "host:port".
+       Address    string
+       AuthSecret string
+       Target     GatewayTarget
+       Log        interface {
                Printf(fmt string, args ...interface{})
        }
+       // If non-nil, set up a ContainerGatewayTunnel, so that the
+       // controller can connect to us even if our external IP
+       // address is unknown or not routable from controller.
+       ArvadosClient *arvados.Client
 
        sshConfig   ssh.ServerConfig
        requestAuth string
@@ -99,7 +120,22 @@ func (gw *Gateway) Start() error {
        // from arvados-controller, and PORT is either the desired
        // port where we should run our gateway server, or "0" if we
        // should choose an available port.
-       host, port, err := net.SplitHostPort(gw.Address)
+       extAddr := gw.Address
+       // Generally we can't know which local interface corresponds
+       // to an externally reachable IP address, so if we expect to
+       // be reachable by external hosts, we listen on all
+       // interfaces.
+       listenHost := ""
+       if extAddr == "" {
+               // If the dispatcher doesn't tell us our external IP
+               // address, controller will only be able to connect
+               // through the tunnel (see runTunnel), so our gateway
+               // server only needs to listen on the loopback
+               // interface.
+               extAddr = "127.0.0.1:0"
+               listenHost = "127.0.0.1"
+       }
+       extHost, extPort, err := net.SplitHostPort(extAddr)
        if err != nil {
                return err
        }
@@ -121,26 +157,87 @@ func (gw *Gateway) Start() error {
                                Certificates: []tls.Certificate{cert},
                        },
                },
-               Addr: ":" + port,
+               Addr: net.JoinHostPort(listenHost, extPort),
        }
        err = srv.Start()
        if err != nil {
                return err
        }
-       // Get the port number we are listening on (the port might be
+       // Get the port number we are listening on (extPort might be
        // "0" or a port name, in which case this will be different).
-       _, port, err = net.SplitHostPort(srv.Addr)
+       _, listenPort, err := net.SplitHostPort(srv.Addr)
        if err != nil {
                return err
        }
-       // When changing state to Running, we will set
-       // gateway_address to "HOST:PORT" where HOST is our
-       // external hostname/IP as provided by arvados-dispatch-cloud,
-       // and PORT is the port number we ended up listening on.
-       gw.Address = net.JoinHostPort(host, port)
+       // When changing state to Running, the caller will want to set
+       // gateway_address to a "HOST:PORT" that, if controller
+       // connects to it, will reach this gateway server.
+       //
+       // The most likely thing to work is: HOST is our external
+       // hostname/IP as provided by the caller
+       // (arvados-dispatch-cloud) or 127.0.0.1 to indicate
+       // non-tunnel connections aren't available; and PORT is the
+       // port number we are listening on.
+       gw.Address = net.JoinHostPort(extHost, listenPort)
+       if gw.ArvadosClient != nil {
+               go gw.maintainTunnel(gw.Address)
+       }
        return nil
 }
 
+func (gw *Gateway) maintainTunnel(addr string) {
+       for ; ; time.Sleep(5 * time.Second) {
+               err := gw.runTunnel(addr)
+               gw.Log.Printf("runTunnel: %s", err)
+       }
+}
+
+// runTunnel connects to controller and sets up a tunnel through
+// which controller can connect to the gateway server at the given
+// addr.
+func (gw *Gateway) runTunnel(addr string) error {
+       ctx := auth.NewContext(context.Background(), auth.NewCredentials(gw.ArvadosClient.AuthToken))
+       arpc := rpc.NewConn("", &url.URL{Scheme: "https", Host: gw.ArvadosClient.APIHost}, gw.ArvadosClient.Insecure, rpc.PassthroughTokenProvider)
+       tun, err := arpc.ContainerGatewayTunnel(ctx, arvados.ContainerGatewayTunnelOptions{
+               UUID:       gw.ContainerUUID,
+               AuthSecret: gw.AuthSecret,
+       })
+       if err != nil {
+               return fmt.Errorf("error creating gateway tunnel: %s", err)
+       }
+       mux, err := yamux.Client(tun.Conn, nil)
+       if err != nil {
+               return fmt.Errorf("error setting up mux client end: %s", err)
+       }
+       for {
+               muxconn, err := mux.Accept()
+               if err != nil {
+                       return err
+               }
+               gw.Log.Printf("receiving connection from tunnel, remoteAddr %s", muxconn.RemoteAddr().String())
+               go func() {
+                       defer muxconn.Close()
+                       gwconn, err := net.Dial("tcp", addr)
+                       if err != nil {
+                               gw.Log.Printf("error connecting to %s on behalf of tunnel connection: %s", addr, err)
+                               return
+                       }
+                       defer gwconn.Close()
+                       var wg sync.WaitGroup
+                       wg.Add(2)
+                       go func() {
+                               defer wg.Done()
+                               io.Copy(gwconn, muxconn)
+                       }()
+                       go func() {
+                               defer wg.Done()
+                               io.Copy(muxconn, gwconn)
+                       }()
+                       wg.Wait()
+               }()
+       }
+}
+
 // handleSSH connects to an SSH server that allows the caller to run
 // interactive commands as root (or any other desired user) inside the
 // container. The tunnel itself can only be created by an
@@ -166,7 +263,7 @@ func (gw *Gateway) handleSSH(w http.ResponseWriter, req *http.Request) {
        // In future we'll handle browser traffic too, but for now the
        // only traffic we expect is an SSH tunnel from
        // (*lib/controller/localdb.Conn)ContainerSSH()
-       if req.Method != "GET" || req.Header.Get("Upgrade") != "ssh" {
+       if req.Method != "POST" || req.Header.Get("Upgrade") != "ssh" {
                http.Error(w, "path not found", http.StatusNotFound)
                return
        }
@@ -204,7 +301,9 @@ func (gw *Gateway) handleSSH(w http.ResponseWriter, req *http.Request) {
        ctx := req.Context()
 
        conn, newchans, reqs, err := ssh.NewServerConn(netconn, &gw.sshConfig)
-       if err != nil {
+       if err == io.EOF {
+               return
+       } else if err != nil {
                gw.Log.Printf("ssh.NewServerConn: %s", err)
                return
        }