Merge pull request #89 from AkihiroSuda/b

port/builtin: refactor (no substantial change)

pkg/port/builtin/child/child.go

@@ -0,0 +1,134 @@
+package child
+
+import (
+	"fmt"
+	"io"
+	"net"
+	"os"
+
+	"github.com/pkg/errors"
+	"golang.org/x/sys/unix"
+
+	"github.com/rootless-containers/rootlesskit/pkg/msgutil"
+	"github.com/rootless-containers/rootlesskit/pkg/port"
+	"github.com/rootless-containers/rootlesskit/pkg/port/builtin/msg"
+	opaquepkg "github.com/rootless-containers/rootlesskit/pkg/port/builtin/opaque"
+)
+
+func NewDriver(logWriter io.Writer) port.ChildDriver {
+	return &childDriver{
+		logWriter: logWriter,
+	}
+}
+
+type childDriver struct {
+	logWriter io.Writer
+}
+
+func (d *childDriver) RunChildDriver(opaque map[string]string, quit <-chan struct{}) error {
+	socketPath := opaque[opaquepkg.SocketPath]
+	if socketPath == "" {
+		return errors.New("socket path not set")
+	}
+	childReadyPipePath := opaque[opaquepkg.ChildReadyPipePath]
+	if childReadyPipePath == "" {
+		return errors.New("child ready pipe path not set")
+	}
+	childReadyPipeW, err := os.OpenFile(childReadyPipePath, os.O_WRONLY, os.ModeNamedPipe)
+	if err != nil {
+		return err
+	}
+	ln, err := net.ListenUnix("unix", &net.UnixAddr{
+		Name: socketPath,
+		Net:  "unix",
+	})
+	if err != nil {
+		return err
+	}
+	// write nothing, just close
+	if err = childReadyPipeW.Close(); err != nil {
+		return err
+	}
+	stopAccept := make(chan struct{}, 1)
+	go func() {
+		<-quit
+		stopAccept <- struct{}{}
+		ln.Close()
+	}()
+	for {
+		c, err := ln.AcceptUnix()
+		if err != nil {
+			select {
+			case <-stopAccept:
+				return nil
+			default:
+			}
+			return err
+		}
+		go func() {
+			if rerr := d.routine(c); rerr != nil {
+				rep := msg.Reply{
+					Error: rerr.Error(),
+				}
+				msgutil.MarshalToWriter(c, &rep)
+			}
+			c.Close()
+		}()
+	}
+	return nil
+}
+
+func (d *childDriver) routine(c *net.UnixConn) error {
+	var req msg.Request
+	if _, err := msgutil.UnmarshalFromReader(c, &req); err != nil {
+		return err
+	}
+	switch req.Type {
+	case msg.RequestTypeInit:
+		return d.handleConnectInit(c, &req)
+	case msg.RequestTypeConnect:
+		return d.handleConnectRequest(c, &req)
+	default:
+		return errors.Errorf("unknown request type %q", req.Type)
+	}
+}
+
+func (d *childDriver) handleConnectInit(c *net.UnixConn, req *msg.Request) error {
+	_, err := msgutil.MarshalToWriter(c, nil)
+	return err
+}
+
+func (d *childDriver) handleConnectRequest(c *net.UnixConn, req *msg.Request) error {
+	switch req.Proto {
+	case "tcp":
+	case "udp":
+	default:
+		return errors.Errorf("unknown proto: %q", req.Proto)
+	}
+	var dialer net.Dialer
+	targetConn, err := dialer.Dial(req.Proto, fmt.Sprintf("127.0.0.1:%d", req.Port))
+	if err != nil {
+		return err
+	}
+	defer targetConn.Close() // no effect on duplicated FD
+	targetConnFiler, ok := targetConn.(filer)
+	if !ok {
+		return errors.Errorf("unknown target connection: %+v", targetConn)
+	}
+	targetConnFile, err := targetConnFiler.File()
+	if err != nil {
+		return err
+	}
+	oob := unix.UnixRights(int(targetConnFile.Fd()))
+	f, err := c.File()
+	if err != nil {
+		return err
+	}
+	err = unix.Sendmsg(int(f.Fd()), []byte("dummy"), oob, nil, 0)
+	return err
+}
+
+// filer is implemented by *net.TCPConn and *net.UDPConn
+type filer interface {
+	File() (f *os.File, err error)
+}

pkg/port/builtin/msg/msg.go

