319 lines
8.5 KiB
Go
319 lines
8.5 KiB
Go
package outbound
|
|
|
|
import (
|
|
"context"
|
|
"crypto/ed25519"
|
|
"crypto/rand"
|
|
"errors"
|
|
"io"
|
|
"linkport/internal/model"
|
|
"net"
|
|
"strconv"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"golang.org/x/crypto/ssh"
|
|
)
|
|
|
|
type persistentSSHFixture struct {
|
|
proxy model.Proxy
|
|
handshakes atomic.Int32
|
|
authentications atomic.Int32
|
|
connections chan *ssh.ServerConn
|
|
closed chan struct{}
|
|
keepalives chan struct{}
|
|
dropKeepalive atomic.Bool
|
|
delayedChannel chan struct{}
|
|
releaseChannel chan struct{}
|
|
}
|
|
|
|
func newPersistentSSHFixture(t *testing.T) *persistentSSHFixture {
|
|
t.Helper()
|
|
_, key, err := ed25519.GenerateKey(rand.Reader)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
signer, err := ssh.NewSignerFromKey(key)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
f := &persistentSSHFixture{connections: make(chan *ssh.ServerConn, 32), closed: make(chan struct{}, 32), keepalives: make(chan struct{}, 8), delayedChannel: make(chan struct{}, 1), releaseChannel: make(chan struct{})}
|
|
config := &ssh.ServerConfig{PasswordCallback: func(meta ssh.ConnMetadata, password []byte) (*ssh.Permissions, error) {
|
|
f.authentications.Add(1)
|
|
if meta.User() != "fixture-user" || string(password) != "persistent-fixture-secret" {
|
|
return nil, errors.New("denied")
|
|
}
|
|
return nil, nil
|
|
}}
|
|
config.AddHostKey(signer)
|
|
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { listener.Close() })
|
|
host, port, _ := net.SplitHostPort(listener.Addr().String())
|
|
portNumber, _ := strconv.Atoi(port)
|
|
f.proxy = model.Proxy{TargetHost: "database.internal", TargetPort: 3306, Outbound: model.Outbound{Mode: "ssh", Host: host, Port: portNumber, Auth: "password", Username: "fixture-user", Password: "persistent-fixture-secret", HostFingerprint: ssh.FingerprintSHA256(signer.PublicKey())}}
|
|
go func() {
|
|
for {
|
|
raw, err := listener.Accept()
|
|
if err != nil {
|
|
return
|
|
}
|
|
go func() {
|
|
defer raw.Close()
|
|
conn, channels, requests, err := ssh.NewServerConn(raw, config)
|
|
if err != nil {
|
|
return
|
|
}
|
|
defer conn.Close()
|
|
defer func() { f.closed <- struct{}{} }()
|
|
f.handshakes.Add(1)
|
|
f.connections <- conn
|
|
go func() {
|
|
for request := range requests {
|
|
if request.Type == "keepalive@openssh.com" {
|
|
f.keepalives <- struct{}{}
|
|
if f.dropKeepalive.Load() {
|
|
continue
|
|
}
|
|
}
|
|
if request.WantReply {
|
|
request.Reply(false, nil)
|
|
}
|
|
}
|
|
}()
|
|
for newChannel := range channels {
|
|
if newChannel.ChannelType() != "direct-tcpip" {
|
|
newChannel.Reject(ssh.UnknownChannelType, "unsupported")
|
|
continue
|
|
}
|
|
var destination struct {
|
|
Host string
|
|
Port uint32
|
|
Origin string
|
|
OriginPort uint32
|
|
}
|
|
if ssh.Unmarshal(newChannel.ExtraData(), &destination) != nil {
|
|
newChannel.Reject(ssh.ConnectionFailed, "invalid")
|
|
continue
|
|
}
|
|
if destination.Host == "delayed.internal" {
|
|
f.delayedChannel <- struct{}{}
|
|
go func(ch ssh.NewChannel) {
|
|
<-f.releaseChannel
|
|
channel, requests, err := ch.Accept()
|
|
if err == nil {
|
|
go ssh.DiscardRequests(requests)
|
|
io.Copy(io.Discard, channel)
|
|
channel.Close()
|
|
}
|
|
}(newChannel)
|
|
continue
|
|
}
|
|
channel, requests, err := newChannel.Accept()
|
|
if err != nil {
|
|
continue
|
|
}
|
|
go ssh.DiscardRequests(requests)
|
|
go func(host string) {
|
|
defer channel.Close()
|
|
switch host {
|
|
case "half-close.internal":
|
|
payload, err := io.ReadAll(channel)
|
|
if err == nil {
|
|
channel.Write(append([]byte("reply:"), payload...))
|
|
}
|
|
case "eof.internal":
|
|
channel.Write([]byte("last-bytes"))
|
|
default:
|
|
io.Copy(channel, channel)
|
|
}
|
|
}(destination.Host)
|
|
}
|
|
}()
|
|
}
|
|
}()
|
|
return f
|
|
}
|
|
|
|
func TestSSHPersistentIdleKeepalive(t *testing.T) {
|
|
for _, drop := range []bool{false, true} {
|
|
t.Run(strconv.FormatBool(drop), func(t *testing.T) {
|
|
t.Parallel()
|
|
f := newPersistentSSHFixture(t)
|
|
f.dropKeepalive.Store(drop)
|
|
tunnel := f.open(t)
|
|
<-f.connections
|
|
select {
|
|
case <-f.keepalives:
|
|
case <-time.After(19 * time.Second):
|
|
t.Fatal("idle SSH connection did not send keepalive")
|
|
}
|
|
if drop {
|
|
select {
|
|
case <-tunnel.Done():
|
|
case <-time.After(8 * time.Second):
|
|
t.Fatal("unanswered keepalive did not stop tunnel")
|
|
}
|
|
} else {
|
|
c, err := tunnel.Dial(context.Background(), "tcp", "database.internal:3306")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer c.Close()
|
|
if err := persistentEcho(c, "idle-reuse"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if n := f.handshakes.Load(); n != 1 {
|
|
t.Fatalf("idle keepalive reconnected: %d", n)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestSSHPersistentCanceledChannelDoesNotStopTunnel(t *testing.T) {
|
|
f := newPersistentSSHFixture(t)
|
|
tunnel := f.open(t)
|
|
<-f.connections
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
defer close(f.releaseChannel)
|
|
result := make(chan error, 1)
|
|
go func() {
|
|
c, err := tunnel.Dial(ctx, "tcp", "delayed.internal:3306")
|
|
if c != nil {
|
|
c.Close()
|
|
}
|
|
result <- err
|
|
}()
|
|
select {
|
|
case <-f.delayedChannel:
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("delayed channel not opened")
|
|
}
|
|
cancel()
|
|
select {
|
|
case err := <-result:
|
|
if !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("canceled channel: %v", err)
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("canceled channel blocked")
|
|
}
|
|
c, err := tunnel.Dial(context.Background(), "tcp", "database.internal:3306")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer c.Close()
|
|
if err := persistentEcho(c, "after-channel-cancel"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func (f *persistentSSHFixture) open(t *testing.T) Tunnel {
|
|
t.Helper()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
tunnel, err := OpenPersistentTunnel(ctx, f.proxy)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { tunnel.Close() })
|
|
return tunnel
|
|
}
|
|
|
|
func TestSSHPersistentMultiplexLifecycle(t *testing.T) {
|
|
f := newPersistentSSHFixture(t)
|
|
tunnel := f.open(t)
|
|
// Handshake completion is observed by the fixture before counting reuse.
|
|
select {
|
|
case <-f.connections:
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("handshake not observed")
|
|
}
|
|
exercisePersistentTunnel(t, tunnel, f.handshakes.Load)
|
|
waitPersistentDone(t, f.closed)
|
|
}
|
|
|
|
func TestSSHPersistentHalfCloseEOFAndReopen(t *testing.T) {
|
|
f := newPersistentSSHFixture(t)
|
|
tunnel := f.open(t)
|
|
raw := <-f.connections
|
|
c, err := tunnel.Dial(context.Background(), "tcp", "half-close.internal:3306")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer c.Close()
|
|
c.SetDeadline(time.Now().Add(3 * time.Second))
|
|
c.Write([]byte("request"))
|
|
if err := c.(interface{ CloseWrite() error }).CloseWrite(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
data, err := io.ReadAll(c)
|
|
if err != nil || string(data) != "reply:request" {
|
|
t.Fatalf("half-close: %q %v", data, err)
|
|
}
|
|
c.Close()
|
|
eof, err := tunnel.Dial(context.Background(), "tcp", "eof.internal:3306")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer eof.Close()
|
|
eof.SetReadDeadline(time.Now().Add(3 * time.Second))
|
|
data, err = io.ReadAll(eof)
|
|
if err != nil || string(data) != "last-bytes" {
|
|
t.Fatalf("remote EOF: %q %v", data, err)
|
|
}
|
|
eof.Close()
|
|
raw.Close()
|
|
waitPersistentDone(t, tunnel.Done())
|
|
next := f.open(t)
|
|
<-f.connections
|
|
if n := f.handshakes.Load(); n != 2 {
|
|
t.Fatalf("reopen handshake count %d", n)
|
|
}
|
|
stream, err := next.Dial(context.Background(), "tcp", "database.internal:3306")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer stream.Close()
|
|
if err := persistentEcho(stream, "reopened"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func TestSSHPersistentNeverBypassesIdentityOrCredentials(t *testing.T) {
|
|
f := newPersistentSSHFixture(t)
|
|
for _, failure := range []string{"fingerprint", "password", "missing-fingerprint"} {
|
|
p := f.proxy
|
|
switch failure {
|
|
case "fingerprint":
|
|
p.Outbound.HostFingerprint = "SHA256:wrong"
|
|
case "password":
|
|
p.Outbound.Password = "wrong-private-password"
|
|
case "missing-fingerprint":
|
|
p.Outbound.HostFingerprint = ""
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
|
tunnel, err := OpenPersistentTunnel(ctx, p)
|
|
cancel()
|
|
if err == nil {
|
|
tunnel.Close()
|
|
t.Fatalf("accepted %s", failure)
|
|
}
|
|
if strings.Contains(err.Error(), p.Outbound.Password) || strings.Contains(err.Error(), p.Outbound.Username) {
|
|
t.Fatal("credential echoed in error")
|
|
}
|
|
}
|
|
if n := f.handshakes.Load(); n != 0 {
|
|
t.Fatalf("failed identities established %d tunnels", n)
|
|
}
|
|
if n := f.authentications.Load(); n != 1 {
|
|
t.Fatalf("fingerprint failure reached authentication: %d", n)
|
|
}
|
|
}
|