teraWeb/backend/internal/portservice/gateway.go

// Package portservice exposes per-channel SSH and Telnet operations without
// granting network-facing processes direct access to raw UART device paths.
package portservice

import (
	"bufio"
	"bytes"
	"context"
	"crypto/subtle"
	"errors"
	"fmt"
	"io"
	"net"
	"reflect"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

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

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

type AuthenticateFunc func(context.Context, string, string, string) (string, error)
type AuthenticateMethodFunc func(context.Context, string, string, string, string) (string, error)
type RecordFunc func(context.Context, domain.PortLogEvent) error
type ListenFunc func(string, string) (net.Listener, error)
type DialFunc func(context.Context, string, string) (net.Conn, error)

type Options struct {
	TransportFactory   directstream.TransportFactory
	Authenticate       AuthenticateFunc
	AuthenticateMethod AuthenticateMethodFunc
	Record             RecordFunc
	Listen             ListenFunc
	Dial               DialFunc
	ListenHost         string
	SSHSigner          ssh.Signer
	KnownHostsPath     string
	RetryDelay         time.Duration
}

type Controller interface {
	Apply(context.Context, []domain.Port) error
}

type SnapshotProvider interface {
	Snapshot() Snapshot
}

type NoopController struct{}

func (NoopController) Apply(_ context.Context, ports []domain.Port) error {
	_, err := operationConfigs(ports)
	return err
}

type Gateway struct {
	options Options
	broker  *directstream.Broker

	mu       sync.Mutex
	services map[string]*service
	sequence atomic.Uint64
}

type serviceConfig struct {
	key       string
	channel   string
	number    int
	operation domain.PortOperation
	sources   []string
	targets   []string
}

type service struct {
	config   serviceConfig
	listener net.Listener
	cancel   context.CancelFunc
	done     chan struct{}
	clients  atomic.Int32
	pending  atomic.Int32
}

type Snapshot struct {
	Services []ServiceSnapshot `json:"services"`
}

type ServiceSnapshot struct {
	Channel       string   `json:"channel"`
	Operation     int      `json:"operation"`
	Mode          string   `json:"mode"`
	Port          int      `json:"port"`
	Host          string   `json:"host,omitempty"`
	State         string   `json:"state"`
	ActiveClients int      `json:"active_clients"`
	Sources       []string `json:"sources,omitempty"`
	Targets       []string `json:"targets,omitempty"`
}

func New(options Options) (*Gateway, error) {
	if options.TransportFactory == nil {
		return nil, errors.New("port service transport factory is required")
	}
	if options.Listen == nil {
		options.Listen = net.Listen
	}
	if options.Dial == nil {
		dialer := &net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}
		options.Dial = dialer.DialContext
	}
	if strings.TrimSpace(options.ListenHost) == "" {
		options.ListenHost = "0.0.0.0"
	}
	if options.RetryDelay <= 0 {
		options.RetryDelay = 5 * time.Second
	}
	broker, err := directstream.NewBroker(options.TransportFactory, directstream.BrokerOptions{MaxAttachmentsPerChannel: 16})
	if err != nil {
		return nil, err
	}
	return &Gateway{options: options, broker: broker, services: make(map[string]*service)}, nil
}

// Apply reconciles the active operation set. Unchanged services are retained so
// configuration changes on one channel do not interrupt sessions on another.
// Callers should pass the running configuration only after the repository
// transaction has selected it.
func (g *Gateway) Apply(ctx context.Context, ports []domain.Port) error {
	desired, err := operationConfigs(ports)
	if err != nil {
		return err
	}
	g.mu.Lock()
	defer g.mu.Unlock()
	if err := ctx.Err(); err != nil {
		return err
	}
	desiredByKey := make(map[string]serviceConfig, len(desired))
	for _, config := range desired {
		desiredByKey[config.key] = config
	}
	oldConfigs := make([]serviceConfig, 0, len(g.services))
	for _, current := range g.services {
		oldConfigs = append(oldConfigs, current.config)
	}
	sort.Slice(oldConfigs, func(left, right int) bool { return oldConfigs[left].key < oldConfigs[right].key })
	retained := make(map[string]*service, len(g.services))
	stoppedConfigs := make([]serviceConfig, 0, len(oldConfigs))
	for _, previous := range oldConfigs {
		current := g.services[previous.key]
		next, exists := desiredByKey[previous.key]
		if exists && reflect.DeepEqual(previous, next) {
			retained[previous.key] = current
			continue
		}
		current.stop()
		stoppedConfigs = append(stoppedConfigs, previous)
	}
	toStart := make([]serviceConfig, 0, len(desired)-len(retained))
	for _, config := range desired {
		if _, exists := retained[config.key]; !exists {
			toStart = append(toStart, config)
		}
	}
	if len(stoppedConfigs) == 0 && len(toStart) == 0 {
		return nil
	}
	nextServices := make(map[string]*service, len(desired))
	for key, current := range retained {
		nextServices[key] = current
	}
	started := make([]*service, 0, len(toStart))
	for _, config := range toStart {
		current, startErr := g.startLocked(config)
		if startErr != nil {
			for _, item := range started {
				item.stop()
			}
			restoredServices := make(map[string]*service, len(oldConfigs))
			for key, item := range retained {
				restoredServices[key] = item
			}
			var rollbackErrors []error
			for _, previous := range stoppedConfigs {
				restored, restoreErr := g.startLocked(previous)
				if restoreErr != nil {
					rollbackErrors = append(rollbackErrors, fmt.Errorf("restore %s: %w", previous.key, restoreErr))
					continue
				}
				restoredServices[previous.key] = restored
			}
			g.services = restoredServices
			applyErr := fmt.Errorf("start %s: %w", config.key, startErr)
			if len(rollbackErrors) > 0 {
				return errors.Join(append([]error{applyErr}, rollbackErrors...)...)
			}
			return applyErr
		}
		nextServices[config.key] = current
		started = append(started, current)
	}
	g.services = nextServices
	return nil
}

