[WIP]

Signed-off-by: Matej Vašek <[email protected]>

Dockerfile.utils

@@ -22,7 +22,8 @@ RUN apk add --no-cache socat tar
 COPY --from=builder /workspace/func-util /usr/local/bin/
 RUN ln -s /usr/local/bin/func-util /usr/local/bin/deploy && \
     ln -s /usr/local/bin/func-util /usr/local/bin/scaffold && \
-    ln -s /usr/local/bin/func-util /usr/local/bin/s2i
+    ln -s /usr/local/bin/func-util /usr/local/bin/s2i && \
+    ln -s /usr/local/bin/func-util /usr/local/bin/socat
 
 LABEL \
    org.opencontainers.image.description="Knative Func Utils Image" \

cmd/func-util/main.go

@@ -44,6 +44,8 @@ func main() {
 		cmd = scaffold
 	case "s2i":
 		cmd = s2iCmd
+	case "socat":
+		cmd = socat
 	}
 
 	err := cmd(ctx)
@@ -57,6 +59,12 @@ func unknown(_ context.Context) error {
 	return fmt.Errorf("unknown command: " + os.Args[0])
 }
 
+func socat(ctx context.Context) error {
+	cmd := newSocatCmd()
+	cmd.SetContext(ctx)
+	return cmd.Execute()
+}
+
 func scaffold(ctx context.Context) error {
 
 	if len(os.Args) != 2 {

cmd/func-util/socat.go

@@ -0,0 +1,139 @@
+package main
+
+import (
+	"fmt"
+	"golang.org/x/sync/errgroup"
+	"io"
+	"net"
+	"os"
+	"strings"
+
+	"github.com/spf13/cobra"
+)
+
+func newSocatCmd() *cobra.Command {
+	var uniDir bool
+	cmd := cobra.Command{
+		Use:   "socat [-u] <address> <address>",
+		Short: "Minimalistic socat.",
+		Long: `Minimalistic socat.
+Implements only TCP, OPEN and stdio ("-") addresses with no options.
+Only supported flag is -u.`,
+		Args: cobra.ExactArgs(2),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			stdio := rwc{
+				ReadCloser:  cmd.InOrStdin().(io.ReadCloser),
+				WriteCloser: cmd.OutOrStdout().(io.WriteCloser),
+			}
+			left, err := createConnection(args[0], stdio)
+			if err != nil {
+				return err
+			}
+			defer left.Close()
+			right, err := createConnection(args[1], stdio)
+			if err != nil {
+				return err
+			}
+			defer right.Close()
+			return connect(left, right, uniDir)
+		},
+	}
+
+	var dbg string
+	cmd.Flags().BoolVarP(&uniDir, "unidirect", "u", false, "unidirectional mode (left to right)")
+	cmd.Flags().StringVarP(&dbg, "debug", "d", "", "log level")
+
+	return &cmd
+}
+
+func createConnection(address string, stdio connection) (connection, error) {
+	if address == "-" {
+		return stdio, nil
+	}
+	parts := strings.SplitN(address, ":", 2)
+	if len(parts) != 2 {
+		return nil, fmt.Errorf("cannot parse address: %q", address)
+	}
+	typ := strings.ToLower(parts[0])
+	parts = strings.Split(parts[1], ",")
+	if len(parts) > 1 {
+		_, _ = fmt.Fprintf(os.Stderr, "ignored options: %q\n", parts[1])
+	}
+	addr := parts[0]
+	switch typ {
+	case "tcp", "tcp4", "tcp6":
+		_, _ = fmt.Fprintln(os.Stderr, "opening connection")
+		var laddr net.TCPAddr
+		raddr, err := net.ResolveTCPAddr(typ, addr)
+		if err != nil {
+			return nil, fmt.Errorf("name does not resolve: %w", err)
+		}
+
+		conn, err := net.DialTCP(typ, &laddr, raddr)
+		if err == nil {
+			_, _ = fmt.Fprintf(os.Stderr, "successfully connected\n\n")
+		}
+		return conn, err
+	case "open":
+		return os.OpenFile(addr, os.O_RDWR, 0644)
+	default:
+		return nil, fmt.Errorf("unsupported address: %q", address)
+	}
+}
+
+func connect(left, right connection, uniDir bool) error {
+	g := errgroup.Group{}
+	g.SetLimit(2)
+
+	if !uniDir {
+		g.Go(func() error {
+			_, err := io.Copy(left, right)
+			tryCloseWriteSide(left)
+			return err
+		})
+	}
+
+	g.Go(func() error {
+		_, err := io.Copy(right, left)
+		tryCloseWriteSide(right)
+		return err
+	})
+
+	return g.Wait()
+}
+
+type connection interface {
+	io.Reader
+	io.Writer
+	io.Closer
+}
+
+type writeCloser interface {
+	CloseWrite() error
+}
+
+type rwc struct {
+	io.ReadCloser
+	io.WriteCloser
+}
+
+func (r rwc) Close() error {
+	err := r.WriteCloser.Close()
+	if err != nil {
+		return err
+	}
+	return r.ReadCloser.Close()
+}
+
+func (r rwc) CloseWrite() error {
+	return r.WriteCloser.Close()
+}
+
+func tryCloseWriteSide(c connection) {
+	if wc, ok := c.(writeCloser); ok {
+		err := wc.CloseWrite()
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "waring: cannot close write side: %+v\n", err)
+		}
+	}
+}

cmd/func-util/socat_test.go

