teraWeb/backend/internal/portservice/gateway_test.go

package portservice

import (
	"bufio"
	"bytes"
	"context"
	"crypto/ed25519"
	"crypto/rand"
	"fmt"
	"io"
	"net"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"golang.org/x/crypto/ssh"
	"golang.org/x/crypto/ssh/knownhosts"

	"tera-clone/services/tera-api/internal/domain"
)

type pipeFactory struct{ peers chan net.Conn }

func (f *pipeFactory) Open(context.Context, string) (io.ReadWriteCloser, error) {
	left, right := net.Pipe()
	f.peers <- right
	return left, nil
}

type channelPipeFactory struct {
	mu    sync.Mutex
	peers map[string]chan net.Conn
}

func (f *channelPipeFactory) Open(_ context.Context, channel string) (io.ReadWriteCloser, error) {
	left, right := net.Pipe()
	f.mu.Lock()
	peers := f.peers[channel]
	if peers == nil {
		peers = make(chan net.Conn, 1)
		f.peers[channel] = peers
	}
	f.mu.Unlock()
	peers <- right
	return left, nil
}

func (f *channelPipeFactory) peer(t *testing.T, channel string) net.Conn {
	t.Helper()
	deadline := time.Now().Add(time.Second)
	for {
		f.mu.Lock()
		peers := f.peers[channel]
		f.mu.Unlock()
		if peers != nil {
			select {
			case peer := <-peers:
				return peer
			case <-time.After(time.Until(deadline)):
				t.Fatalf("timed out waiting for %s transport", channel)
			}
		}
		if time.Now().After(deadline) {
			t.Fatalf("timed out waiting for %s transport", channel)
		}
		time.Sleep(time.Millisecond)
	}
}

type eventRecorder struct {
	mu     sync.Mutex
	events []domain.PortLogEvent
}

func (r *eventRecorder) record(_ context.Context, event domain.PortLogEvent) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.events = append(r.events, event)
	return nil
}

func (r *eventRecorder) snapshot() []domain.PortLogEvent {
	r.mu.Lock()
	defer r.mu.Unlock()
	return append([]domain.PortLogEvent(nil), r.events...)
}