func (g *Gateway) Snapshot() Snapshot {
	g.mu.Lock()
	defer g.mu.Unlock()
	result := Snapshot{Services: make([]ServiceSnapshot, 0, len(g.services))}
	for _, item := range g.services {
		result.Services = append(result.Services, ServiceSnapshot{
			Channel: item.config.channel, Operation: item.config.number, Mode: item.config.operation.Mode,
			Port: item.config.operation.Port, Host: item.config.operation.Host, State: serviceState(item.config),
			ActiveClients: int(item.clients.Load()),
			Sources:       append([]string(nil), item.config.sources...),
			Targets:       append([]string(nil), item.config.targets...),
		})
	}
	sort.Slice(result.Services, func(left, right int) bool {
		if result.Services[left].Channel != result.Services[right].Channel {
			return result.Services[left].Channel < result.Services[right].Channel
		}
		return result.Services[left].Operation < result.Services[right].Operation
	})
	return result
}

func (g *Gateway) Close() error {
	g.mu.Lock()
	for _, current := range g.services {
		current.stop()
	}
	g.services = make(map[string]*service)
	g.mu.Unlock()
	return g.broker.Close()
}

func (s *service) stop() {
	if s.cancel != nil {
		s.cancel()
	}
	if s.listener != nil {
		_ = s.listener.Close()
	}
	if s.done != nil {
		select {
		case <-s.done:
		case <-time.After(2 * time.Second):
		}
	}
}

func operationConfigs(ports []domain.Port) ([]serviceConfig, error) {
	result := make([]serviceConfig, 0, len(ports)*2)
	listeners := make(map[int]string)
	redirects := make(map[string][]string)
	for _, port := range ports {
		if directstream.ValidateChannel(port.Channel) != nil {
			return nil, fmt.Errorf("invalid operation channel %q", port.Channel)
		}
		for index, operation := range []domain.PortOperation{port.Operation1, port.Operation2} {
			if operation.Mode == "" || operation.Mode == "disabled" {
				continue
			}
			config := serviceConfig{key: fmt.Sprintf("%s-operation-%d", port.Channel, index+1), channel: port.Channel, number: index + 1, operation: operation}
			switch operation.Mode {
			case "reverse_telnet", "reverse_ssh":
				if operation.Port < 1 || operation.Port > 65535 {
					return nil, fmt.Errorf("%s listener port is invalid", config.key)
				}
				if previous := listeners[operation.Port]; previous != "" {
					return nil, fmt.Errorf("listener port %d conflicts between %s and %s", operation.Port, previous, config.key)
				}
				listeners[operation.Port] = config.key
			case "telnet", "ssh":
				if strings.TrimSpace(operation.Host) == "" || operation.Port < 1 || operation.Port > 65535 {
					return nil, fmt.Errorf("%s normal mode requires host and port", config.key)
				}
			default:
				return nil, fmt.Errorf("%s mode is unsupported", config.key)
			}
			result = append(result, config)
		}
		if len(port.RedirectTo) > 0 {
			targets := append([]string(nil), port.RedirectTo...)
			sort.Strings(targets)
			for _, target := range targets {
				if directstream.ValidateChannel(target) != nil || target == port.Channel {
					return nil, fmt.Errorf("%s redirect target %q is invalid", port.Channel, target)
				}
			}
			redirects[port.Channel] = targets
		}
	}
	if err := validateRedirectGraph(redirects); err != nil {
		return nil, err
	}
	sourcesByTarget := make(map[string][]string)
	for source, targets := range redirects {
		for _, target := range targets {
			sourcesByTarget[target] = append(sourcesByTarget[target], source)
		}
	}
	for target, sources := range sourcesByTarget {
		sort.Strings(sources)
		result = append(result, serviceConfig{
			key: "redirect-to-" + target, channel: target, operation: domain.PortOperation{Mode: "redirect"},
			sources: sources, targets: []string{target},
		})
	}
	sort.Slice(result, func(left, right int) bool { return result[left].key < result[right].key })
	return result, nil
}

