Add an API for Attach over HTTP API

The new APIv2 branch provides an HTTP-based remote API to Podman.
The requirements of this are, unfortunately, incompatible with
the existing Attach API. For non-terminal attach, we need append
a header to what was copied from the container, to multiplex
STDOUT and STDERR; to do this with the old API, we'd need to copy
into an intermediate buffer first, to handle the headers.

To avoid this, provide a new API to handle all aspects of
terminal and non-terminal attach, including closing the hijacked
HTTP connection. This might be a bit too specific, but for now,
it seems to be the simplest approach.

At the same time, add a Resize endpoint. This needs to be a
separate endpoint, so our existing channel approach does not work
here.

I wanted to rework the rest of attach at the same time (some
parts of it, particularly how we start the Attach session and how
we do resizing, are (in my opinion) handled much better here.
That may still be on the table, but I wanted to avoid breaking
existing APIs in this already massive change.

Signed-off-by: Matthew Heon <matthew.heon@pm.me>

5 files changed, 393 insertions, 2 deletions

diff --git a/libpod/container_api.go b/libpod/container_api.go
index e36623529..d74a14f15 100644
--- a/libpod/container_api.go
+++ b/libpod/container_api.go
@@ -5,6 +5,7 @@ import (
 	"context"
 	"io"
 	"io/ioutil"
+	"net"
 	"os"
 	"time"
 
@@ -374,7 +375,9 @@ type AttachStreams struct {
 	AttachInput bool
 }
 
-// Attach attaches to a container
+// Attach attaches to a container.
+// This function returns when the attach finishes. It does not hold the lock for
+// the duration of its runtime, only using it at the beginning to verify state.
 func (c *Container) Attach(streams *AttachStreams, keys string, resize <-chan remotecommand.TerminalSize) error {
 	if !c.batched {
 		c.lock.Lock()
@@ -382,6 +385,7 @@ func (c *Container) Attach(streams *AttachStreams, keys string, resize <-chan re
 			c.lock.Unlock()
 			return err
 		}
+		// We are NOT holding the lock for the duration of the function.
 		c.lock.Unlock()
 	}
 
@@ -389,10 +393,71 @@ func (c *Container) Attach(streams *AttachStreams, keys string, resize <-chan re
 		return errors.Wrapf(define.ErrCtrStateInvalid, "can only attach to created or running containers")
 	}
 
-	defer c.newContainerEvent(events.Attach)
+	c.newContainerEvent(events.Attach)
 	return c.attach(streams, keys, resize, false, nil)
 }
 
+// HTTPAttach forwards an attach session over a hijacked HTTP session.
+// HTTPAttach will consume and close the included httpCon, which is expected to
+// be sourced from a hijacked HTTP connection.
+// The cancel channel is optional, and can be used to asyncronously cancel the
+// attach session.
+// The streams variable is only supported if the container was not a terminal,
+// and allows specifying which of the container's standard streams will be
+// forwarded to the client.
+// This function returns when the attach finishes. It does not hold the lock for
+// the duration of its runtime, only using it at the beginning to verify state.
+func (c *Container) HTTPAttach(httpCon net.Conn, httpBuf *bufio.ReadWriter, streams *HTTPAttachStreams, detachKeys *string, cancel <-chan bool) error {
+	if !c.batched {
+		c.lock.Lock()
+		if err := c.syncContainer(); err != nil {
+			c.lock.Unlock()
+
+			// Write any errors to the HTTP buffer before we close.
+			hijackWriteErrorAndClose(err, c.ID(), httpCon, httpBuf)
+
+			return err
+		}
+		// We are NOT holding the lock for the duration of the function.
+		c.lock.Unlock()
+	}
+
+	if !c.ensureState(define.ContainerStateCreated, define.ContainerStateRunning) {
+		toReturn := errors.Wrapf(define.ErrCtrStateInvalid, "can only attach to created or running containers")
+
+		// Write any errors to the HTTP buffer before we close.
+		hijackWriteErrorAndClose(toReturn, c.ID(), httpCon, httpBuf)
+
+		return toReturn
+	}
+
+	logrus.Infof("Performing HTTP Hijack attach to container %s", c.ID())
+
+	c.newContainerEvent(events.Attach)
+	return c.ociRuntime.HTTPAttach(c, httpCon, httpBuf, streams, detachKeys, cancel)
+}
+
+// AttachResize resizes the container's terminal, which is displayed by Attach
+// and HTTPAttach.
+func (c *Container) AttachResize(newSize remotecommand.TerminalSize) error {
+	if !c.batched {
+		c.lock.Lock()
+		defer c.lock.Unlock()
+
+		if err := c.syncContainer(); err != nil {
+			return err
+		}
+	}
+
+	if !c.ensureState(define.ContainerStateCreated, define.ContainerStateRunning) {
+		return errors.Wrapf(define.ErrCtrStateInvalid, "can only resize created or running containers")
+	}
+
+	logrus.Infof("Resizing TTY of container %s", c.ID())
+
+	return c.ociRuntime.AttachResize(c, newSize)
+}
+
 // Mount mounts a container's filesystem on the host
 // The path where the container has been mounted is returned
 func (c *Container) Mount() (string, error) {
diff --git a/libpod/oci.go b/libpod/oci.go
index 05a2f37db..2ea61851f 100644
--- a/libpod/oci.go
+++ b/libpod/oci.go
@@ -1,6 +1,9 @@
 package libpod
 
 import (
+	"bufio"
+	"net"
+
 	"k8s.io/client-go/tools/remotecommand"
 )
 
@@ -47,6 +50,23 @@ type OCIRuntime interface {
 	// UnpauseContainer unpauses the given container.
 	UnpauseContainer(ctr *Container) error
 
+	// HTTPAttach performs an attach intended to be transported over HTTP.
+	// For terminal attach, the container's output will be directly streamed
+	// to output; otherwise, STDOUT and STDERR will be multiplexed, with
+	// a header prepended as follows: 1-byte STREAM (0, 1, 2 for STDIN,
+	// STDOUT, STDERR), 3 null (0x00) bytes, 4-byte big endian length.
+	// If a cancel channel is provided, it can be used to asyncronously
+	// termninate the attach session. Detach keys, if given, will also cause
+	// the attach session to be terminated if provided via the STDIN
+	// channel. If they are not provided, the default detach keys will be
+	// used instead. Detach keys of "" will disable detaching via keyboard.
+	// The streams parameter may be passed for containers that did not
+	// create a terminal and will determine which streams to forward to the
+	// client.
+	HTTPAttach(ctr *Container, httpConn net.Conn, httpBuf *bufio.ReadWriter, streams *HTTPAttachStreams, detachKeys *string, cancel <-chan bool) error
+	// AttachResize resizes the terminal in use by the given container.
+	AttachResize(ctr *Container, newSize remotecommand.TerminalSize) error
+
 	// ExecContainer executes a command in a running container.
 	// Returns an int (exit code), error channel (errors from attach), and
 	// error (errors that occurred attempting to start the exec session).
@@ -130,3 +150,12 @@ type ExecOptions struct {
 	// detach from the container.
 	DetachKeys string
 }
+
+// HTTPAttachStreams informs the HTTPAttach endpoint which of the container's
+// standard streams should be streamed to the client. If this is passed, at
+// least one of the streams must be set to true.
+type HTTPAttachStreams struct {
+	Stdin  bool
+	Stdout bool
+	Stderr bool
+}
diff --git a/libpod/oci_conmon_linux.go b/libpod/oci_conmon_linux.go
index 5ab0e73c4..0e8a64865 100644
--- a/libpod/oci_conmon_linux.go
+++ b/libpod/oci_conmon_linux.go
@@ -5,8 +5,11 @@ package libpod
 import (
 	"bufio"
 	"bytes"
+	"encoding/binary"
 	"fmt"
+	"io"
 	"io/ioutil"
+	"net"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -17,6 +20,7 @@ import (
 	"text/template"
 	"time"
 
+	conmonConfig "github.com/containers/conmon/runner/config"
 	"github.com/containers/libpod/libpod/config"
 	"github.com/containers/libpod/libpod/define"
 	"github.com/containers/libpod/pkg/cgroups"
@@ -33,6 +37,13 @@ import (
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"golang.org/x/sys/unix"
+	"k8s.io/client-go/tools/remotecommand"
+)
+
+const (
+	// This is Conmon's STDIO_BUF_SIZE. I don't believe we have access to it
+	// directly from the Go cose, so const it here
+	bufferSize = conmonConfig.BufSize
 )
 
 // ConmonOCIRuntime is an OCI runtime managed by Conmon.
@@ -465,6 +476,123 @@ func (r *ConmonOCIRuntime) UnpauseContainer(ctr *Container) error {
 	return utils.ExecCmdWithStdStreams(os.Stdin, os.Stdout, os.Stderr, env, r.path, "resume", ctr.ID())
 }
 
+// HTTPAttach performs an attach for the HTTP API.
+// This will consume, and automatically close, the hijacked HTTP session.
+// It is not necessary to close it independently.
+// The cancel channel is not closed; it is up to the caller to do so after
+// this function returns.
+// If this is a container with a terminal, we will stream raw. If it is not, we
+// will stream with an 8-byte header to multiplex STDOUT and STDERR.
+func (r *ConmonOCIRuntime) HTTPAttach(ctr *Container, httpConn net.Conn, httpBuf *bufio.ReadWriter, streams *HTTPAttachStreams, detachKeys *string, cancel <-chan bool) (deferredErr error) {
+	isTerminal := false
+	if ctr.config.Spec.Process != nil {
+		isTerminal = ctr.config.Spec.Process.Terminal
+	}
+
+	// Ensure that our contract of closing the HTTP connection is honored.
+	defer hijackWriteErrorAndClose(deferredErr, ctr.ID(), httpConn, httpBuf)
+
+	if streams != nil {
+		if isTerminal {
+			return errors.Wrapf(define.ErrInvalidArg, "cannot specify which streams to attach as container %s has a terminal", ctr.ID())
+		}
+		if !streams.Stdin && !streams.Stdout && !streams.Stderr {
+			return errors.Wrapf(define.ErrInvalidArg, "must specify at least one stream to attach to")
+		}
+	}
+
+	attachSock, err := r.AttachSocketPath(ctr)
+	if err != nil {
+		return err
+	}
+	socketPath := buildSocketPath(attachSock)
+
+	conn, err := net.DialUnix("unixpacket", nil, &net.UnixAddr{Name: socketPath, Net: "unixpacket"})
+	if err != nil {
+		return errors.Wrapf(err, "failed to connect to container's attach socket: %v", socketPath)
+	}
+	defer func() {
+		if err := conn.Close(); err != nil {
+			logrus.Errorf("unable to close container %s attach socket: %q", ctr.ID(), err)
+		}
+	}()
+
+	logrus.Debugf("Successfully connected to container %s attach socket %s", ctr.ID(), socketPath)
+
+	detachString := define.DefaultDetachKeys
+	if detachKeys != nil {
+		detachString = *detachKeys
+	}
+	detach, err := processDetachKeys(detachString)
+	if err != nil {
+		return err
+	}
+
+	// Make a channel to pass errors back
+	errChan := make(chan error)
+
+	attachStdout := true
+	attachStderr := true
+	attachStdin := true
+	if streams != nil {
+		attachStdout = streams.Stdout
+		attachStderr = streams.Stderr
+		attachStdin = streams.Stdin
+	}
+
+	// Handle STDOUT/STDERR
+	go func() {
+		var err error
+		if isTerminal {
+			logrus.Debugf("Performing terminal HTTP attach for container %s", ctr.ID())
+			err = httpAttachTerminalCopy(conn, httpBuf, ctr.ID())
+		} else {
+			logrus.Debugf("Performing non-terminal HTTP attach for container %s", ctr.ID())
+			err = httpAttachNonTerminalCopy(conn, httpBuf, ctr.ID(), attachStdin, attachStdout, attachStderr)
+		}
+		errChan <- err
+		logrus.Debugf("STDOUT/ERR copy completed")
+	}()
+	// Next, STDIN. Avoid entirely if attachStdin unset.
+	if attachStdin {
+		go func() {
+			_, err := utils.CopyDetachable(conn, httpBuf, detach)
+			logrus.Debugf("STDIN copy completed")
+			errChan <- err
+		}()
+	}
+
+	if cancel != nil {
+		select {
+		case err := <-errChan:
+			return err
+		case <-cancel:
+			return nil
+		}
+	} else {
+		var connErr error = <-errChan
+		return connErr
+	}
+}
+
+// AttachResize resizes the terminal used by the given container.
+func (r *ConmonOCIRuntime) AttachResize(ctr *Container, newSize remotecommand.TerminalSize) error {
+	// TODO: probably want a dedicated function to get ctl file path?
+	controlPath := filepath.Join(ctr.bundlePath(), "ctl")
+	controlFile, err := os.OpenFile(controlPath, unix.O_WRONLY, 0)
+	if err != nil {
+		return errors.Wrapf(err, "could not open ctl file for terminal resize")
+	}
+	defer controlFile.Close()
+
+	logrus.Debugf("Received a resize event for container %s: %+v", ctr.ID(), newSize)
+	if _, err = fmt.Fprintf(controlFile, "%d %d %d\n", 1, newSize.Height, newSize.Width); err != nil {
+		return errors.Wrapf(err, "failed to write to ctl file to resize terminal")
+	}
+
+	return nil
+}
+
 // ExecContainer executes a command in a running container
 // TODO: Split into Create/Start/Attach/Wait
 func (r *ConmonOCIRuntime) ExecContainer(c *Container, sessionID string, options *ExecOptions) (int, chan error, error) {
@@ -1465,3 +1593,139 @@ func (r *ConmonOCIRuntime) getOCIRuntimeVersion() (string, error) {
 	}
 	return strings.TrimSuffix(output, "\n"), nil
 }
+
+// Copy data from container to HTTP connection, for terminal attach.
+// Container is the container's attach socket connection, http is a buffer for
+// the HTTP connection. cid is the ID of the container the attach session is
+// running for (used solely for error messages).
+func httpAttachTerminalCopy(container *net.UnixConn, http *bufio.ReadWriter, cid string) error {
+	buf := make([]byte, bufferSize)
+	for {
+		numR, err := container.Read(buf)
+		if numR > 0 {
+			switch buf[0] {
+			case AttachPipeStdout:
+				// Do nothing
+			default:
+				logrus.Errorf("Received unexpected attach type %+d, discarding %d bytes", buf[0], numR)
+				continue
+			}
+
+			numW, err2 := http.Write(buf[1:numR])
+			if err2 != nil {
+				if err != nil {
+					logrus.Errorf("Error reading container %s STDOUT: %v", cid, err)
+				}
+				return err2
+			} else if numW+1 != numR {
+				return io.ErrShortWrite
+			}
+			// We need to force the buffer to write immediately, so
+			// there isn't a delay on the terminal side.
+			if err2 := http.Flush(); err2 != nil {
+				if err != nil {
+					logrus.Errorf("Error reading container %s STDOUT: %v", cid, err)
+				}
+				return err2
+			}
+		}
+		if err != nil {
+			if err == io.EOF {
+				return nil
+			}
+			return err
+		}
+	}
+}
+
+// Copy data from a container to an HTTP connection, for non-terminal attach.
+// Appends a header to multiplex input.
+func httpAttachNonTerminalCopy(container *net.UnixConn, http *bufio.ReadWriter, cid string, stdin, stdout, stderr bool) error {
+	buf := make([]byte, bufferSize)
+	for {
+		numR, err := container.Read(buf)
+		if numR > 0 {
+			headerBuf := []byte{0, 0, 0, 0}
+
+			// Practically speaking, we could make this buf[0] - 1,
+			// but we need to validate it anyways...
+			switch buf[0] {
+			case AttachPipeStdin:
+				headerBuf[0] = 0
+				if !stdin {
+					continue
+				}
+			case AttachPipeStdout:
+				if !stdout {
+					continue
+				}
+				headerBuf[0] = 1
+			case AttachPipeStderr:
+				if !stderr {
+					continue
+				}
+				headerBuf[0] = 2
+			default:
+				logrus.Errorf("Received unexpected attach type %+d, discarding %d bytes", buf[0], numR)
+				continue
+			}
+
+			// Get big-endian length and append.
+			// Subtract 1 because we strip the first byte (used for
+			// multiplexing by Conmon).
+			lenBuf := []byte{0, 0, 0, 0}
+			binary.BigEndian.PutUint32(lenBuf, uint32(numR-1))
+			headerBuf = append(headerBuf, lenBuf...)
+
+			numH, err2 := http.Write(headerBuf)
+			if err2 != nil {
+				if err != nil {
+					logrus.Errorf("Error reading container %s standard streams: %v", cid, err)
+				}
+
+				return err2
+			}
+			// Hardcoding header length is pretty gross, but
+			// fast. Should be safe, as this is a fixed part
+			// of the protocol.
+			if numH != 8 {
+				if err != nil {
+					logrus.Errorf("Error reading container %s standard streams: %v", cid, err)
+				}
+
+				return io.ErrShortWrite
+			}
+
+			numW, err2 := http.Write(buf[1:numR])
+			if err2 != nil {
+				if err != nil {
+					logrus.Errorf("Error reading container %s standard streams: %v", cid, err)
+				}
+
+				return err2
+			} else if numW+1 != numR {
+				if err != nil {
+					logrus.Errorf("Error reading container %s standard streams: %v", cid, err)
+				}
+
+				return io.ErrShortWrite
+			}
+			// We need to force the buffer to write immediately, so
+			// there isn't a delay on the terminal side.
+			if err2 := http.Flush(); err2 != nil {
+				if err != nil {
+					logrus.Errorf("Error reading container %s STDOUT: %v", cid, err)
+				}
+				return err2
+			}
+		}
+		if err != nil {
+			if err == io.EOF {
+				return nil
+			}
+
+			return err
+		}
+	}
+
+}
diff --git a/libpod/oci_missing.go b/libpod/oci_missing.go
index 0faa1805b..ff7eea625 100644
--- a/libpod/oci_missing.go
+++ b/libpod/oci_missing.go
@@ -1,13 +1,16 @@
 package libpod
 
 import (
+	"bufio"
 	"fmt"
+	"net"
 	"path/filepath"
 	"sync"
 
 	"github.com/containers/libpod/libpod/define"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
+	"k8s.io/client-go/tools/remotecommand"
 )
 
 var (
@@ -107,6 +110,16 @@ func (r *MissingRuntime) UnpauseContainer(ctr *Container) error {
 	return r.printError()
 }
 
+// HTTPAttach is not available as the runtime is missing
+func (r *MissingRuntime) HTTPAttach(ctr *Container, httpConn net.Conn, httpBuf *bufio.ReadWriter, streams *HTTPAttachStreams, detachKeys *string, cancel <-chan bool) error {
+	return r.printError()
+}
+
+// AttachResize is not available as the runtime is missing
+func (r *MissingRuntime) AttachResize(ctr *Container, newSize remotecommand.TerminalSize) error {
+	return r.printError()
+}
+
 // ExecContainer is not available as the runtime is missing
 func (r *MissingRuntime) ExecContainer(ctr *Container, sessionID string, options *ExecOptions) (int, chan error, error) {
 	return -1, nil, r.printError()
diff --git a/libpod/util.go b/libpod/util.go
index 30e5cd4c3..f79d6c09b 100644
--- a/libpod/util.go
+++ b/libpod/util.go
@@ -1,7 +1,9 @@
 package libpod
 
 import (
+	"bufio"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -16,6 +18,7 @@ import (
 	"github.com/fsnotify/fsnotify"
 	spec "github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 )
 
 // Runtime API constants
@@ -231,3 +234,20 @@ func checkDependencyContainer(depCtr, ctr *Container) error {
 
 	return nil
 }
+
+// hijackWriteErrorAndClose writes an error to a hijacked HTTP session and
+// closes it. Intended to HTTPAttach function.
+// If error is nil, it will not be written; we'll only close the connection.
+func hijackWriteErrorAndClose(toWrite error, cid string, httpCon io.Closer, httpBuf *bufio.ReadWriter) {
+	if toWrite != nil {
+		if _, err := httpBuf.Write([]byte(toWrite.Error())); err != nil {
+			logrus.Errorf("Error writing error %q to container %s HTTP attach connection: %v", toWrite, cid, err)
+		} else if err := httpBuf.Flush(); err != nil {
+			logrus.Errorf("Error flushing HTTP buffer for container %s HTTP attach connection: %v", cid, err)
+		}
+	}
+
+	if err := httpCon.Close(); err != nil {
+		logrus.Errorf("Error closing container %s HTTP attach connection: %v", cid, err)
+	}
+}