@@ -0,0 +1,129 @@
+package msg
+
+import (
+	"net"
+	"time"
+
+	"github.com/pkg/errors"
+	"golang.org/x/sys/unix"
+
+	"github.com/rootless-containers/rootlesskit/pkg/msgutil"
+	"github.com/rootless-containers/rootlesskit/pkg/port"
+)
+
+const (
+	RequestTypeInit    = "init"
+	RequestTypeConnect = "connect"
+)
+
+// Request and Response are encoded as JSON with uint32le length header.
+type Request struct {
+	Type  string // "init" or "connect"
+	Proto string // "tcp" or "udp"
+	Port  int
+}
+
+// Reply may contain FD as OOB
+type Reply struct {
+	Error string
+}
+
+// Initiate sends "init" request to the child UNIX socket.
+func Initiate(c *net.UnixConn) error {
+	req := Request{
+		Type: RequestTypeInit,
+	}
+	if _, err := msgutil.MarshalToWriter(c, &req); err != nil {
+		return err
+	}
+	if err := c.CloseWrite(); err != nil {
+		return err
+	}
+	var rep Reply
+	if _, err := msgutil.UnmarshalFromReader(c, &rep); err != nil {
+		return err
+	}
+	return c.CloseRead()
+}
+
+// ConnectToChild connects to the child UNIX socket, and obtains TCP or UDP socket FD
+// that corresponds to the port spec.
+func ConnectToChild(c *net.UnixConn, spec port.Spec) (int, error) {
+	req := Request{
+		Type:  RequestTypeConnect,
+		Proto: spec.Proto,
+		Port:  spec.ChildPort,
+	}
+	if _, err := msgutil.MarshalToWriter(c, &req); err != nil {
+		return 0, err
+	}
+	if err := c.CloseWrite(); err != nil {
+		return 0, err
+	}
+	oobSpace := unix.CmsgSpace(4)
+	oob := make([]byte, oobSpace)
+	_, oobN, _, _, err := c.ReadMsgUnix(nil, oob)
+	if err != nil {
+		return 0, err
+	}
+	if oobN != oobSpace {
+		return 0, errors.Errorf("expected OOB space %d, got %d", oobSpace, oobN)
+	}
+	oob = oob[:oobN]
+	fd, err := parseFDFromOOB(oob)
+	if err != nil {
+		return 0, err
+	}
+	if err := c.CloseRead(); err != nil {
+		return 0, err
+	}
+	return fd, nil
+}
+
+// ConnectToChildWithSocketPath wraps ConnectToChild
+func ConnectToChildWithSocketPath(socketPath string, spec port.Spec) (int, error) {
+	var dialer net.Dialer
+	conn, err := dialer.Dial("unix", socketPath)
+	if err != nil {
+		return 0, err
+	}
+	defer conn.Close()
+	c := conn.(*net.UnixConn)
+	return ConnectToChild(c, spec)
+}
+
+// ConnectToChildWithRetry retries ConnectToChild every (i*5) milliseconds.
+func ConnectToChildWithRetry(socketPath string, spec port.Spec, retries int) (int, error) {
+	for i := 0; i < retries; i++ {
+		fd, err := ConnectToChildWithSocketPath(socketPath, spec)
+		if i == retries-1 && err != nil {
+			return 0, err
+		}
+		if err == nil {
+			return fd, err
+		}
+		// TODO: backoff
+		time.Sleep(time.Duration(i*5) * time.Millisecond)
+	}
+	// NOT REACHED
+	return 0, errors.New("reached max retry")
+}
+
+func parseFDFromOOB(oob []byte) (int, error) {
+	scms, err := unix.ParseSocketControlMessage(oob)
+	if err != nil {
+		return 0, err
+	}
+	if len(scms) != 1 {
+		return 0, errors.Errorf("unexpected scms: %v", scms)
+	}
+	scm := scms[0]
+	fds, err := unix.ParseUnixRights(&scm)
+	if err != nil {
+		return 0, err
+	}
+	if len(fds) != 1 {
+		return 0, errors.Errorf("unexpected fds: %v", fds)
+	}
+	return fds[0], nil
+}

pkg/port/builtin/opaque/opaque.go

@@ -0,0 +1,6 @@
+package opaque
+
+const (
+	SocketPath         = "builtin.socketpath"
+	ChildReadyPipePath = "builtin.readypipepath"
+)