diff --git a/server/internal/proxy/registry.go b/server/internal/proxy/registry.go index d6f71b8..932d9a0 100644 --- a/server/internal/proxy/registry.go +++ b/server/internal/proxy/registry.go @@ -29,6 +29,7 @@ type Entry struct { ProxyID string InstanceID string ServerID string + Session *Session } type Registry struct { diff --git a/server/internal/proxy/session.go b/server/internal/proxy/session.go new file mode 100644 index 0000000..8d8cd0e --- /dev/null +++ b/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 +}