func TestReverseTelnetBridgesSerialWritesWelcomeAndRecordsSource(t *testing.T) {
	factory := &pipeFactory{peers: make(chan net.Conn, 1)}
	recorder := &eventRecorder{}
	listeners := make(chan net.Listener, 1)
	gateway, err := New(Options{
		TransportFactory: factory,
		Record:           recorder.record,
		Listen: func(_, _ string) (net.Listener, error) {
			listener, err := net.Listen("tcp", "127.0.0.1:0")
			if err == nil {
				listeners <- listener
			}
			return listener, err
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	defer gateway.Close()
	port := domain.Port{Channel: "CH12", Operation1: domain.PortOperation{
		Mode: "reverse_telnet", Port: 2012, MaxClients: 2, Authentication: "none", Newline: "crlf",
	}}
	if err := gateway.Apply(context.Background(), []domain.Port{port}); err != nil {
		t.Fatal(err)
	}
	listener := <-listeners
	connection, err := net.Dial("tcp", listener.Addr().String())
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	reader := bufio.NewReader(connection)
	peer := <-factory.peers
	defer peer.Close()
	_ = connection.SetReadDeadline(time.Now().Add(time.Second))
	welcome, err := reader.ReadString('\n')
	if err != nil {
		t.Fatal(err)
	}
	for !strings.Contains(welcome, "Connected at:") {
		line, readErr := reader.ReadString('\n')
		if readErr != nil {
			t.Fatal(readErr)
		}
		welcome += line
	}
	for _, expected := range []string{"Welcome to TERA CORA-X48 PRO K", "Serial channel: CH12", "Operation: 1", "Protocol: reverse_telnet", "Connected from: 127.0.0.1:"} {
		if !strings.Contains(welcome, expected) {
			t.Fatalf("welcome = %q, missing %q", welcome, expected)
		}
	}
	writeDone := make(chan error, 1)
	go func() { _, err := connection.Write([]byte("show version\n")); writeDone <- err }()
	serialInput := make([]byte, len("show version\r\n"))
	_ = peer.SetReadDeadline(time.Now().Add(time.Second))
	if _, err := io.ReadFull(peer, serialInput); err != nil {
		t.Fatal(err)
	}
	if err := <-writeDone; err != nil || string(serialInput) != "show version\r\n" {
		t.Fatalf("serial input = %q, write error = %v", serialInput, err)
	}
	responseDone := make(chan error, 1)
	go func() { _, err := peer.Write([]byte("Cisco IOS\r\n")); responseDone <- err }()
	line, err := reader.ReadString('\n')
	if err != nil || line != "\r\n" {
		// The welcome frame ends in a blank line; consume it before data.
		t.Fatalf("welcome terminator = %q, %v", line, err)
	}
	line, err = reader.ReadString('\n')
	if err != nil || line != "Cisco IOS\r\n" {
		t.Fatalf("serial output = %q, %v", line, err)
	}
	if err := <-responseDone; err != nil {
		t.Fatal(err)
	}
	events := recorder.snapshot()
	if len(events) == 0 || events[0].Channel != "CH12" || events[0].RemoteAddress != "127.0.0.1" || !strings.Contains(events[0].Message, "Serial channel: CH12") {
		t.Fatalf("session events = %#v", events)
	}
	if events[0].EventType != "session" || events[0].Direction != "" || events[0].ShowDirection {
		t.Fatalf("invalid session log schema = %#v", events[0])
	}
}

func TestReverseTelnetKeepsSessionOpenWhenLogStoreRejectsEvent(t *testing.T) {
	factory := &pipeFactory{peers: make(chan net.Conn, 1)}
	listeners := make(chan net.Listener, 1)
	gateway, err := New(Options{
		TransportFactory: factory,
		Record: func(context.Context, domain.PortLogEvent) error {
			return fmt.Errorf("log store unavailable")
		},
		Listen: func(_, _ string) (net.Listener, error) {
			listener, listenErr := net.Listen("tcp", "127.0.0.1:0")
			if listenErr == nil {
				listeners <- listener
			}
			return listener, listenErr
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	defer gateway.Close()
	port := domain.Port{Channel: "CH13", Operation1: domain.PortOperation{
		Mode: "reverse_telnet", Port: 2013, MaxClients: 1, Authentication: "none", Newline: "cr",
	}}
	if err := gateway.Apply(context.Background(), []domain.Port{port}); err != nil {
		t.Fatal(err)
	}
	connection, err := net.Dial("tcp", (<-listeners).Addr().String())
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	defer (<-factory.peers).Close()
	_ = connection.SetReadDeadline(time.Now().Add(time.Second))
	welcome := ""
	reader := bufio.NewReader(connection)
	for !strings.Contains(welcome, "Connected at:") {
		line, readErr := reader.ReadString('\n')
		if readErr != nil {
			t.Fatalf("session closed because logging failed: %v", readErr)
		}
		welcome += line
	}
}

func TestApplyKeepsCH01SessionWhenOnlyCH18ListenerChanges(t *testing.T) {
	factory := &channelPipeFactory{peers: make(map[string]chan net.Conn)}
	var listenerMu sync.Mutex
	listeners := make(map[string]net.Listener)
	listenCounts := make(map[string]int)
	gateway, err := New(Options{
		TransportFactory: factory,
		Listen: func(_, address string) (net.Listener, error) {
			listener, listenErr := net.Listen("tcp", "127.0.0.1:0")
			if listenErr != nil {
				return nil, listenErr
			}
			_, port, splitErr := net.SplitHostPort(address)
			if splitErr != nil {
				_ = listener.Close()
				return nil, splitErr
			}
			listenerMu.Lock()
			listeners[port] = listener
			listenCounts[port]++
			listenerMu.Unlock()
			return listener, nil
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	defer gateway.Close()
	ports := []domain.Port{
		{Channel: "CH01", Operation1: domain.PortOperation{Mode: "reverse_telnet", Port: 10001, MaxClients: 1, Authentication: "none", Newline: "cr"}},
		{Channel: "CH18", Operation1: domain.PortOperation{Mode: "reverse_telnet", Port: 2534, MaxClients: 1, Authentication: "none", Newline: "cr"}},
	}
	if err := gateway.Apply(context.Background(), ports); err != nil {
		t.Fatal(err)
	}
	gateway.mu.Lock()
	ch01Before := gateway.services["CH01-operation-1"]
	ch18Before := gateway.services["CH18-operation-1"]
	gateway.mu.Unlock()
	listenerMu.Lock()
	ch01Listener := listeners["10001"]
	listenerMu.Unlock()
	connection, err := net.Dial("tcp", ch01Listener.Addr().String())
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	peer := factory.peer(t, "CH01")
	defer peer.Close()
	reader := bufio.NewReader(connection)
	_ = connection.SetReadDeadline(time.Now().Add(time.Second))
	welcome := ""
	for !strings.Contains(welcome, "\r\n\r\n") {
		line, readErr := reader.ReadString('\n')
		if readErr != nil {
			t.Fatalf("read CH01 welcome: %v", readErr)
		}
		welcome += line
	}
	ports[1].Operation1.Port = 2535
	if err := gateway.Apply(context.Background(), ports); err != nil {
		t.Fatal(err)
	}
	gateway.mu.Lock()
	ch01After := gateway.services["CH01-operation-1"]
	ch18After := gateway.services["CH18-operation-1"]
	gateway.mu.Unlock()
	if ch01After != ch01Before {
		t.Fatal("unchanged CH01 service was restarted")
	}
	if ch18After == ch18Before {
		t.Fatal("changed CH18 service was not restarted")
	}
	listenerMu.Lock()
	ch01ListenCount := listenCounts["10001"]
	ch18OldListenCount := listenCounts["2534"]
	ch18NewListenCount := listenCounts["2535"]
	listenerMu.Unlock()
	if ch01ListenCount != 1 || ch18OldListenCount != 1 || ch18NewListenCount != 1 {
		t.Fatalf("listener starts: CH01=%d CH18-old=%d CH18-new=%d", ch01ListenCount, ch18OldListenCount, ch18NewListenCount)
	}
	writeDone := make(chan error, 1)
	go func() {
		_, writeErr := peer.Write([]byte("session-still-active\r\n"))
		writeDone <- writeErr
	}()
	line, err := reader.ReadString('\n')
	if err != nil || line != "session-still-active\r\n" {
		t.Fatalf("CH01 session after CH18 change = %q, %v", line, err)
	}
	if err := <-writeDone; err != nil {
		t.Fatal(err)
	}
}

func TestReverseTelnetKeepsFourSharedSessionsBidirectionalAndHandlesShortcuts(t *testing.T) {
	factory := &pipeFactory{peers: make(chan net.Conn, 1)}
	listeners := make(chan net.Listener, 1)
	gateway, err := New(Options{
		TransportFactory: factory,
		Listen: func(_, _ string) (net.Listener, error) {
			listener, err := net.Listen("tcp", "127.0.0.1:0")
			if err == nil {
				listeners <- listener
			}
			return listener, err
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	defer gateway.Close()
	port := domain.Port{Channel: "CH24", Operation1: domain.PortOperation{
		Mode: "reverse_telnet", Port: 2024, MaxClients: 4, Authentication: "none", Newline: "cr",
		ControlShortcut: "^]", SwitchShortcut: "^T", QuitShortcut: "^Q",
	}}
	if err := gateway.Apply(context.Background(), []domain.Port{port}); err != nil {
		t.Fatal(err)
	}
	listener := <-listeners
	connections := make([]net.Conn, 0, 4)
	readers := make([]*bufio.Reader, 0, 4)
	for index := 0; index < 4; index++ {
		connection, dialErr := net.Dial("tcp", listener.Addr().String())
		if dialErr != nil {
			t.Fatal(dialErr)
		}
		connections = append(connections, connection)
		reader := bufio.NewReader(connection)
		readers = append(readers, reader)
		_ = connection.SetReadDeadline(time.Now().Add(time.Second))
		welcome := ""
		for !strings.Contains(welcome, "\r\n\r\n") {
			chunk, readErr := reader.ReadString('\n')
			if readErr != nil {
				t.Fatalf("read welcome %d: %v", index+1, readErr)
			}
			welcome += chunk
		}
		if !strings.Contains(welcome, "Access: shared writer") || !strings.Contains(welcome, "control ^]") {
			t.Fatalf("welcome %d = %q", index+1, welcome)
		}
		_ = connection.SetReadDeadline(time.Time{})
	}
	defer func() {
		for _, connection := range connections {
			_ = connection.Close()
		}
	}()
	peer := <-factory.peers
	defer peer.Close()

	for index, connection := range connections {
		command := fmt.Sprintf("show-%d\n", index+1)
		go connection.Write([]byte(command))
		want := fmt.Sprintf("show-%d\r", index+1)
		payload := make([]byte, len(want))
		_ = peer.SetReadDeadline(time.Now().Add(time.Second))
		if _, err := io.ReadFull(peer, payload); err != nil || string(payload) != want {
			t.Fatalf("shared command %d = %q, %v", index+1, payload, err)
		}
	}

	go peer.Write([]byte("fanout\r\n"))
	for index, reader := range readers {
		connections[index].SetReadDeadline(time.Now().Add(time.Second))
		line, readErr := reader.ReadString('\n')
		if readErr != nil || line != "fanout\r\n" {
			t.Fatalf("fanout %d = %q, %v", index+1, line, readErr)
		}
		connections[index].SetReadDeadline(time.Time{})
	}

	if _, err := connections[0].Write([]byte{0x14}); err != nil { // Ctrl+T
		t.Fatal(err)
	}
	connections[0].SetReadDeadline(time.Now().Add(time.Second))
	status, err := readUntilContains(readers[0], "all connected terminal sessions remain online")
	if err != nil || !strings.Contains(status, "all connected terminal sessions remain online") {
		t.Fatalf("Ctrl+T status = %q, %v", status, err)
	}
	if _, err := connections[0].Write([]byte{0x11}); err != nil { // Ctrl+Q
		t.Fatal(err)
	}
	closed, err := readUntilContains(readers[0], "current session closed")
	if err != nil || !strings.Contains(closed, "current session closed") {
		t.Fatalf("Ctrl+Q response = %q, %v", closed, err)
	}
}

func readUntilContains(reader *bufio.Reader, expected string) (string, error) {
	var result string
	for !strings.Contains(result, expected) {
		line, err := reader.ReadString('\n')
		result += line
		if err != nil {
			return result, err
		}
	}
	return result, nil
}

func TestReverseSSHUsesLocalAuthenticationAndBridgesSession(t *testing.T) {
	_, privateKey, err := ed25519.GenerateKey(rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	signer, err := ssh.NewSignerFromKey(privateKey)
	if err != nil {
		t.Fatal(err)
	}
	factory := &pipeFactory{peers: make(chan net.Conn, 1)}
	recorder := &eventRecorder{}
	listeners := make(chan net.Listener, 1)
	gateway, err := New(Options{
		TransportFactory: factory, SSHSigner: signer, Record: recorder.record,
		Authenticate: func(_ context.Context, username, password, channel string) (string, error) {
			if username != "operator" || password != "secret" || channel != "CH01" {
				return "", io.EOF
			}
			return "operator", nil
		},
		Listen: func(_, _ string) (net.Listener, error) {
			listener, err := net.Listen("tcp", "127.0.0.1:0")
			if err == nil {
				listeners <- listener
			}
			return listener, err
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	defer gateway.Close()
	port := domain.Port{Channel: "CH01", Operation2: domain.PortOperation{
		Mode: "reverse_ssh", Port: 2201, MaxClients: 1, Authentication: "local", Newline: "cr",
	}}
	if err := gateway.Apply(context.Background(), []domain.Port{port}); err != nil {
		t.Fatal(err)
	}
	listener := <-listeners
	client, err := ssh.Dial("tcp", listener.Addr().String(), &ssh.ClientConfig{
		User: "operator", Auth: []ssh.AuthMethod{ssh.Password("secret")},
		HostKeyCallback: ssh.InsecureIgnoreHostKey(), Timeout: time.Second,
	})
	if err != nil {
		t.Fatal(err)
	}
	defer client.Close()
	session, err := client.NewSession()
	if err != nil {
		t.Fatal(err)
	}
	defer session.Close()
	input, _ := session.StdinPipe()
	output, _ := session.StdoutPipe()
	if err := session.Shell(); err != nil {
		t.Fatal(err)
	}
	peer := <-factory.peers
	defer peer.Close()
	reader := bufio.NewReader(output)
	welcome := ""
	for !strings.Contains(welcome, "Connected at:") {
		line, readErr := reader.ReadString('\n')
		if readErr != nil {
			t.Fatal(readErr)
		}
		welcome += line
	}
	if !strings.Contains(welcome, "Serial channel: CH01") || !strings.Contains(welcome, "User: operator") || !strings.Contains(welcome, "Protocol: reverse_ssh") {
		t.Fatalf("SSH welcome = %q", welcome)
	}
	go func() { _, _ = input.Write([]byte("show\n")) }()
	serialInput := make([]byte, len("show\r"))
	_ = peer.SetReadDeadline(time.Now().Add(time.Second))
	if _, err := io.ReadFull(peer, serialInput); err != nil || string(serialInput) != "show\r" {
		t.Fatalf("SSH serial input = %q, %v", serialInput, err)
	}
	events := recorder.snapshot()
	if len(events) == 0 || events[0].EventType != "session" || events[0].Direction != "" || events[0].ShowDirection {
		t.Fatalf("invalid SSH session log schema = %#v", events)
	}
}

func TestRecordFailureUsesQueryableCommandEvent(t *testing.T) {
	recorder := &eventRecorder{}
	gateway := &Gateway{options: Options{Record: recorder.record}}
	gateway.recordFailure(serviceConfig{
		channel: "CH18",
		key:     "CH18-operation-1",
		operation: domain.PortOperation{
			Mode: "reverse_ssh", Host: "0.0.0.0", Port: 2534,
		},
	}, fmt.Errorf("listener failed"))
	events := recorder.snapshot()
	if len(events) != 1 {
		t.Fatalf("failure events = %#v", events)
	}
	event := events[0]
	if event.EventType != "command" || event.Direction != "" || event.Command != "CH18-operation-1" || event.ReturnValue != "listener failed" {
		t.Fatalf("invalid failure log schema = %#v", event)
	}
}

func TestReverseTelnetUsesProtectedStaticPassword(t *testing.T) {
	factory := &pipeFactory{peers: make(chan net.Conn, 1)}
	listeners := make(chan net.Listener, 1)
	gateway, err := New(Options{
		TransportFactory: factory,
		Listen: func(_, _ string) (net.Listener, error) {
			listener, listenErr := net.Listen("tcp", "127.0.0.1:0")
			if listenErr == nil {
				listeners <- listener
			}
			return listener, listenErr
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	defer gateway.Close()
	port := domain.Port{Channel: "CH02", Operation1: domain.PortOperation{
		Mode: "reverse_telnet", Port: 2002, MaxClients: 1, Authentication: "password",
		Username: "operator", PasswordStored: true, Credential: "static-secret", Newline: "cr",
	}}
	if err := gateway.Apply(context.Background(), []domain.Port{port}); err != nil {
		t.Fatal(err)
	}
	listener := <-listeners
	connection, err := net.Dial("tcp", listener.Addr().String())
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	reader := bufio.NewReader(connection)
	_ = connection.SetDeadline(time.Now().Add(time.Second))
	if prompt, err := reader.ReadString(':'); err != nil || !strings.Contains(prompt, "Username:") {
		t.Fatalf("username prompt = %q, %v", prompt, err)
	}
	_, _ = connection.Write([]byte("operator\r\n"))
	if prompt, err := reader.ReadString(':'); err != nil || !strings.Contains(prompt, "Password:") {
		t.Fatalf("password prompt = %q, %v", prompt, err)
	}
	_, _ = connection.Write([]byte("static-secret\r\n"))
	welcome := ""
	for !strings.Contains(welcome, "Connected at:") {
		line, readErr := reader.ReadString('\n')
		if readErr != nil {
			t.Fatal(readErr)
		}
		welcome += line
	}
	if !strings.Contains(welcome, "User: operator") {
		t.Fatalf("welcome = %q", welcome)
	}
	peer := <-factory.peers
	_ = peer.Close()
}

func TestNormalTelnetAutoLoginAndBidirectionalBridge(t *testing.T) {
	factory := &pipeFactory{peers: make(chan net.Conn, 1)}
	remotePeers := make(chan net.Conn, 1)
	gateway, err := New(Options{
		TransportFactory: factory,
		RetryDelay:       time.Second,
		Dial: func(_ context.Context, network, address string) (net.Conn, error) {
			if network != "tcp" || address != "192.0.2.10:23" {
				t.Fatalf("Dial(%q, %q)", network, address)
			}
			left, right := net.Pipe()
			remotePeers <- right
			return left, nil
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	defer gateway.Close()
	port := domain.Port{Channel: "CH04", Operation1: domain.PortOperation{
		Mode: "telnet", Host: "192.0.2.10", Port: 23, MaxClients: 1,
		Authentication: "password", Username: "cisco", PasswordStored: true, Credential: "enable-secret", Newline: "crlf",
	}}
	if err := gateway.Apply(context.Background(), []domain.Port{port}); err != nil {
		t.Fatal(err)
	}
	remote := <-remotePeers
	defer remote.Close()
	_ = remote.SetReadDeadline(time.Now().Add(time.Second))
	autoLogin := make([]byte, len("cisco\r\nenable-secret\r\n"))
	if _, err := io.ReadFull(remote, autoLogin); err != nil || string(autoLogin) != "cisco\r\nenable-secret\r\n" {
		t.Fatalf("auto login = %q, %v", autoLogin, err)
	}
	serialPeer := <-factory.peers
	defer serialPeer.Close()
	go func() { _, _ = remote.Write([]byte("show version\r\n")) }()
	serialInput := make([]byte, len("show version\r\n"))
	_ = serialPeer.SetReadDeadline(time.Now().Add(time.Second))
	if _, err := io.ReadFull(serialPeer, serialInput); err != nil || string(serialInput) != "show version\r\n" {
		t.Fatalf("normal Telnet serial input = %q, %v", serialInput, err)
	}
	go func() { _, _ = serialPeer.Write([]byte("Cisco IOS\r\n")) }()
	response := make([]byte, len("Cisco IOS\r\n"))
	if _, err := io.ReadFull(remote, response); err != nil || string(response) != "Cisco IOS\r\n" {
		t.Fatalf("normal Telnet remote response = %q, %v", response, err)
	}
}

func TestNormalSSHVerifiesKnownHostAndBridgesSession(t *testing.T) {
	_, serverPrivateKey, err := ed25519.GenerateKey(rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	serverSigner, err := ssh.NewSignerFromKey(serverPrivateKey)
	if err != nil {
		t.Fatal(err)
	}
	knownHostsPath := filepath.Join(t.TempDir(), "known_hosts")
	line := knownhosts.Line([]string{"console.example"}, serverSigner.PublicKey()) + "\n"
	if err := os.WriteFile(knownHostsPath, []byte(line), 0o600); err != nil {
		t.Fatal(err)
	}
	factory := &pipeFactory{peers: make(chan net.Conn, 1)}
	serverChannels := make(chan ssh.Channel, 1)
	serverErrors := make(chan error, 1)
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	defer listener.Close()
	go func() {
		serverSide, acceptErr := listener.Accept()
		if acceptErr != nil {
			serverErrors <- acceptErr
			return
		}
		serverConfig := &ssh.ServerConfig{NoClientAuth: true}
		serverConfig.AddHostKey(serverSigner)
		serverConnection, channels, requests, serverErr := ssh.NewServerConn(serverSide, serverConfig)
		if serverErr != nil {
			serverErrors <- serverErr
			return
		}
		_ = serverConnection
		go ssh.DiscardRequests(requests)
		request := <-channels
		channel, channelRequests, acceptErr := request.Accept()
		if acceptErr != nil {
			serverErrors <- acceptErr
			return
		}
		go func() {
			for request := range channelRequests {
				_ = request.Reply(request.Type == "shell", nil)
			}
		}()
		serverChannels <- channel
	}()
	gateway, err := New(Options{
		TransportFactory: factory, KnownHostsPath: knownHostsPath, RetryDelay: time.Second,
		Dial: func(_ context.Context, _, address string) (net.Conn, error) {
			if address != "console.example:22" {
				t.Fatalf("SSH address = %q", address)
			}
			return net.Dial("tcp", listener.Addr().String())
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	defer gateway.Close()
	port := domain.Port{Channel: "CH05", Operation1: domain.PortOperation{
		Mode: "ssh", Host: "console.example", Port: 22, MaxClients: 1,
		Authentication: "none", Username: "operator", Newline: "cr",
	}}
	if err := gateway.Apply(context.Background(), []domain.Port{port}); err != nil {
		t.Fatal(err)
	}
	var remote ssh.Channel
	select {
	case remote = <-serverChannels:
	case serverErr := <-serverErrors:
		t.Fatal(serverErr)
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for normal SSH session")
	}
	defer remote.Close()
	serialPeer := <-factory.peers
	defer serialPeer.Close()
	go func() { _, _ = remote.Write([]byte("remote SSH data\r\n")) }()
	serialInput := make([]byte, len("remote SSH data\r"))
	_ = serialPeer.SetReadDeadline(time.Now().Add(time.Second))
	if _, err := io.ReadFull(serialPeer, serialInput); err != nil || string(serialInput) != "remote SSH data\r" {
		t.Fatalf("normal SSH serial input = %q, %v", serialInput, err)
	}
	go func() { _, _ = serialPeer.Write([]byte("serial response\r\n")) }()
	response := make([]byte, len("serial response\r\n"))
	if _, err := io.ReadFull(remote, response); err != nil || string(response) != "serial response\r\n" {
		t.Fatalf("normal SSH remote response = %q, %v", response, err)
	}
}

func TestNormalizeNewlines(t *testing.T) {
	input := []byte("one\r\ntwo\nthree\rfour")
	if value := string(normalizeNewlines(input, "lf")); value != "one\ntwo\nthree\nfour" {
		t.Fatalf("LF value = %q", value)
	}
	if value := string(normalizeNewlines(input, "crlf")); value != "one\r\ntwo\r\nthree\r\nfour" {
		t.Fatalf("CRLF value = %q", value)
	}
}

func TestRedirectForwardsSourceRXToEveryConfiguredTarget(t *testing.T) {
	factory := &channelPipeFactory{peers: make(map[string]chan net.Conn)}
	gateway, err := New(Options{TransportFactory: factory, RetryDelay: time.Second})
	if err != nil {
		t.Fatal(err)
	}
	defer gateway.Close()
	if err := gateway.Apply(context.Background(), []domain.Port{
		{Channel: "CH01", RedirectTo: []string{"CH02", "CH03"}},
		{Channel: "CH02"}, {Channel: "CH03"},
	}); err != nil {
		t.Fatal(err)
	}
	source := factory.peer(t, "CH01")
	target2 := factory.peer(t, "CH02")
	target3 := factory.peer(t, "CH03")
	defer source.Close()
	defer target2.Close()
	defer target3.Close()
	payload := []byte("redirect payload\r\n")
	go func() { _, _ = source.Write(payload) }()
	for channel, target := range map[string]net.Conn{"CH02": target2, "CH03": target3} {
		_ = target.SetReadDeadline(time.Now().Add(time.Second))
		actual := make([]byte, len(payload))
		if _, err := io.ReadFull(target, actual); err != nil || !bytes.Equal(actual, payload) {
			t.Fatalf("%s redirected payload = %q, %v", channel, actual, err)
		}
	}
	snapshot := gateway.Snapshot()
	if len(snapshot.Services) != 2 || snapshot.Services[0].Mode != "redirect" || snapshot.Services[1].Mode != "redirect" {
		t.Fatalf("snapshot = %#v", snapshot)
	}
}

func TestRedirectRejectsCycles(t *testing.T) {
	ports := []domain.Port{
		{Channel: "CH01", RedirectTo: []string{"CH02"}},
		{Channel: "CH02", RedirectTo: []string{"CH01"}},
	}
	if _, err := operationConfigs(ports); err == nil {
		t.Fatal("operationConfigs() error = nil")
	}
}

func TestApplyRestoresPreviousServicesWhenReplacementFails(t *testing.T) {
	factory := &pipeFactory{peers: make(chan net.Conn, 1)}
	failNext := false
	gateway, err := New(Options{
		TransportFactory: factory,
		Listen: func(_, _ string) (net.Listener, error) {
			if failNext {
				failNext = false
				return nil, io.ErrClosedPipe
			}
			return net.Listen("tcp", "127.0.0.1:0")
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	defer gateway.Close()
	oldPort := domain.Port{Channel: "CH01", Operation1: domain.PortOperation{Mode: "reverse_telnet", Port: 2001, MaxClients: 1, Authentication: "none"}}
	if err := gateway.Apply(context.Background(), []domain.Port{oldPort}); err != nil {
		t.Fatal(err)
	}
	failNext = true
	newPort := domain.Port{Channel: "CH01", Operation1: domain.PortOperation{Mode: "reverse_telnet", Port: 2002, MaxClients: 1, Authentication: "none"}}
	if err := gateway.Apply(context.Background(), []domain.Port{newPort}); err == nil {
		t.Fatal("Apply(replacement) error = nil")
	}
	snapshot := gateway.Snapshot()
	if len(snapshot.Services) != 1 || snapshot.Services[0].Port != 2001 {
		t.Fatalf("rollback snapshot = %#v", snapshot)
	}
}

type chunkReader struct {
	chunks [][]byte
}

func (r *chunkReader) Read(output []byte) (int, error) {
	if len(r.chunks) == 0 {
		return 0, io.EOF
	}
	chunk := r.chunks[0]
	r.chunks = r.chunks[1:]
	return copy(output, chunk), nil
}

func TestTelnetReaderStripsNegotiationAcrossReadBoundaries(t *testing.T) {
	reader := &telnetReader{reader: &chunkReader{chunks: [][]byte{
		{'A', 255}, {251}, {1, 'B', 255, 250, 24}, {1, 255}, {240, 'C', 255}, {255, 'D'},
	}}}
	actual, err := io.ReadAll(reader)
	if err != nil {
		t.Fatal(err)
	}
	want := []byte{'A', 'B', 'C', 255, 'D'}
	if !bytes.Equal(actual, want) {
		t.Fatalf("telnet payload = %v, want %v", actual, want)
	}
}