@@ -0,0 +1,224 @@
+package main
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"net"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func TestRootCmd(t *testing.T) {
+
+	/* Begin prepare TCP server and the files */
+	addr := startTCPEcho(t)
+
+	const testData = "file-content\n"
+	tmpDir := t.TempDir()
+	inputFile := filepath.Join(tmpDir, "a.txt")
+	err := os.WriteFile(inputFile, []byte(testData), 0644)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	outputFile := filepath.Join(tmpDir, "b.txt")
+	err = os.WriteFile(outputFile, []byte{}, 0644)
+	if err != nil {
+		t.Fatal(err)
+	}
+	/* End prepare TCP server and the files */
+
+	type matcher = func(string) bool
+	contains := func(pattern string) func(string) bool {
+		return func(s string) bool { return strings.Contains(s, pattern) }
+	}
+	equalsTo := func(pattern string) func(string) bool {
+		return func(s string) bool { return s == pattern }
+	}
+
+	type args struct {
+		args           []string
+		inputString    string
+		outMatcher     matcher
+		errOutMatcher  matcher
+		outFileMatcher matcher
+		wantErr        bool
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		{
+			name: "stdio<->tcp",
+			args: args{
+				args:        []string{"-", "TCP:" + addr},
+				inputString: testData,
+				outMatcher:  equalsTo(testData),
+			},
+		},
+		{
+			name: "tcp<->stdio",
+			args: args{
+				args:        []string{"TCP:" + addr, "-"},
+				inputString: testData,
+				outMatcher:  equalsTo(testData),
+			},
+		},
+		{
+			name: "tcp-no-such-host",
+			args: args{
+				args:          []string{"-", "TCP:does.not.exist:10000"},
+				inputString:   "tcp-echo",
+				errOutMatcher: contains("not resolve"),
+				wantErr:       true,
+			},
+		},
+		{
+			name: "file->stdio",
+			args: args{
+				args:        []string{"-u", "OPEN:" + inputFile, "-"},
+				inputString: "",
+				outMatcher:  equalsTo(testData),
+			},
+		},
+		{
+			name: "stdio->file",
+			args: args{
+				args:           []string{"-u", "-", "OPEN:" + outputFile},
+				inputString:    testData,
+				outFileMatcher: equalsTo(testData),
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var out, errOut bytes.Buffer
+
+			stdout := &testWriter{Writer: &out}
+			stderr := &testWriter{Writer: &errOut}
+			cmd := newSocatCmd()
+			cmd.SetIn(io.NopCloser(strings.NewReader(tt.args.inputString)))
+			cmd.SetOut(stdout)
+			cmd.SetErr(stderr)
+			cmd.SetArgs(tt.args.args)
+
+			err = cmd.Execute()
+			if err != nil && !tt.args.wantErr {
+				t.Error(err)
+				t.Logf("errOut: %q", errOut.String())
+			}
+
+			if err == nil && tt.args.wantErr {
+				t.Error("expected error but got nil")
+			}
+
+			if tt.args.outMatcher != nil && !tt.args.outMatcher(out.String()) {
+				t.Error("bad standard output")
+			}
+			if tt.args.errOutMatcher != nil && !tt.args.errOutMatcher(errOut.String()) {
+				t.Error("bad standard error output")
+			}
+			if tt.args.outFileMatcher != nil {
+				bs, e := os.ReadFile(outputFile)
+				if e != nil {
+					t.Fatal(e)
+				}
+				if !tt.args.outFileMatcher(string(bs)) {
+					t.Error("bad content of the output file")
+				}
+			}
+		})
+	}
+}
+
+type testWriter struct {
+	io.Writer
+}
+
+func (n *testWriter) Close() error {
+	return nil
+}
+
+func startTCPEcho(t *testing.T) (addr string) {
+	l, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	addr = l.Addr().String()
+	go func() {
+		for {
+			conn, err := l.Accept()
+			if err != nil {
+				if errors.Is(err, net.ErrClosed) {
+					return
+				}
+				panic(err)
+			}
+			go func(conn net.Conn) {
+				defer conn.Close()
+				_, err = io.Copy(conn, conn)
+				if err != nil {
+					panic(err)
+				}
+			}(conn)
+		}
+	}()
+	t.Cleanup(func() {
+		l.Close()
+	})
+	return addr
+}
+
+func TestNewRootCmdWithPipe(t *testing.T) {
+	addr := startTCPEcho(t)
+
+	r, stdOut, err := os.Pipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	stdIn, w, err := os.Pipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var data = []byte("testing data")
+
+	go func() {
+		_, err = w.Write(data)
+		if err != nil {
+			t.Error(err)
+		}
+		err = w.Close()
+		if err != nil {
+			t.Error(err)
+		}
+	}()
+
+	go func() {
+		var errBuff bytes.Buffer
+		cmd := newSocatCmd()
+		cmd.SetIn(stdIn)
+		cmd.SetOut(stdOut)
+		cmd.SetErr(&errBuff)
+		cmd.SetArgs([]string{"-dd", "-", "TCP:" + addr})
+
+		err = cmd.Execute()
+		if err != nil {
+			t.Error(err)
+		}
+
+	}()
+
+	bs, e := io.ReadAll(r)
+	if e != nil {
+		t.Error(e)
+	}
+	t.Log(string(data))
+	if !bytes.Equal(data, bs) {
+		t.Errorf("bad data: %q", string(bs))
+	}
+}