recover sshpiper.go

ssh/sshpiper.go

@@ -0,0 +1,656 @@
+// Copyright 2014, 2015 tgic<farmer1992@gmail.com>. All rights reserved.
+// this file is governed by MIT-license
+//
+// https://github.com/tg123/sshpiper
+
+package ssh
+
+import (
+	"errors"
+	"fmt"
+	"net"
+)
+
+// SSHPiperConfig holds SSHPiper specific configuration data.
+type SSHPiperConfig struct {
+	Config
+
+	hostKeys []Signer
+
+	// AdditionalChallenge, if non-nil, is called before calling FindUpstream.
+	// This allows you do a KeyboardInteractiveChallenge before connecting to upstream.
+	// It must return true if downstream passed the challenge, otherwise,
+	// the piped connection will be closed.
+	AdditionalChallenge func(conn ConnMetadata, client KeyboardInteractiveChallenge) (bool, error)
+
+	// FindUpstream, must not be nil, is called when SSHPiper decided to establish a
+	// ssh connection to upstream server.  a connection, net.Conn, to upstream
+	// and upstream username should be returned.
+	// SSHPiper will use the username from downstream if empty username is returned.
+	// If any error occurs, the piped connection will be closed.
+	FindUpstream func(conn ConnMetadata) (net.Conn, string, error)
+
+	// MapPublicKey, if non-nil, is called when downstream requests a publickey auth.
+	// SSHPiper will sign the auth packet message using the returned Signer.
+	// This func might be called twice, one is for query message, the other
+	// is real auth packet message.
+	// If any error occurs during this period, a NoneAuth packet will be sent to
+	// upstream ssh server instead.
+	//
+	// More info: https://github.com/tg123/sshpiper#publickey-sign-again
+	MapPublicKey func(conn ConnMetadata, key PublicKey) (Signer, error)
+}
+
+type upstream struct{ *connection }
+type downstream struct{ *connection }
+
+type pipedConn struct {
+	upstream   *upstream
+	downstream *downstream
+
+	processAuthMsg func(msg *userAuthRequestMsg) (*userAuthRequestMsg, error)
+
+	hookUpstreamMsg   func(msg []byte) ([]byte, error)
+	hookDownstreamMsg func(msg []byte) ([]byte, error)
+}
+
+// SSHPiperConn is a piped SSH connection, linking upstream ssh server and
+// downstream ssh client together. After the piped connection was created,
+// The downstream ssh client is authenticated by upstream ssh server and
+// AdditionalChallenge from SSHPiper.
+type SSHPiperConn struct {
+	*pipedConn
+
+	HookUpstreamMsg   func(conn ConnMetadata, msg []byte) ([]byte, error)
+	HookDownstreamMsg func(conn ConnMetadata, msg []byte) ([]byte, error)
+}
+
+// Wait blocks until the piped connection has shut down, and returns the
+// error causing the shutdown.
+func (p *SSHPiperConn) Wait() error {
+
+	p.pipedConn.hookUpstreamMsg = func(msg []byte) ([]byte, error) {
+		if p.HookUpstreamMsg != nil {
+			// api always using p.downstream as conn meta
+			return p.HookUpstreamMsg(p.downstream, msg)
+		}
+
+		return msg, nil
+	}
+
+	p.pipedConn.hookDownstreamMsg = func(msg []byte) ([]byte, error) {
+		if p.HookDownstreamMsg != nil {
+			return p.HookDownstreamMsg(p.downstream, msg)
+		}
+
+		return msg, nil
+	}
+
+	return p.pipedConn.loop()
+}
+
+// Close the piped connection create by SSHPiper
+func (p *SSHPiperConn) Close() {
+	p.pipedConn.Close()
+}
+
+func (p *SSHPiperConn) UpstreamConnMeta() ConnMetadata {
+	return p.pipedConn.upstream
+}
+
+func (p *SSHPiperConn) DownstreamConnMeta() ConnMetadata {
+	return p.pipedConn.downstream
+}
+
+// AddHostKey adds a private key as a SSHPiper host key. If an existing host
+// key exists with the same algorithm, it is overwritten. Each SSHPiper
+// config must have at least one host key.
+func (s *SSHPiperConfig) AddHostKey(key Signer) {
+	for i, k := range s.hostKeys {
+		if k.PublicKey().Type() == key.PublicKey().Type() {
+			s.hostKeys[i] = key
+			return
+		}
+	}
+
+	s.hostKeys = append(s.hostKeys, key)
+}
+
+// NewSSHPiperConn starts a piped ssh connection witch conn as its downstream transport.
+// It handshake with downstream ssh client and upstream ssh server provicde by FindUpstream.
+// If either handshake is unsuccessful, the whole piped connection will be closed.
+func NewSSHPiperConn(conn net.Conn, piper *SSHPiperConfig) (pipe *SSHPiperConn, err error) {
+
+	if piper.FindUpstream == nil {
+		panic("FindUpstream func not found")
+	}
+
+	d, err := newDownstream(conn, &ServerConfig{
+		Config:   piper.Config,
+		hostKeys: piper.hostKeys,
+	})
+	if err != nil {
+		return nil, err
+	}
+	defer func() {
+		if pipe == nil {
+			d.Close()
+		}
+	}()
+
+	userAuthReq, err := d.nextAuthMsg()
+	if err != nil {
+		return nil, err
+	}
+
+	d.user = userAuthReq.User
+
+	// need additional challenge
+	if piper.AdditionalChallenge != nil {
+
+		for {
+			err := d.transport.writePacket(Marshal(&userAuthFailureMsg{
+				Methods: []string{"keyboard-interactive"},
+			}))
+
+			if err != nil {
+				return nil, err
+			}
+
+			userAuthReq, err := d.nextAuthMsg()
+
+			if err != nil {
+				return nil, err
+			}
+
+			if userAuthReq.Method == "keyboard-interactive" {
+				break
+			}
+		}
+
+		prompter := &sshClientKeyboardInteractive{d.connection}
+		ok, err := piper.AdditionalChallenge(d, prompter.Challenge)
+
+		if err != nil {
+			return nil, err
+		}
+
+		if !ok {
+			return nil, fmt.Errorf("additional challenge failed")
+		}
+	}
+
+	upconn, mappedUser, err := piper.FindUpstream(d)
+	if err != nil {
+		return nil, err
+	}
+
+	addr := upconn.RemoteAddr().String()
+
+	if mappedUser == "" {
+		mappedUser = d.user
+	}
+
+	u, err := newUpstream(upconn, addr, &ClientConfig{})
+	if err != nil {
+		return nil, err
+	}
+	defer func() {
+		if pipe == nil {
+			u.Close()
+		}
+	}()
+
+	p := &pipedConn{
+		upstream:   u,
+		downstream: d,
+	}
+
+	p.processAuthMsg = func(msg *userAuthRequestMsg) (*userAuthRequestMsg, error) {
+
+		// only public msg need
+		if msg.Method != "publickey" || piper.MapPublicKey == nil {
+			msg.User = mappedUser
+			return msg, nil
+		}
+
+		user := msg.User
+		// pubKey MAP
+		downKey, isQuery, sig, err := parsePublicKeyMsg(msg)
+		if err != nil {
+			return nil, err
+		}
+
+		signer, err := piper.MapPublicKey(d, downKey)
+
+		// no mapped user change it to none or error occur
+		if err != nil || signer == nil {
+			return noneAuthMsg(user), nil
+		}
+
+		upKey := signer.PublicKey()
+
+		if isQuery {
+			// reply for query msg
+			msg, err = p.validAndAck(mappedUser, upKey, downKey)
+		} else {
+
+			ok, err := p.checkPublicKey(msg, downKey, sig)
+
+			if err != nil {
+				return nil, err
+			}
+
+			if !ok {
+				return noneAuthMsg(user), nil
+			}
+
+			msg, err = p.signAgain(mappedUser, msg, signer, downKey)
+		}
+
+		if err != nil {
+			return nil, err
+		}
+
+		return msg, nil
+	}
+
+	err = p.pipeAuth(userAuthReq)
+	if err != nil {
+		return nil, err
+	}
+
+	return &SSHPiperConn{pipedConn: p}, nil
+}
+
+func (pipe *pipedConn) validAndAck(user string, upKey, downKey PublicKey) (*userAuthRequestMsg, error) {
+
+	ok, err := validateKey(upKey, user, pipe.upstream.transport)
+
+	if ok {
+		okMsg := userAuthPubKeyOkMsg{
+			Algo:   downKey.Type(),
+			PubKey: downKey.Marshal(),
+		}
+
+		if err = pipe.downstream.transport.writePacket(Marshal(&okMsg)); err != nil {
+			return nil, err
+		}
+
+		return nil, nil
+	}
+
+	return noneAuthMsg(user), nil
+}
+
+func (pipe *pipedConn) checkPublicKey(msg *userAuthRequestMsg, pubkey PublicKey, sig *Signature) (bool, error) {
+
+	if !isAcceptableAlgo(sig.Format) {
+		return false, nil
+	}
+	signedData := buildDataSignedForAuth(pipe.downstream.transport.getSessionID(), *msg, []byte(pubkey.Type()), pubkey.Marshal())
+
+	if err := pubkey.Verify(signedData, sig); err != nil {
+		return false, nil
+	}
+
+	return true, nil
+}
+
+func (pipe *pipedConn) signAgain(user string, msg *userAuthRequestMsg, signer Signer, downKey PublicKey) (*userAuthRequestMsg, error) {
+
+	rand := pipe.upstream.transport.config.Rand
+	session := pipe.upstream.transport.getSessionID()
+
+	upKey := signer.PublicKey()
+	upKeyData := upKey.Marshal()
+
+	sign, err := signer.Sign(rand, buildDataSignedForAuth(session, userAuthRequestMsg{
+		User:    user,
+		Service: serviceSSH,
+		Method:  "publickey",
+	}, []byte(upKey.Type()), upKeyData))
+	if err != nil {
+		return nil, err
+	}
+
+	// manually wrap the serialized signature in a string
+	s := Marshal(sign)
+	sig := make([]byte, stringLength(len(s)))
+	marshalString(sig, s)
+
+	pubkeyMsg := &publickeyAuthMsg{
+		User:     user,
+		Service:  serviceSSH,
+		Method:   "publickey",
+		HasSig:   true,
+		Algoname: upKey.Type(),
+		PubKey:   upKeyData,
+		Sig:      sig,
+	}
+
+	Unmarshal(Marshal(pubkeyMsg), msg)
+
+	return msg, nil
+}
+
+func parsePublicKeyMsg(userAuthReq *userAuthRequestMsg) (PublicKey, bool, *Signature, error) {
+	if userAuthReq.Method != "publickey" {
+		return nil, false, nil, fmt.Errorf("not a publickey auth msg")
+	}
+
+	payload := userAuthReq.Payload
+	if len(payload) < 1 {
+		return nil, false, nil, parseError(msgUserAuthRequest)
+	}
+	isQuery := payload[0] == 0
+	payload = payload[1:]
+	algoBytes, payload, ok := parseString(payload)
+	if !ok {
+		return nil, false, nil, parseError(msgUserAuthRequest)
+	}
+	algo := string(algoBytes)
+	if !isAcceptableAlgo(algo) {
+		return nil, false, nil, fmt.Errorf("ssh: algorithm %q not accepted", algo)
+	}
+
+	pubKeyData, payload, ok := parseString(payload)
+	if !ok {
+		return nil, false, nil, parseError(msgUserAuthRequest)
+	}
+
+	pubKey, err := ParsePublicKey(pubKeyData)
+	if err != nil {
+		return nil, false, nil, err
+	}
+
+	var sig *Signature
+	if !isQuery {
+		sig, payload, ok = parseSignature(payload)
+		if !ok || len(payload) > 0 {
+			return nil, false, nil, parseError(msgUserAuthRequest)
+		}
+	}
+
+	return pubKey, isQuery, sig, nil
+}
+
+func piping(dst, src packetConn, hooker func(msg []byte) ([]byte, error)) error {
+	for {
+		p, err := src.readPacket()
+
+		if err != nil {
+			return err
+		}
+
+		p, err = hooker(p)
+
+		if err != nil {
+			return err
+		}
+
+		err = dst.writePacket(p)
+
+		if err != nil {
+			return err
+		}
+	}
+}
+
+func (pipe *pipedConn) loop() error {
+	c := make(chan error)
+
+	go func() {
+		c <- piping(pipe.upstream.transport, pipe.downstream.transport, pipe.hookDownstreamMsg)
+	}()
+
+	go func() {
+		c <- piping(pipe.downstream.transport, pipe.upstream.transport, pipe.hookUpstreamMsg)
+	}()
+
+	defer pipe.Close()
+
+	// wait until either connection closed
+	return <-c
+}
+
+func (pipe *pipedConn) Close() {
+	pipe.upstream.transport.Close()
+	pipe.downstream.transport.Close()
+}
+
+func (pipe *pipedConn) pipeAuth(initUserAuthMsg *userAuthRequestMsg) error {
+	err := pipe.upstream.sendAuthReq()
+	if err != nil {
+		return err
+	}
+
+	userAuthMsg := initUserAuthMsg
+
+	for {
+		// hook msg
+		userAuthMsg, err = pipe.processAuthMsg(userAuthMsg)
+
+		if err != nil {
+			return err
+		}
+
+		// nil for ignore
+		if userAuthMsg != nil {
+			err = pipe.upstream.transport.writePacket(Marshal(userAuthMsg))
+			if err != nil {
+				return err
+			}
+
+			packet, err := pipe.upstream.transport.readPacket()
+			if err != nil {
+				return err
+			}
+
+			success := packet[0] == msgUserAuthSuccess || packet[0] == msgUserAuthBanner
+
+			if err = pipe.downstream.transport.writePacket(packet); err != nil {
+				return err
+			}
+
+			if success {
+				return nil
+			}
+		}
+
+		var packet []byte
+
+		for {
+
+			if packet, err = pipe.downstream.transport.readPacket(); err != nil {
+				return err
+			}
+
+			// we can only handle auth req at the moment
+			if packet[0] == msgUserAuthRequest {
+				break
+			}
+
+			// pipe to auth succ if not a authreq
+			// typically, authinfo see RFC 4256
+			// TODO support hook this msg
+			err = pipe.upstream.transport.writePacket(packet)
+			if err != nil {
+				return err
+			}
+
+			// TODO clean up dup code
+			packet, err := pipe.upstream.transport.readPacket()
+			if err != nil {
+				return err
+			}
+
+			success := packet[0] == msgUserAuthSuccess || packet[0] == msgUserAuthBanner
+
+			if err = pipe.downstream.transport.writePacket(packet); err != nil {
+				return err
+			}
+
+			if success {
+				return nil
+			}
+		}
+
+		var userAuthReq userAuthRequestMsg
+
+		if err = Unmarshal(packet, &userAuthReq); err != nil {
+			return err
+		}
+
+		userAuthMsg = &userAuthReq
+
+	}
+}
+
+func (u *upstream) sendAuthReq() error {
+	if err := u.transport.writePacket(Marshal(&serviceRequestMsg{serviceUserAuth})); err != nil {
+		return err
+	}
+
+	packet, err := u.transport.readPacket()
+	if err != nil {
+		return err
+	}
+	var serviceAccept serviceAcceptMsg
+	if err := Unmarshal(packet, &serviceAccept); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func newDownstream(c net.Conn, config *ServerConfig) (*downstream, error) {
+	fullConf := *config
+	fullConf.SetDefaults()
+
+	s := &connection{
+		sshConn: sshConn{conn: c},
+	}
+
+	_, err := s.serverHandshakeNoAuth(&fullConf)
+	if err != nil {
+		c.Close()
+		return nil, err
+	}
+
+	return &downstream{s}, nil
+}
+
+func newUpstream(c net.Conn, addr string, config *ClientConfig) (*upstream, error) {
+	fullConf := *config
+	fullConf.SetDefaults()
+
+	conn := &connection{
+		sshConn: sshConn{conn: c},
+	}
+
+	if err := conn.clientHandshakeNoAuth(addr, &fullConf); err != nil {
+		c.Close()
+		return nil, err
+	}
+
+	return &upstream{conn}, nil
+}
+
+func (d *downstream) nextAuthMsg() (*userAuthRequestMsg, error) {
+	var userAuthReq userAuthRequestMsg
+
+	if packet, err := d.transport.readPacket(); err != nil {
+		return nil, err
+	} else if err = Unmarshal(packet, &userAuthReq); err != nil {
+		return nil, err
+	}
+
+	if userAuthReq.Service != serviceSSH {
+		return nil, errors.New("ssh: client attempted to negotiate for unknown service: " + userAuthReq.Service)
+	}
+
+	return &userAuthReq, nil
+}
+
+func noneAuthMsg(user string) *userAuthRequestMsg {
+	return &userAuthRequestMsg{
+		User:    user,
+		Service: serviceSSH,
+		Method:  "none",
+	}
+}
+
+func (c *connection) clientHandshakeNoAuth(dialAddress string, config *ClientConfig) error {
+	c.clientVersion = []byte(packageVersion)
+	if config.ClientVersion != "" {
+		c.clientVersion = []byte(config.ClientVersion)
+	}
+
+	var err error
+	c.serverVersion, err = exchangeVersions(c.sshConn.conn, c.clientVersion)
+	if err != nil {
+		return err
+	}
+
+	c.transport = newClientTransport(
+		newTransport(c.sshConn.conn, config.Rand, true /* is client */),
+		c.clientVersion, c.serverVersion, config, dialAddress, c.sshConn.RemoteAddr())
+	if err := c.transport.requestKeyChange(); err != nil {
+		return err
+	}
+
+	if packet, err := c.transport.readPacket(); err != nil {
+		return err
+	} else if packet[0] != msgNewKeys {
+		return unexpectedMessageError(msgNewKeys, packet[0])
+	}
+	return nil
+}
+
+func (s *connection) serverHandshakeNoAuth(config *ServerConfig) (*Permissions, error) {
+	if len(config.hostKeys) == 0 {
+		return nil, errors.New("ssh: server has no host keys")
+	}
+
+	var err error
+	s.serverVersion = []byte("SSH-2.0-SSHPiper")
+	s.clientVersion, err = exchangeVersions(s.sshConn.conn, s.serverVersion)
+	if err != nil {
+		return nil, err
+	}
+
+	tr := newTransport(s.sshConn.conn, config.Rand, false /* not client */)
+	s.transport = newServerTransport(tr, s.clientVersion, s.serverVersion, config)
+
+	if err := s.transport.requestKeyChange(); err != nil {
+		return nil, err
+	}
+
+	if packet, err := s.transport.readPacket(); err != nil {
+		return nil, err
+	} else if packet[0] != msgNewKeys {
+		return nil, unexpectedMessageError(msgNewKeys, packet[0])
+	}
+
+	var packet []byte
+	if packet, err = s.transport.readPacket(); err != nil {
+		return nil, err
+	}
+
+	var serviceRequest serviceRequestMsg
+	if err = Unmarshal(packet, &serviceRequest); err != nil {
+		return nil, err
+	}
+	if serviceRequest.Service != serviceUserAuth {
+		return nil, errors.New("ssh: requested service '" + serviceRequest.Service + "' before authenticating")
+	}
+	serviceAccept := serviceAcceptMsg{
+		Service: serviceUserAuth,
+	}
+	if err := s.transport.writePacket(Marshal(&serviceAccept)); err != nil {
+		return nil, err
+	}
+
+	return nil, nil
+}

ssh/sshpiper_test.go

@@ -0,0 +1,366 @@
+// Copyright 2014, 2015 tgic<farmer1992@gmail.com>. All rights reserved.
+// this file is governed by MIT-license
+//
+// https://github.com/tg123/sshpiper
+
+package ssh
+
+import (
+	"bytes"
+	"fmt"
+	"io/ioutil"
+	"net"
+	"testing"
+)
+
+// {{{ Example NewSSHPiperConn
+
+func ExampleNewSSHPiperConn() {
+
+	// upstream addr
+	const serverAddr = "127.0.0.1:22"
+
+	piper := &SSHPiperConfig{
+		// return conn dial to serverAddr
+		FindUpstream: func(conn ConnMetadata) (net.Conn, string, error) {
+			c, err := net.Dial("tcp", serverAddr)
+			if err != nil {
+				return nil, "", err
+			}
+
+			// change upstream username to root
+			return c, "root", nil
+		},
+	}
+
+	// add private key
+	privateBytes, err := ioutil.ReadFile("id_rsa")
+	if err != nil {
+		panic("Failed to load private key")
+	}
+
+	private, err := ParsePrivateKey(privateBytes)
+	if err != nil {
+		panic("Failed to parse private key")
+	}
+
+	piper.AddHostKey(private)
+
+	// serve at a address
+	listener, err := net.Listen("tcp", "0.0.0.0:2022")
+	if err != nil {
+		panic("failed to listen for connection")
+	}
+	nConn, err := listener.Accept()
+	if err != nil {
+		panic("failed to accept incoming connection")
+	}
+
+	// accept nConn and build a SSHPipe
+	p, err := NewSSHPiperConn(nConn, piper)
+	if err != nil {
+		panic("failed to establish piped connection")
+	}
+
+	// wait util either side shutdown
+	p.Wait()
+}
+
+// }}}
+
+func dialPiper(piper *SSHPiperConfig) (net.Conn, error) {
+	c, s, err := netPipe()
+	if err != nil {
+		return nil, err
+	}
+
+	piper.AddHostKey(testSigners["rsa"])
+
+	go func() {
+		defer c.Close()
+		defer s.Close()
+
+		p, err := NewSSHPiperConn(s, piper)
+
+		if err != nil {
+			return
+		}
+
+		p.Wait()
+	}()
+
+	return c, nil
+}
+
+func TestFindUpstreamCallback(t *testing.T) {
+
+	const username = "testuser"
+
+	var called bool
+
+	c, err := dialPiper(&SSHPiperConfig{
+		FindUpstream: func(conn ConnMetadata) (net.Conn, string, error) {
+			if username != conn.User() {
+				t.Errorf("different username")
+			}
+
+			s, err := dialUpstream(simpleEchoHandler, &ServerConfig{
+				PasswordCallback: func(conn ConnMetadata, password []byte) (*Permissions, error) {
+					called = true
+
+					if conn.User() != username {
+						t.Errorf("default username changed")
+					}
+
+					return nil, nil
+				},
+			}, t)
+
+			return s, "", err
+		},
+	})
+
+	if err != nil {
+		t.Fatalf("connect dial to piper: %v", err)
+	}
+
+	NewClientConn(c, "", &ClientConfig{
+		User: username,
+		Auth: []AuthMethod{Password("password")},
+	})
+
+	if !called {
+		t.Fatalf("FindUpstream not called")
+
+	}
+}
+
+// TODO clean up duplicate code
+func TestFindUpstreamWithUserCallback(t *testing.T) {
+	const username = "testuser"
+	const mappedname = "mappedname"
+
+	var called bool
+
+	c, err := dialPiper(&SSHPiperConfig{
+		FindUpstream: func(conn ConnMetadata) (net.Conn, string, error) {
+
+			s, err := dialUpstream(simpleEchoHandler, &ServerConfig{
+				PasswordCallback: func(conn ConnMetadata, password []byte) (*Permissions, error) {
+					called = true
+
+					if conn.User() != mappedname {
+						t.Errorf("bad mapped username")
+					}
+
+					return nil, nil
+				},
+			}, t)
+
+			return s, mappedname, err
+		},
+	})
+
+	if err != nil {
+		t.Fatalf("connect dial to piper: %v", err)
+	}
+
+	NewClientConn(c, "", &ClientConfig{
+		User: username,
+		Auth: []AuthMethod{Password("password")},
+	})
+
+	if !called {
+		t.Fatalf("FindUpstream not called")
+	}
+}
+
+func TestMapPublicKey(t *testing.T) {
+
+	certChecker := CertChecker{
+		IsAuthority: func(k PublicKey) bool {
+			return bytes.Equal(k.Marshal(), testPublicKeys["ecdsa"].Marshal())
+		},
+		UserKeyFallback: func(conn ConnMetadata, key PublicKey) (*Permissions, error) {
+			if bytes.Equal(key.Marshal(), testPublicKeys["rsa"].Marshal()) {
+				return nil, nil
+			}
+
+			return nil, fmt.Errorf("pubkey for %q not acceptable", conn.User())
+		},
+		IsRevoked: func(c *Certificate) bool {
+			return c.Serial == 666
+		},
+	}
+
+	c, err := dialPiper(&SSHPiperConfig{
+		FindUpstream: func(conn ConnMetadata) (net.Conn, string, error) {
+			s, err := dialUpstream(simpleEchoHandler, &ServerConfig{
+				PublicKeyCallback: certChecker.Authenticate,
+			}, t)
+			return s, "", err
+		},
+
+		MapPublicKey: func(conn ConnMetadata, key PublicKey) (Signer, error) {
+			return testSigners["rsa"], nil
+		},
+	})
+
+	if err != nil {
+		t.Fatalf("connect dial to piper: %v", err)
+	}
+
+	_, _, _, err = NewClientConn(c, "", &ClientConfig{
+		User: "testuser",
+		Auth: []AuthMethod{
+			PublicKeys(testSigners["rsa"]),
+		},
+	})
+
+	if err != nil {
+		t.Fatalf("can connect to piper %v", err)
+	}
+}
+
+func TestAdditionalChallenge(t *testing.T) {
+	c, err := dialPiper(&SSHPiperConfig{
+		AdditionalChallenge: func(conn ConnMetadata, challenge KeyboardInteractiveChallenge) (bool, error) {
+			ans, err := challenge("user",
+				"instruction",
+				[]string{"question1", "question2"},
+				[]bool{true, true})
+
+			if err != nil {
+				return false, err
+			}
+
+			ok := conn.User() == "testuser" && ans[0] == "answer1" && ans[1] == "answer2"
+			if ok {
+				challenge("user", "motd", nil, nil)
+				return true, nil
+			}
+			return false, fmt.Errorf("keyboard-interactive failed")
+		},
+		FindUpstream: func(conn ConnMetadata) (net.Conn, string, error) {
+			s, err := dialUpstream(simpleEchoHandler, &ServerConfig{NoClientAuth: true}, t)
+			return s, "", err
+		},
+	})
+
+	if err != nil {
+		t.Fatalf("connect dial to piper: %v", err)
+	}
+
+	answers := keyboardInteractive(map[string]string{
+		"question1": "answer1",
+		// TODO "question2": "WRONG",
+		"question2": "answer2",
+	})
+
+	_, _, _, err = NewClientConn(c, "", &ClientConfig{
+		User: "testuser",
+		Auth: []AuthMethod{
+			KeyboardInteractive(answers.Challenge),
+		},
+	})
+
+	if err != nil {
+		t.Fatalf("can connect to piper %v", err)
+	}
+
+}
+
+func fakeUpstreamServer(s net.Conn, upstream *ServerConfig, handler serverType, t *testing.T) {
+	defer s.Close()
+
+	upstream.AddHostKey(testSigners["rsa"])
+
+	_, chans, reqs, err := NewServerConn(s, upstream)
+	if err != nil {
+		t.Errorf("cannot start upstream %v", err)
+	}
+
+	go DiscardRequests(reqs)
+
+	for newCh := range chans {
+		if newCh.ChannelType() != "session" {
+			newCh.Reject(UnknownChannelType, "unknown channel type")
+			continue
+		}
+
+		ch, inReqs, err := newCh.Accept()
+		if err != nil {
+			t.Errorf("Accept: %v", err)
+			continue
+		}
+		go func() {
+			handler(ch, inReqs, t)
+		}()
+	}
+}
+
+func dialUpstream(handler serverType, upstream *ServerConfig, t *testing.T) (net.Conn, error) {
+	c, s, err := netPipe()
+	if err != nil {
+		t.Errorf("netPipe piper->upstream: %v", err)
+		return nil, err
+	}
+
+	go fakeUpstreamServer(s, upstream, handler, t)
+	return c, nil
+}
+
+func TestPipeData(t *testing.T) {
+
+	c, err := dialPiper(&SSHPiperConfig{
+		FindUpstream: func(conn ConnMetadata) (net.Conn, string, error) {
+			s, err := dialUpstream(simpleEchoHandler, &ServerConfig{NoClientAuth: true}, t)
+			return s, "", err
+		},
+	})
+
+	if err != nil {
+		t.Fatalf("connect dial to piper: %v", err)
+	}
+
+	// {{{ copy from session_test.go TestClientWriteEOF(t *testing.T)
+	sshc, chans, reqs, err := NewClientConn(c, "", &ClientConfig{User: "testuser"})
+	if err != nil {
+		t.Fatalf("error create client %v", err)
+	}
+
+	conn := NewClient(sshc, chans, reqs)
+	defer conn.Close()
+
+	session, err := conn.NewSession()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer session.Close()
+
+	stdin, err := session.StdinPipe()
+	if err != nil {
+		t.Fatalf("StdinPipe failed: %v", err)
+	}
+	stdout, err := session.StdoutPipe()
+	if err != nil {
+		t.Fatalf("StdoutPipe failed: %v", err)
+	}
+
+	data := []byte(`0000`)
+	_, err = stdin.Write(data)
+	if err != nil {
+		t.Fatalf("Write failed: %v", err)
+	}
+	stdin.Close()
+
+	res, err := ioutil.ReadAll(stdout)
+	if err != nil {
+		t.Fatalf("Read failed: %v", err)
+	}
+
+	if !bytes.Equal(data, res) {
+		t.Fatalf("Read differed from write, wrote: %v, read: %v", data, res)
+	}
+	// }}}
+}