func validateRedirectGraph(graph map[string][]string) error {
	state := make(map[string]uint8, len(graph))
	var visit func(string) error
	visit = func(channel string) error {
		switch state[channel] {
		case 1:
			return fmt.Errorf("redirect graph contains a cycle at %s", channel)
		case 2:
			return nil
		}
		state[channel] = 1
		for _, target := range graph[channel] {
			if _, redirects := graph[target]; redirects {
				if err := visit(target); err != nil {
					return err
				}
			}
		}
		state[channel] = 2
		return nil
	}
	for channel := range graph {
		if err := visit(channel); err != nil {
			return err
		}
	}
	return nil
}

func serviceState(config serviceConfig) string {
	switch config.operation.Mode {
	case "reverse_ssh", "reverse_telnet":
		return "listening"
	case "redirect":
		return "forwarding"
	default:
		return "reconnecting"
	}
}

func (g *Gateway) startLocked(config serviceConfig) (*service, error) {
	ctx, cancel := context.WithCancel(context.Background())
	current := &service{config: config, cancel: cancel, done: make(chan struct{})}
	switch config.operation.Mode {
	case "reverse_telnet", "reverse_ssh":
		if config.operation.Mode == "reverse_ssh" && g.options.SSHSigner == nil {
			cancel()
			return nil, errors.New("reverse SSH host key is unavailable")
		}
		if err := validateOperationCredential(config); err != nil {
			cancel()
			return nil, err
		}
		listener, err := g.options.Listen("tcp", net.JoinHostPort(g.options.ListenHost, strconv.Itoa(config.operation.Port)))
		if err != nil {
			cancel()
			return nil, err
		}
		current.listener = listener
		go g.runListener(ctx, current)
	case "telnet", "ssh":
		if config.operation.Mode == "ssh" && strings.TrimSpace(g.options.KnownHostsPath) == "" {
			cancel()
			return nil, errors.New("normal SSH requires a known_hosts path")
		}
		if err := validateOperationCredential(config); err != nil {
			cancel()
			return nil, err
		}
		go g.runNormal(ctx, current)
	case "redirect":
		go g.runRedirect(ctx, current)
	}
	return current, nil
}

func validateOperationCredential(config serviceConfig) error {
	operation := config.operation
	if operation.Authentication == "password" && operation.Credential == "" {
		return errors.New("password authentication credential is unavailable")
	}
	if operation.Authentication != "key" {
		return nil
	}
	if operation.Credential == "" {
		return errors.New("key authentication credential is unavailable")
	}
	if operation.Mode == "reverse_ssh" {
		if _, _, _, _, err := ssh.ParseAuthorizedKey([]byte(operation.Credential)); err != nil {
			return fmt.Errorf("parse authorized SSH key: %w", err)
		}
		return nil
	}
	if operation.Mode == "ssh" {
		if _, err := ssh.ParsePrivateKey([]byte(operation.Credential)); err != nil {
			return fmt.Errorf("parse SSH private key: %w", err)
		}
		return nil
	}
	return errors.New("key authentication is only valid for SSH")
}

func (g *Gateway) runListener(ctx context.Context, current *service) {
	defer close(current.done)
	for {
		connection, err := current.listener.Accept()
		if err != nil {
			if ctx.Err() != nil || errors.Is(err, net.ErrClosed) {
				return
			}
			continue
		}
		// Authentication handshakes are not active terminal sessions and must not
		// consume MaxClients. Keep a separate bounded pre-authentication ceiling
		// so incomplete handshakes still cannot exhaust the process.
		if current.pending.Add(1) > 16 {
			current.pending.Add(-1)
			_, _ = connection.Write([]byte("Too many pending connections.\r\n"))
			_ = connection.Close()
			continue
		}
		go func() {
			defer current.pending.Add(-1)
			defer connection.Close()
			if current.config.operation.Mode == "reverse_ssh" {
				g.serveSSH(ctx, connection, current)
			} else {
				g.serveTelnet(ctx, connection, current)
			}
		}()
	}
}

