19943658143

Committed 04 Dec 2025 08:56PM UTC coverage: 52.785% (-0.6%) from 53.375%

Build # 19943658143

Build Type

push

github

Committed by

web-flow

Commit Message

ssh: policy index api + in-memory implementation (#5948)

This adds a new PolicyIndexer interface, which is responsible for
processing updates to configuration, session records, and active ssh
streams. All active streams can subscribe to receive incremental updates
as state changes.

This allows reverse tunnel routes to scale, since each reverse tunnel
connection does not need to independently compute its own set of
authorized upstream tunnel routes. The implementation can be swapped out
to support different environments.

An in-memory implementation is included, as well as a conformance test
suite that can be run against any implementation of the PolicyIndexer
interface to validate that it works as expected in a variety of
scenarios.

---------

Co-authored-by: Denis Mishin <dmishin@pomerium.com>

Run Details

395 of 1313 new or added lines in 11 files covered. (30.08%)

10 existing lines in 5 files now uncovered.

28880 of 54713 relevant lines covered (52.78%)

85.28 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

44.05

/pkg/ssh/cli.go

package ssh

import (
        "errors"
        "fmt"
        "io"
        "strings"

        tea "charm.land/bubbletea/v2"
        "github.com/muesli/termenv"
        "github.com/spf13/cobra"

        "github.com/pomerium/envoy-custom/api/extensions/filters/network/ssh"
        "github.com/pomerium/pomerium/config"
        "github.com/pomerium/pomerium/pkg/ssh/tui"
)

type CLI struct {
        *cobra.Command
        tui      *tea.Program
        tuiDone  chan struct{}
        msgQueue chan tea.Msg
        ptyInfo  *ssh.SSHDownstreamPTYInfo
        username string
        stdin    io.Reader
        stdout   io.Writer
        stderr   io.Writer
}

func NewCLI(
        cfg *config.Config,
        ctrl ChannelControlInterface,
        ptyInfo *ssh.SSHDownstreamPTYInfo,
        stdin io.Reader,
        stdout io.Writer,
        stderr io.Writer,
) *CLI {
        cmd := &cobra.Command{
                Use: "pomerium",
                PersistentPreRunE: func(cmd *cobra.Command, _ []string) error {
                        _, cmdIsInteractive := cmd.Annotations["interactive"]
                        switch {
                        case (ptyInfo == nil) && cmdIsInteractive:
                                return fmt.Errorf("\x1b[31m'%s' is an interactive command and requires a TTY (try passing '-t' to ssh)\x1b[0m", cmd.Use)
                        }
                        return nil
                },
        }

        cmd.CompletionOptions.DisableDefaultCmd = true
        // set a non-nil args list, otherwise it will read from os.Args by default
        cmd.SetArgs([]string{})
        cmd.SetIn(stdin)
        cmd.SetOut(stderr)
        cmd.SetErr(stderr)
        cmd.SilenceUsage = true

        cli := &CLI{
                Command:  cmd,
                tui:      nil,
                tuiDone:  make(chan struct{}),
                msgQueue: make(chan tea.Msg, 256),
                ptyInfo:  ptyInfo,
                username: *ctrl.Username(),
                stdin:    stdin,
                stdout:   stdout,
                stderr:   stderr,
        }

        if cfg.Options.IsRuntimeFlagSet(config.RuntimeFlagSSHRoutesPortal) {
                cli.AddPortalCommand(ctrl)
        }
        cli.AddTunnelCommand(ctrl)
        cli.AddLogoutCommand(ctrl)
        cli.AddWhoamiCommand(ctrl)

        return cli
}

func (cli *CLI) AddLogoutCommand(_ ChannelControlInterface) {
        cli.AddCommand(&cobra.Command{
                Use:           "logout",
                Short:         "Log out",
                SilenceErrors: true,
                RunE: func(_ *cobra.Command, _ []string) error {
                        _, _ = cli.stderr.Write([]byte("Logged out successfully\n"))
                        return ErrDeleteSessionOnExit
                },
        })
}

func (cli *CLI) AddWhoamiCommand(ctrl ChannelControlInterface) {
        cli.AddCommand(&cobra.Command{
                Use:   "whoami",
                Short: "Show details for the current session",
                RunE: func(cmd *cobra.Command, _ []string) error {
                        s, err := ctrl.FormatSession(cmd.Context())
                        if err != nil {
                                return fmt.Errorf("couldn't fetch session: %w", err)
                        }
                        _, _ = cli.stderr.Write(s)
                        return nil
                },
        })
}

type sshEnviron struct {
        Env map[string]string
}

// Environ implements termenv.Environ.
func (s *sshEnviron) Environ() []string {
        kv := make([]string, 0, len(s.Env))
        for k, v := range s.Env {
                kv = append(kv, fmt.Sprintf("%s=%s", k, v))
        }
        return kv
}

// Getenv implements termenv.Environ.
func (s *sshEnviron) Getenv(key string) string {
        return s.Env[key]
}

var _ termenv.Environ = (*sshEnviron)(nil)

const (
        ptyWidthMax  = 512
        ptyHeightMax = 512
)

func (cli *CLI) AddTunnelCommand(ctrl ChannelControlInterface) {
        cli.AddCommand(&cobra.Command{
                Use:    "tunnel",
                Short:  "tunnel status",
                Hidden: true,
                Annotations: map[string]string{
                        "interactive": "",
                },
                RunE: func(cmd *cobra.Command, _ []string) error {
                        env := &sshEnviron{
                                Env: map[string]string{
                                        "TERM":      cli.ptyInfo.TermEnv,
                                        "TTY_FORCE": "1",

                                        // Important: disables synchronized output querying which I think
                                        // might be causing the renderer to get stuck
                                        "SSH_TTY": "1",
                                },
                        }

                        prog := tui.NewTunnelStatusProgram(cmd.Context(),
                                tea.WithInput(cli.stdin),
                                tea.WithWindowSize(int(min(cli.ptyInfo.WidthColumns, ptyWidthMax)), int(min(cli.ptyInfo.HeightRows, ptyHeightMax))),
                                tea.WithOutput(termenv.NewOutput(cli.stdout, termenv.WithEnvironment(env), termenv.WithUnsafe())),
                                tea.WithEnvironment(env.Environ()),
                        )
                        cli.tui = prog.Program
                        defer close(cli.tuiDone)

                        initDone := make(chan struct{})
                        mgr := ctrl.PortForwardManager()

                        go func() {
                                mgr.AddUpdateListener(prog)
                                defer mgr.RemoveUpdateListener(prog)
                                close(initDone)
                                for {
                                        select {
                                        case <-cli.tuiDone:
                                                return
                                        case msg := <-cli.msgQueue:
                                                cli.tui.Send(msg)
                                        }
                                }
                        }()
                        _, err := prog.Run()
                        <-initDone
                        if err != nil {
                                return err
                        }
                        return nil
                },
        })
}

// ErrHandoff is a sentinel error to indicate that the command triggered a handoff,
// and we should not automatically disconnect
var ErrHandoff = errors.New("handoff")

// ErrDeleteSessionOnExit is a sentinel error to indicate that the authorized
// session should be deleted once the SSH connection ends.
var ErrDeleteSessionOnExit = errors.New("delete_session_on_exit")

func (cli *CLI) AddPortalCommand(ctrl ChannelControlInterface) {
        cli.AddCommand(&cobra.Command{
                Use:   "portal",
                Short: "Interactive route portal",
                Annotations: map[string]string{
                        "interactive": "",
                },
                RunE: func(cmd *cobra.Command, _ []string) error {
                        var routes []string
                        for r := range ctrl.AllSSHRoutes() {
                                routes = append(routes, fmt.Sprintf("%s@%s", *ctrl.Username(), strings.TrimPrefix(r.From, "ssh://")))
                        }
                        env := &sshEnviron{
                                Env: map[string]string{
                                        "TERM":      cli.ptyInfo.TermEnv,
                                        "TTY_FORCE": "1",
                                        "SSH_TTY":   "1",
                                },
                        }
                        signedWidth := int(min(cli.ptyInfo.WidthColumns, ptyWidthMax))
                        signedHeight := int(min(cli.ptyInfo.HeightRows, ptyHeightMax))
                        prog := tui.NewPortalProgram(cmd.Context(), routes, max(0, signedWidth-2), max(0, signedHeight-2),
                                tea.WithInput(cli.stdin),
                                tea.WithWindowSize(signedWidth, signedHeight),
                                tea.WithOutput(termenv.NewOutput(cli.stdout, termenv.WithEnvironment(env), termenv.WithUnsafe())),
                                tea.WithEnvironment(env.Environ()),
                        )
                        cli.tui = prog.Program

                        choice, err := prog.Run()
                        if err != nil {
                                return err
                        }
                        if choice == "" {
                                return nil // quit/ctrl+c
                        }

                        username, hostname, _ := strings.Cut(choice, "@")
                        // Perform authorize check for this route
                        if username != cli.username {
                                panic("bug: username mismatch")
                        }
                        if hostname == "" {
                                panic("bug: hostname is empty")
                        }
                        handoffMsg, err := ctrl.PrepareHandoff(cmd.Context(), hostname, cli.ptyInfo)
                        if err != nil {
                                return err
                        }
                        if err := ctrl.SendControlAction(handoffMsg); err != nil {
                                return err
                        }
                        return ErrHandoff
                },
        })
}

func (cli *CLI) SendTeaMsg(msg tea.Msg) {
        select {
        case <-cli.tuiDone:
        case cli.msgQueue <- msg:
        }
}

1	package ssh
2
3	import (
4	"errors"
5	"fmt"
6	"io"
7	"strings"
8
9	tea "charm.land/bubbletea/v2"
10	"github.com/muesli/termenv"
11	"github.com/spf13/cobra"
12
13	"github.com/pomerium/envoy-custom/api/extensions/filters/network/ssh"
14	"github.com/pomerium/pomerium/config"
15	"github.com/pomerium/pomerium/pkg/ssh/tui"
16	)
17
18	type CLI struct {
19	*cobra.Command
20	tui *tea.Program
21	tuiDone chan struct{}
22	msgQueue chan tea.Msg
23	ptyInfo *ssh.SSHDownstreamPTYInfo
24	username string
25	stdin io.Reader
26	stdout io.Writer
27	stderr io.Writer
28	}
29
30	func NewCLI(
31	cfg *config.Config,
32	ctrl ChannelControlInterface,
33	ptyInfo *ssh.SSHDownstreamPTYInfo,
34	stdin io.Reader,
35	stdout io.Writer,
36	stderr io.Writer,
37	) *CLI {	29✔
38	cmd := &cobra.Command{	29✔
39	Use: "pomerium",	29✔
40	PersistentPreRunE: func(cmd *cobra.Command, _ []string) error {	52✔
41	_, cmdIsInteractive := cmd.Annotations["interactive"]	23✔
42	switch {	23✔
43	case (ptyInfo == nil) && cmdIsInteractive:	1✔
44	return fmt.Errorf("\x1b[31m'%s' is an interactive command and requires a TTY (try passing '-t' to ssh)\x1b[0m", cmd.Use)	1✔
45	}
46	return nil	22✔
47	},
48	}
49
50	cmd.CompletionOptions.DisableDefaultCmd = true	29✔
51	// set a non-nil args list, otherwise it will read from os.Args by default	29✔
52	cmd.SetArgs([]string{})	29✔
53	cmd.SetIn(stdin)	29✔
54	cmd.SetOut(stderr)	29✔
55	cmd.SetErr(stderr)	29✔
56	cmd.SilenceUsage = true	29✔
57		29✔
58	cli := &CLI{	29✔
59	Command: cmd,	29✔
60	tui: nil,	29✔
61	tuiDone: make(chan struct{}),	29✔
62	msgQueue: make(chan tea.Msg, 256),	29✔
63	ptyInfo: ptyInfo,	29✔
64	username: *ctrl.Username(),	29✔
65	stdin: stdin,	29✔
66	stdout: stdout,	29✔
67	stderr: stderr,	29✔
68	}	29✔
69		29✔
70	if cfg.Options.IsRuntimeFlagSet(config.RuntimeFlagSSHRoutesPortal) {	49✔
71	cli.AddPortalCommand(ctrl)	20✔
72	}	20✔
73	cli.AddTunnelCommand(ctrl)	29✔
74	cli.AddLogoutCommand(ctrl)	29✔
75	cli.AddWhoamiCommand(ctrl)	29✔
76		29✔
77	return cli	29✔
78	}
79
80	func (cli *CLI) AddLogoutCommand(_ ChannelControlInterface) {	29✔
81	cli.AddCommand(&cobra.Command{	29✔
82	Use: "logout",	29✔
83	Short: "Log out",	29✔
84	SilenceErrors: true,	29✔
85	RunE: func(_ *cobra.Command, _ []string) error {	39✔
86	_, _ = cli.stderr.Write([]byte("Logged out successfully\n"))	10✔
87	return ErrDeleteSessionOnExit	10✔
88	},	10✔
89	})
90	}
91
92	func (cli *CLI) AddWhoamiCommand(ctrl ChannelControlInterface) {	29✔
93	cli.AddCommand(&cobra.Command{	29✔
94	Use: "whoami",	29✔
95	Short: "Show details for the current session",	29✔
96	RunE: func(cmd *cobra.Command, _ []string) error {	41✔
97	s, err := ctrl.FormatSession(cmd.Context())	12✔
98	if err != nil {	14✔
99	return fmt.Errorf("couldn't fetch session: %w", err)	2✔
100	}	2✔
101	_, _ = cli.stderr.Write(s)	10✔
102	return nil	10✔
103	},
104	})
105	}
106
107	type sshEnviron struct {
108	Env map[string]string
109	}
110
111	// Environ implements termenv.Environ.
112	func (s *sshEnviron) Environ() []string {	×
113	kv := make([]string, 0, len(s.Env))	×
114	for k, v := range s.Env {	×
115	kv = append(kv, fmt.Sprintf("%s=%s", k, v))	×
116	}	×
117	return kv	×
118	}
119
120	// Getenv implements termenv.Environ.
121	func (s *sshEnviron) Getenv(key string) string {	×
122	return s.Env[key]	×
123	}	×
124
125	var _ termenv.Environ = (*sshEnviron)(nil)
126
127	const (
128	ptyWidthMax = 512
129	ptyHeightMax = 512
130	)
131
132	func (cli *CLI) AddTunnelCommand(ctrl ChannelControlInterface) {	29✔
133	cli.AddCommand(&cobra.Command{	29✔
134	Use: "tunnel",	29✔
135	Short: "tunnel status",	29✔
136	Hidden: true,	29✔
137	Annotations: map[string]string{	29✔
138	"interactive": "",	29✔
139	},	29✔
140	RunE: func(cmd *cobra.Command, _ []string) error {	29✔
141	env := &sshEnviron{	×
142	Env: map[string]string{	×
143	"TERM": cli.ptyInfo.TermEnv,	×
144	"TTY_FORCE": "1",	×
145		×
146	// Important: disables synchronized output querying which I think	×
147	// might be causing the renderer to get stuck	×
148	"SSH_TTY": "1",	×
149	},	×
150	}	×
151		×
152	prog := tui.NewTunnelStatusProgram(cmd.Context(),	×
153	tea.WithInput(cli.stdin),	×
154	tea.WithWindowSize(int(min(cli.ptyInfo.WidthColumns, ptyWidthMax)), int(min(cli.ptyInfo.HeightRows, ptyHeightMax))),	×
155	tea.WithOutput(termenv.NewOutput(cli.stdout, termenv.WithEnvironment(env), termenv.WithUnsafe())),	×
156	tea.WithEnvironment(env.Environ()),	×
157	)	×
158	cli.tui = prog.Program	×
159	defer close(cli.tuiDone)	×
160		×
161	initDone := make(chan struct{})	×
NEW 162	mgr := ctrl.PortForwardManager()	×
NEW 163		×
164	go func() {	×
NEW 165	mgr.AddUpdateListener(prog)	×
NEW 166	defer mgr.RemoveUpdateListener(prog)	×
167	close(initDone)	×
168	for {	×
169	select {	×
170	case <-cli.tuiDone:	×
171	return	×
172	case msg := <-cli.msgQueue:	×
173	cli.tui.Send(msg)	×
174	}
175	}
176	}()
177	_, err := prog.Run()	×
178	<-initDone	×
179	if err != nil {	×
180	return err	×
181	}	×
182	return nil	×
183	},
184	})
185	}
186
187	// ErrHandoff is a sentinel error to indicate that the command triggered a handoff,
188	// and we should not automatically disconnect
189	var ErrHandoff = errors.New("handoff")
190
191	// ErrDeleteSessionOnExit is a sentinel error to indicate that the authorized
192	// session should be deleted once the SSH connection ends.
193	var ErrDeleteSessionOnExit = errors.New("delete_session_on_exit")
194
195	func (cli *CLI) AddPortalCommand(ctrl ChannelControlInterface) {	20✔
196	cli.AddCommand(&cobra.Command{	20✔
197	Use: "portal",	20✔
198	Short: "Interactive route portal",	20✔
199	Annotations: map[string]string{	20✔
200	"interactive": "",	20✔
201	},	20✔
202	RunE: func(cmd *cobra.Command, _ []string) error {	20✔
203	var routes []string	×
204	for r := range ctrl.AllSSHRoutes() {	×
205	routes = append(routes, fmt.Sprintf("%s@%s", *ctrl.Username(), strings.TrimPrefix(r.From, "ssh://")))	×
206	}	×
207	env := &sshEnviron{	×
208	Env: map[string]string{	×
209	"TERM": cli.ptyInfo.TermEnv,	×
210	"TTY_FORCE": "1",	×
211	"SSH_TTY": "1",	×
212	},	×
213	}	×
214	signedWidth := int(min(cli.ptyInfo.WidthColumns, ptyWidthMax))	×
215	signedHeight := int(min(cli.ptyInfo.HeightRows, ptyHeightMax))	×
216	prog := tui.NewPortalProgram(cmd.Context(), routes, max(0, signedWidth-2), max(0, signedHeight-2),	×
217	tea.WithInput(cli.stdin),	×
218	tea.WithWindowSize(signedWidth, signedHeight),	×
219	tea.WithOutput(termenv.NewOutput(cli.stdout, termenv.WithEnvironment(env), termenv.WithUnsafe())),	×
220	tea.WithEnvironment(env.Environ()),	×
221	)	×
222	cli.tui = prog.Program	×
223		×
224	choice, err := prog.Run()	×
225	if err != nil {	×
226	return err	×
227	}	×
228	if choice == "" {	×
229	return nil // quit/ctrl+c	×
230	}	×
231
232	username, hostname, _ := strings.Cut(choice, "@")	×
233	// Perform authorize check for this route	×
234	if username != cli.username {	×
235	panic("bug: username mismatch")	×
236	}
237	if hostname == "" {	×
238	panic("bug: hostname is empty")	×
239	}
240	handoffMsg, err := ctrl.PrepareHandoff(cmd.Context(), hostname, cli.ptyInfo)	×
241	if err != nil {	×
242	return err	×
243	}	×
244	if err := ctrl.SendControlAction(handoffMsg); err != nil {	×
245	return err	×
246	}	×
247	return ErrHandoff	×
248	},
249	})
250	}
251
252	func (cli *CLI) SendTeaMsg(msg tea.Msg) {	×
253	select {	×
254	case <-cli.tuiDone:	×
255	case cli.msgQueue <- msg:	×
256	}
257	}

pomerium / pomerium / 19943658143

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous