feat: add console proxy session relay

server/internal/proxy/registry.go

@@ -29,6 +29,7 @@ type Entry struct {
 	ProxyID    string
 	InstanceID string
 	ServerID   string
+	Session    *Session
 }
 
 type Registry struct {

server/internal/proxy/session.go

@@ -0,0 +1,180 @@
+package proxy
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"sync"
+	"time"
+
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/grpc/pb"
+)
+
+const (
+	chunkSize         = 32 * 1024
+	rendezvousTimeout = 10 * time.Second
+)
+
+// AgentStream is the server's half of a ProxyStream. It is an interface so the
+// relay can be tested without gRPC.
+type AgentStream interface {
+	Send(*pb.ProxyServerMsg) error
+	Recv() (*pb.ProxyClientMsg, error)
+}
+
+// Session owns one ephemeral listener and relays the single connection that
+// arrives on it to an agent stream.
+type Session struct {
+	listener net.Listener
+	allowed  []string
+	timeout  time.Duration
+
+	once   sync.Once
+	mu     sync.Mutex
+	reason string
+}
+
+// NewSession binds an ephemeral port on listenHost. allowed is the set of IPs
+// permitted to connect; an empty set allows any, which is the degraded case
+// when guacd's host could not be resolved.
+func NewSession(listenHost string, allowed []string) (*Session, error) {
+	ln, err := net.Listen("tcp", net.JoinHostPort(listenHost, "0"))
+	if err != nil {
+		return nil, fmt.Errorf("bind relay listener: %w", err)
+	}
+	return &Session{listener: ln, allowed: allowed, timeout: rendezvousTimeout}, nil
+}
+
+func (s *Session) Port() int {
+	return s.listener.Addr().(*net.TCPAddr).Port
+}
+
+// Reason reports why the session ended, empty if it ended cleanly.
+func (s *Session) Reason() string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.reason
+}
+
+func (s *Session) setReason(r string) {
+	s.mu.Lock()
+	if s.reason == "" {
+		s.reason = r
+	}
+	s.mu.Unlock()
+}
+
+// Close tears the session down once. A non-empty reason is recorded only if no
+// reason has been recorded already.
+func (s *Session) Close(reason string) {
+	if reason != "" {
+		s.setReason(reason)
+	}
+	s.once.Do(func() {
+		_ = s.listener.Close()
+	})
+}
+
+// Serve accepts exactly one connection, verifies its source, and relays until
+// either side ends. It always closes the listener before returning.
+func (s *Session) Serve(stream AgentStream) error {
+	defer s.Close("")
+
+	if l, ok := s.listener.(*net.TCPListener); ok {
+		_ = l.SetDeadline(time.Now().Add(s.timeout))
+	}
+
+	conn, err := s.listener.Accept()
+	if err != nil {
+		s.setReason("guacd_timeout")
+		return fmt.Errorf("waiting for guacd: %w", err)
+	}
+	// One connection only: nothing else may claim this port.
+	s.Close("")
+
+	if !allowedRemote(conn.RemoteAddr().String(), s.allowed) {
+		_ = conn.Close()
+		s.setReason("foreign_source")
+		return fmt.Errorf("relay connection from disallowed source %s", conn.RemoteAddr())
+	}
+	defer conn.Close()
+
+	return s.relay(conn, stream)
+}
+
+func (s *Session) relay(conn net.Conn, stream AgentStream) error {
+	errCh := make(chan error, 2)
+
+	// guacd -> agent
+	go func() {
+		buf := make([]byte, chunkSize)
+		for {
+			n, err := conn.Read(buf)
+			if n > 0 {
+				chunk := make([]byte, n)
+				copy(chunk, buf[:n])
+				if sendErr := stream.Send(&pb.ProxyServerMsg{Data: chunk}); sendErr != nil {
+					errCh <- sendErr
+					return
+				}
+			}
+			if err != nil {
+				if !errors.Is(err, io.EOF) {
+					s.setReason("guacd_read_error")
+				}
+				errCh <- err
+				return
+			}
+		}
+	}()
+
+	// agent -> guacd
+	go func() {
+		for {
+			msg, err := stream.Recv()
+			if err != nil {
+				errCh <- err
+				return
+			}
+			if msg.Close != nil {
+				s.setReason(msg.Close.Reason)
+				errCh <- fmt.Errorf("agent closed relay: %s", msg.Close.Reason)
+				return
+			}
+			if len(msg.Data) > 0 {
+				if _, err := conn.Write(msg.Data); err != nil {
+					errCh <- err
+					return
+				}
+			}
+		}
+	}()
+
+	err := <-errCh
+	_ = conn.Close()
+	if errors.Is(err, io.EOF) {
+		return nil
+	}
+	return err
+}
+
+// allowedRemote reports whether remote (a host:port string) is in allowed. An
+// empty allowed list permits anything.
+func allowedRemote(remote string, allowed []string) bool {
+	host, _, err := net.SplitHostPort(remote)
+	if err != nil {
+		log.Printf("proxy: unparseable remote address %q", remote)
+		return false
+	}
+	if len(allowed) == 0 {
+		return true
+	}
+	for _, a := range allowed {
+		if a == host {
+			return true
+		}
+	}
+	return false
+}