func (g *Gateway) serveTelnet(ctx context.Context, connection net.Conn, current *service) {
	config := current.config
	actor := "anonymous"
	reader := bufio.NewReader(&telnetReader{reader: connection})
	if isManagedAuthentication(config.operation.Authentication) {
		if g.options.Authenticate == nil && g.options.AuthenticateMethod == nil {
			return
		}
		_, _ = connection.Write([]byte("TERA CORA-X48 PRO K " + strings.ToUpper(config.operation.Authentication) + " authentication\r\nUsername: "))
		username, err := readCredentialLine(reader, 64)
		if err != nil {
			return
		}
		_, _ = connection.Write([]byte("Password: "))
		password, err := readCredentialLine(reader, 256)
		_, _ = connection.Write([]byte("\r\n"))
		if err != nil {
			return
		}
		actor, err = g.authenticate(ctx, config.operation.Authentication, username, password, config.channel)
		if err != nil {
			_, _ = connection.Write([]byte("Authentication failed.\r\n"))
			return
		}
	} else if config.operation.Authentication == "password" {
		_, _ = connection.Write([]byte("TERA CORA-X48 PRO K password authentication\r\nUsername: "))
		username, err := readCredentialLine(reader, 64)
		if err != nil {
			return
		}
		_, _ = connection.Write([]byte("Password: "))
		password, err := readCredentialLine(reader, 256)
		_, _ = connection.Write([]byte("\r\n"))
		if err != nil || !credentialMatches(config.operation.Credential, password) || (config.operation.Username != "" && config.operation.Username != username) {
			_, _ = connection.Write([]byte("Authentication failed.\r\n"))
			return
		}
		actor = username
	} else if config.operation.Authentication != "none" {
		return
	}
	if !beginTerminalSession(current) {
		_, _ = connection.Write([]byte("Too many clients.\r\n"))
		return
	}
	defer current.clients.Add(-1)
	g.bridge(ctx, connection, reader, config, actor, remoteEndpoint(connection.RemoteAddr()), true)
}

func (g *Gateway) serveSSH(ctx context.Context, connection net.Conn, current *service) {
	config := current.config
	serverConfig := &ssh.ServerConfig{ServerVersion: "SSH-2.0-TERA-CORA-X48-PRO-K"}
	serverConfig.AddHostKey(g.options.SSHSigner)
	if config.operation.Authentication == "none" {
		serverConfig.NoClientAuth = true
	} else if isManagedAuthentication(config.operation.Authentication) && (g.options.Authenticate != nil || g.options.AuthenticateMethod != nil) {
		serverConfig.PasswordCallback = func(metadata ssh.ConnMetadata, password []byte) (*ssh.Permissions, error) {
			actor, err := g.authenticate(ctx, config.operation.Authentication, metadata.User(), string(password), config.channel)
			if err != nil {
				return nil, err
			}
			return &ssh.Permissions{Extensions: map[string]string{"actor": actor}}, nil
		}
	} else if config.operation.Authentication == "password" {
		serverConfig.PasswordCallback = func(metadata ssh.ConnMetadata, password []byte) (*ssh.Permissions, error) {
			if (config.operation.Username != "" && config.operation.Username != metadata.User()) || !credentialMatches(config.operation.Credential, string(password)) {
				return nil, errors.New("authentication failed")
			}
			return &ssh.Permissions{Extensions: map[string]string{"actor": metadata.User()}}, nil
		}
	} else if config.operation.Authentication == "key" {
		authorized, _, _, _, err := ssh.ParseAuthorizedKey([]byte(config.operation.Credential))
		if err != nil {
			return
		}
		serverConfig.PublicKeyCallback = func(metadata ssh.ConnMetadata, key ssh.PublicKey) (*ssh.Permissions, error) {
			if (config.operation.Username != "" && config.operation.Username != metadata.User()) || !bytes.Equal(authorized.Marshal(), key.Marshal()) {
				return nil, errors.New("authentication failed")
			}
			return &ssh.Permissions{Extensions: map[string]string{"actor": metadata.User()}}, nil
		}
	} else {
		return
	}
	serverConnection, channels, requests, err := ssh.NewServerConn(connection, serverConfig)
	if err != nil {
		return
	}
	defer serverConnection.Close()
	go ssh.DiscardRequests(requests)
	for channelRequest := range channels {
		if channelRequest.ChannelType() != "session" {
			_ = channelRequest.Reject(ssh.UnknownChannelType, "session channel required")
			continue
		}
		channel, channelRequests, err := channelRequest.Accept()
		if err != nil {
			continue
		}
		go acceptShellRequests(channelRequests)
		if !beginTerminalSession(current) {
			_, _ = channel.Write([]byte("Too many clients.\r\n"))
			_ = channel.Close()
			return
		}
		actor := serverConnection.User()
		if serverConnection.Permissions != nil && serverConnection.Permissions.Extensions["actor"] != "" {
			actor = serverConnection.Permissions.Extensions["actor"]
		}
		g.bridge(ctx, channel, channel, config, actor, remoteEndpoint(serverConnection.RemoteAddr()), true)
		current.clients.Add(-1)
		_ = channel.Close()
		return
	}
}

func beginTerminalSession(current *service) bool {
	maximum := current.config.operation.MaxClients
	if maximum < 1 {
		maximum = 1
	}
	if current.clients.Add(1) <= int32(maximum) {
		return true
	}
	current.clients.Add(-1)
	return false
}

func acceptShellRequests(requests <-chan *ssh.Request) {
	for request := range requests {
		switch request.Type {
		case "pty-req", "env", "window-change", "shell":
			if request.WantReply {
				_ = request.Reply(true, nil)
			}
		default:
			if request.WantReply {
				_ = request.Reply(false, nil)
			}
		}
	}
}

func (g *Gateway) runNormal(ctx context.Context, current *service) {
	defer close(current.done)
	for {
		if ctx.Err() != nil {
			return
		}
		if current.config.operation.Mode == "ssh" {
			g.connectNormalSSH(ctx, current.config)
		} else {
			g.connectNormalTelnet(ctx, current.config)
		}
		timer := time.NewTimer(g.options.RetryDelay)
		select {
		case <-ctx.Done():
			timer.Stop()
			return
		case <-timer.C:
		}
	}
}

func (g *Gateway) connectNormalTelnet(ctx context.Context, config serviceConfig) {
	connection, err := g.options.Dial(ctx, "tcp", net.JoinHostPort(config.operation.Host, strconv.Itoa(config.operation.Port)))
	if err != nil {
		g.recordFailure(config, err)
		return
	}
	defer connection.Close()
	if config.operation.Username != "" {
		_, _ = connection.Write(append([]byte(config.operation.Username), operationNewline(config.operation.Newline)...))
	}
	if config.operation.Credential != "" {
		_, _ = connection.Write(append([]byte(config.operation.Credential), operationNewline(config.operation.Newline)...))
	}
	g.bridge(ctx, connection, &telnetReader{reader: connection}, config, config.operation.Username, remoteEndpoint(connection.RemoteAddr()), false)
}

func (g *Gateway) connectNormalSSH(ctx context.Context, config serviceConfig) {
	hostKeyCallback, err := knownhosts.New(g.options.KnownHostsPath)
	if err != nil {
		g.recordFailure(config, err)
		return
	}
	clientConfig := &ssh.ClientConfig{User: config.operation.Username, HostKeyCallback: hostKeyCallback, Timeout: operationTimeout(config.operation)}
	switch config.operation.Authentication {
	case "password", "local":
		if config.operation.Credential != "" {
			clientConfig.Auth = []ssh.AuthMethod{ssh.Password(config.operation.Credential)}
		}
	case "key":
		signer, parseErr := ssh.ParsePrivateKey([]byte(config.operation.Credential))
		if parseErr != nil {
			g.recordFailure(config, parseErr)
			return
		}
		clientConfig.Auth = []ssh.AuthMethod{ssh.PublicKeys(signer)}
	}
	address := net.JoinHostPort(config.operation.Host, strconv.Itoa(config.operation.Port))
	connection, err := g.options.Dial(ctx, "tcp", address)
	if err != nil {
		g.recordFailure(config, err)
		return
	}
	defer connection.Close()
	clientConnection, channels, requests, err := ssh.NewClientConn(connection, address, clientConfig)
	if err != nil {
		g.recordFailure(config, err)
		return
	}
	client := ssh.NewClient(clientConnection, channels, requests)
	defer client.Close()
	channel, channelRequests, err := client.OpenChannel("session", nil)
	if err != nil {
		g.recordFailure(config, err)
		return
	}
	defer channel.Close()
	go ssh.DiscardRequests(channelRequests)
	_, err = channel.SendRequest("shell", true, nil)
	if err != nil {
		g.recordFailure(config, err)
		return
	}
	g.bridge(ctx, channel, channel, config, config.operation.Username, remoteEndpoint(connection.RemoteAddr()), false)
}

func (g *Gateway) runRedirect(ctx context.Context, current *service) {
	defer close(current.done)
	for {
		if ctx.Err() != nil {
			return
		}
		if err := g.forwardRedirect(ctx, current.config); err != nil && ctx.Err() == nil {
			g.recordFailure(current.config, err)
		}
		timer := time.NewTimer(g.options.RetryDelay)
		select {
		case <-ctx.Done():
			timer.Stop()
			return
		case <-timer.C:
		}
	}
}

func (g *Gateway) forwardRedirect(ctx context.Context, config serviceConfig) error {
	sequence := g.sequence.Add(1)
	targetChannel := config.channel
	targetSession := fmt.Sprintf("redirect-to-%s-%d", strings.ToLower(targetChannel), sequence)
	target, err := g.broker.Attach(ctx, directstream.AttachRequest{
		SessionID: targetSession, Channel: targetChannel, Actor: "tera-redirect", Access: directstream.AccessWriter,
	})
	if err != nil {
		return fmt.Errorf("attach target %s: %w", targetChannel, err)
	}
	defer target.Close()
	type sourceAttachment struct {
		channel    string
		attachment *directstream.Attachment
	}
	sources := make([]sourceAttachment, 0, len(config.sources))
	defer func() {
		for _, source := range sources {
			_ = source.attachment.Close()
		}
	}()
	for _, channel := range config.sources {
		source, attachErr := g.broker.Attach(ctx, directstream.AttachRequest{
			SessionID: fmt.Sprintf("redirect-%s-to-%s-%d", strings.ToLower(channel), strings.ToLower(targetChannel), sequence),
			Channel:   channel, Actor: "tera-redirect", Access: directstream.AccessViewer,
		})
		if attachErr != nil {
			return fmt.Errorf("attach source %s: %w", channel, attachErr)
		}
		sources = append(sources, sourceAttachment{channel: channel, attachment: source})
	}
	connectedAt := time.Now().UTC()
	message := fmt.Sprintf("TERA CORA-X48 PRO K serial redirect connected\nSource channels: %s\nTarget channel: %s\nSession: %s\nConnected at: %s", strings.Join(config.sources, ","), targetChannel, targetSession, connectedAt.Format(time.RFC3339))
	_ = g.recordSession(config, "tera-redirect", endpoint{host: "local"}, targetSession, "connected", message, connectedAt)
	errorsChannel := make(chan error, len(sources))
	for _, item := range sources {
		go func(source sourceAttachment) {
			for {
				payload, receiveErr := source.attachment.Receive(ctx)
				if receiveErr != nil {
					errorsChannel <- receiveErr
					return
				}
				if writeErr := target.Write(ctx, payload); writeErr != nil {
					errorsChannel <- fmt.Errorf("write %s to %s: %w", source.channel, targetChannel, writeErr)
					return
				}
			}
		}(item)
	}
	select {
	case <-ctx.Done():
		return ctx.Err()
	case forwardErr := <-errorsChannel:
		return forwardErr
	}
}

type readWriteCloser interface {
	io.Reader
	io.Writer
	io.Closer
}

func (g *Gateway) bridge(ctx context.Context, connection readWriteCloser, reader io.Reader, config serviceConfig, actor string, remote endpoint, welcome bool) {
	if strings.TrimSpace(actor) == "" {
		actor = "anonymous"
	}
	sessionID := fmt.Sprintf("operation-%s-%d-%d", strings.ToLower(config.channel), config.number, g.sequence.Add(1))
	attachment, err := g.broker.Attach(ctx, directstream.AttachRequest{
		SessionID: sessionID, Channel: config.channel, Actor: actor,
		Access: directstream.AccessWriter, SharedWrite: true,
	})
	if err != nil {
		g.recordFailure(config, err)
		return
	}
	defer attachment.Close()
	connectedAt := time.Now().UTC()
	message := welcomeMessage(config, actor, remote, sessionID, connectedAt)
	_ = g.recordSession(config, actor, remote, sessionID, "connected", message, connectedAt)
	if welcome {
		_, _ = connection.Write([]byte("\r\n" + strings.ReplaceAll(message, "\n", "\r\n") + "\r\n\r\n"))
	}
	streamContext, cancel := context.WithCancel(ctx)
	defer cancel()
	shortcuts := terminalShortcutSet(config.operation)
	var connectionWriteMu sync.Mutex
	writeConnection := func(payload []byte) error {
		connectionWriteMu.Lock()
		defer connectionWriteMu.Unlock()
		_, err := connection.Write(payload)
		return err
	}
	writeNotice := func(message string) error {
		return writeConnection([]byte("\r\n[TERA] " + message + "\r\n"))
	}
	done := make(chan struct{}, 2)
	go func() {
		defer func() { done <- struct{}{} }()
		buffer := make([]byte, 32*1024)
		controlMode := false
		for {
			count, readErr := reader.Read(buffer)
			if count > 0 {
				payload := make([]byte, 0, count)
				for _, value := range buffer[:count] {
					if controlMode {
						switch value {
						case '?':
							if writeNotice(controlModeHelp(shortcuts)) != nil {
								return
							}
							continue
						case 'q', 'Q':
							_ = writeNotice("current session closed")
							return
						case 't', 'T':
							controlMode = false
							if writeNotice(config.channel+" terminal resumed; all other Telnet sessions remain connected") != nil {
								return
							}
							continue
						case 'b', 'B':
							controlMode = false
							if err := attachment.SendBreak(streamContext); err != nil {
								if writeNotice("BREAK failed: "+err.Error()) != nil {
									return
								}
							} else if writeNotice("BREAK sent to "+config.channel) != nil {
								return
							}
							continue
						case 'i', 'I':
							_ = writeNotice("session interrupted")
							return
						default:
							controlMode = false
							if writeNotice("unknown control command; terminal resumed") != nil {
								return
							}
							continue
						}
					}
					switch {
					case shortcuts.control.matches(value):
						controlMode = true
						if writeNotice(controlModeHelp(shortcuts)) != nil {
							return
						}
					case shortcuts.switchSession.matches(value):
						if writeNotice(config.channel+" terminal active; all connected terminal sessions remain online") != nil {
							return
						}
					case shortcuts.quit.matches(value):
						_ = writeNotice("current session closed")
						return
					case shortcuts.sendBreak.matches(value):
						if err := attachment.SendBreak(streamContext); err != nil {
							if writeNotice("BREAK failed: "+err.Error()) != nil {
								return
							}
						} else if writeNotice("BREAK sent to "+config.channel) != nil {
							return
						}
					case shortcuts.interrupt.matches(value):
						_ = writeNotice("session interrupted")
						return
					default:
						payload = append(payload, value)
					}
				}
				if len(payload) > 0 {
					payload = normalizeNewlines(payload, config.operation.Newline)
					if writeErr := attachment.Write(streamContext, payload); writeErr != nil {
						return
					}
				}
			}
			if readErr != nil {
				return
			}
		}
	}()
	go func() {
		defer func() { done <- struct{}{} }()
		for {
			payload, receiveErr := attachment.Receive(streamContext)
			if receiveErr != nil {
				return
			}
			if writeErr := writeConnection(payload); writeErr != nil {
				return
			}
		}
	}()
	<-done
	cancel()
	_ = connection.Close()
	<-done
	disconnectedAt := time.Now().UTC()
	disconnect := fmt.Sprintf("TERA CORA-X48 PRO K session disconnected\nConnected from: %s\nSerial channel: %s\nProtocol: %s\nUser: %s\nSession: %s\nDisconnected at: %s", remote.host, config.channel, config.operation.Mode, actor, sessionID, disconnectedAt.Format(time.RFC3339))
	_ = g.recordSession(config, actor, remote, sessionID, "disconnected", disconnect, disconnectedAt)
}

func (g *Gateway) recordSession(config serviceConfig, actor string, remote endpoint, sessionID, eventType, message string, timestamp time.Time) error {
	if g.options.Record == nil {
		return nil
	}
	return g.options.Record(context.Background(), domain.PortLogEvent{
		Timestamp: timestamp, Channel: config.channel, EventType: "session",
		Actor: actor + "/" + sessionID, Message: eventType + ": " + message, RemoteAddress: remote.host, RemotePort: remote.port,
		ShowTimestamp: true, ShowUsername: true, ShowIPAddress: true, ShowPort: true,
	})
}

func (g *Gateway) recordFailure(config serviceConfig, err error) {
	if g.options.Record == nil || err == nil {
		return
	}
	_, port := remoteEndpointString(config.operation.Host, config.operation.Port)
	_ = g.options.Record(context.Background(), domain.PortLogEvent{
		Timestamp: time.Now().UTC(), Channel: config.channel, EventType: "command",
		Actor: "tera-api", Command: config.key, ReturnValue: err.Error(), Message: config.key + ": " + err.Error(), RemoteAddress: config.operation.Host, RemotePort: port,
		ShowTimestamp: true, ShowIPAddress: true, ShowPort: true,
	})
}

func welcomeMessage(config serviceConfig, actor string, remote endpoint, sessionID string, timestamp time.Time) string {
	shortcuts := terminalShortcutSet(config.operation)
	return fmt.Sprintf("Welcome to TERA CORA-X48 PRO K\nConnected from: %s:%d\nSerial channel: %s\nOperation: %d\nProtocol: %s\nUser: %s\nAccess: shared writer\nSession: %s\nShortcuts: control %s, status/switch %s, quit %s, break %s, interrupt %s\nConnected at: %s", remote.host, remote.port, config.channel, config.number, config.operation.Mode, actor, sessionID, shortcuts.control.label(), shortcuts.switchSession.label(), shortcuts.quit.label(), shortcuts.sendBreak.label(), shortcuts.interrupt.label(), timestamp.Format(time.RFC3339))
}

type terminalShortcut struct {
	notation string
	value    byte
	enabled  bool
}

func (s terminalShortcut) matches(value byte) bool { return s.enabled && s.value == value }
func (s terminalShortcut) label() string {
	if !s.enabled {
		return "disabled"
	}
	return s.notation
}

type terminalShortcuts struct {
	control       terminalShortcut
	switchSession terminalShortcut
	quit          terminalShortcut
	sendBreak     terminalShortcut
	interrupt     terminalShortcut
}

func terminalShortcutSet(operation domain.PortOperation) terminalShortcuts {
	return terminalShortcuts{
		control:       makeTerminalShortcut(operation.ControlShortcut, "^]"),
		switchSession: makeTerminalShortcut(operation.SwitchShortcut, "^T"),
		quit:          makeTerminalShortcut(operation.QuitShortcut, "^Q"),
		sendBreak:     makeTerminalShortcut(operation.BreakShortcut, ""),
		interrupt:     makeTerminalShortcut(operation.InterruptShortcut, ""),
	}
}

func makeTerminalShortcut(notation, fallback string) terminalShortcut {
	notation = strings.ToUpper(strings.TrimSpace(notation))
	if notation == "" {
		notation = fallback
	}
	if len(notation) != 2 || notation[0] != '^' || notation[1] < '@' || notation[1] > '_' {
		return terminalShortcut{}
	}
	return terminalShortcut{notation: notation, value: notation[1] & 0x1f, enabled: true}
}

func controlModeHelp(shortcuts terminalShortcuts) string {
	return fmt.Sprintf("control mode: T resume/status, Q close, B send BREAK, I interrupt, ? help (direct: switch %s, quit %s, break %s, interrupt %s)", shortcuts.switchSession.label(), shortcuts.quit.label(), shortcuts.sendBreak.label(), shortcuts.interrupt.label())
}

type endpoint struct {
	host string
	port int
}

func remoteEndpoint(address net.Addr) endpoint {
	if address == nil {
		return endpoint{host: "unknown"}
	}
	host, portText, err := net.SplitHostPort(address.String())
	if err != nil {
		return endpoint{host: address.String()}
	}
	port, _ := strconv.Atoi(portText)
	return endpoint{host: strings.Trim(host, "[]"), port: port}
}

func remoteEndpointString(host string, port int) (string, int) { return host, port }

func operationTimeout(operation domain.PortOperation) time.Duration {
	return 10 * time.Second
}

func isManagedAuthentication(method string) bool {
	switch strings.ToLower(strings.TrimSpace(method)) {
	case "local", "radius", "tacacs":
		return true
	default:
		return false
	}
}

func (g *Gateway) authenticate(ctx context.Context, method, username, credential, channel string) (string, error) {
	if g.options.AuthenticateMethod != nil {
		return g.options.AuthenticateMethod(ctx, method, username, credential, channel)
	}
	if g.options.Authenticate != nil {
		return g.options.Authenticate(ctx, username, credential, channel)
	}
	return "", errors.New("authentication service is unavailable")
}

func credentialMatches(expected, actual string) bool {
	if len(expected) != len(actual) || len(expected) == 0 {
		return false
	}
	return subtle.ConstantTimeCompare([]byte(expected), []byte(actual)) == 1
}

func operationNewline(mode string) []byte {
	switch mode {
	case "lf":
		return []byte{'\n'}
	case "crlf":
		return []byte{'\r', '\n'}
	default:
		return []byte{'\r'}
	}
}

func readCredentialLine(reader *bufio.Reader, limit int) (string, error) {
	line, err := reader.ReadString('\n')
	if err != nil {
		return "", err
	}
	line = strings.TrimRight(line, "\r\n")
	if len(line) == 0 || len(line) > limit {
		return "", errors.New("credential line is invalid")
	}
	return line, nil
}

func normalizeNewlines(payload []byte, mode string) []byte {
	if mode == "" || mode == "none" {
		return append([]byte(nil), payload...)
	}
	result := make([]byte, 0, len(payload)*2)
	for index := 0; index < len(payload); index++ {
		value := payload[index]
		if value == '\r' && index+1 < len(payload) && payload[index+1] == '\n' {
			index++
			value = '\n'
		}
		if value != '\r' && value != '\n' {
			result = append(result, value)
			continue
		}
		switch mode {
		case "cr":
			result = append(result, '\r')
		case "lf":
			result = append(result, '\n')
		case "crlf":
			result = append(result, '\r', '\n')
		}
	}
	return result
}

// telnetReader removes option negotiation bytes before they can reach a UART.
// The parser state is retained because IAC commands and subnegotiation frames
// are routinely split across TCP reads.
type telnetReader struct {
	reader io.Reader
	state  uint8
}

const (
	telnetData uint8 = iota
	telnetIAC
	telnetOption
	telnetSubnegotiation
	telnetSubnegotiationIAC
)

func (r *telnetReader) Read(output []byte) (int, error) {
	if len(output) == 0 {
		return 0, nil
	}
	temporary := make([]byte, len(output))
	for {
		count, err := r.reader.Read(temporary)
		written := 0
		for _, value := range temporary[:count] {
			switch r.state {
			case telnetData:
				if value == 255 {
					r.state = telnetIAC
				} else if written < len(output) {
					output[written] = value
					written++
				}
			case telnetIAC:
				switch value {
				case 255:
					if written < len(output) {
						output[written] = value
						written++
					}
					r.state = telnetData
				case 251, 252, 253, 254:
					r.state = telnetOption
				case 250:
					r.state = telnetSubnegotiation
				default:
					r.state = telnetData
				}
			case telnetOption:
				r.state = telnetData
			case telnetSubnegotiation:
				if value == 255 {
					r.state = telnetSubnegotiationIAC
				}
			case telnetSubnegotiationIAC:
				if value == 240 {
					r.state = telnetData
				} else if value != 255 {
					r.state = telnetSubnegotiation
				}
			}
		}
		if written > 0 {
			return written, nil
		}
		if err != nil {
			return 0, err
		}
	}
}