19475484620

Committed 18 Nov 2025 05:36PM UTC coverage: 54.799% (-0.04%) from 54.843%

Build # 19475484620

Build Type

push

github

Committed by

web-flow

Commit Message

fix: databroker client updates should propagate to ssh codes (#5935)

## Summary

SSH auth code flow is causing Pomerium to not start cleanly / update
properly.

Databroker grpc client changes now propagate to the SSH code manager

## Related issues

N/A, slack thread.

## User Explanation

N/A

## Checklist

- [X] reference any related issues
- [X] updated unit tests
- [X] add appropriate label (`enhancement`, `bug`, `breaking`,
`dependencies`, `ci`)
- [X] ready for review

Run Details

4 of 66 new or added lines in 7 files covered. (6.06%)

25 existing lines in 7 files now uncovered.

28697 of 52368 relevant lines covered (54.8%)

93.63 hits per line

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

0.0

/pkg/ssh/code/reader.go

package code

import (
        "context"

        "github.com/cenkalti/backoff/v4"
        "github.com/rs/zerolog/log"
        "google.golang.org/grpc/codes"
        "google.golang.org/grpc/status"

        "github.com/pomerium/pomerium/pkg/grpc/databroker"
        "github.com/pomerium/pomerium/pkg/grpc/session"
)

type reader struct {
        clientB databroker.ClientGetter
}

var _ Reader = (*reader)(nil)

func NewReader(client databroker.ClientGetter) Reader {
        return &reader{
                clientB: client,
        }
}

func (r *reader) GetBindingRequest(ctx context.Context, id CodeID) (*session.SessionBindingRequest, bool) {
        b := backoff.WithContext(backoff.NewExponentialBackOff(), ctx)
        resp, err := backoff.RetryWithData(func() (*databroker.GetResponse, error) {
                resp, err := r.clientB.GetDataBrokerServiceClient().
                        Get(ctx, &databroker.GetRequest{
                                Type: "type.googleapis.com/session.SessionBindingRequest",
                                Id:   string(id),
                        })
                if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
                        return nil, backoff.Permanent(err)
                }
                if err != nil {
                        return nil, err
                }
                return resp, nil
        }, b)
        if err != nil {
                return nil, false
        }
        s := &session.SessionBindingRequest{}
        if err := resp.GetRecord().GetData().UnmarshalTo(s); err != nil {
                log.Err(err).Ctx(ctx).Msg("GetBindingRequest: failed to unmarshal session binding request")
                return nil, false
        }
        return s, true
}

func (r *reader) GetSessionByUserID(ctx context.Context, userID string) (map[string]*IdentitySessionPair, error) {
        ret := map[string]*IdentitySessionPair{}
        filterByUser := indexedFieldFilter("user_id", userID)

        sessBindingRecs, err := r.clientB.GetDataBrokerServiceClient().
                Query(ctx, &databroker.QueryRequest{
                        Type:   "type.googleapis.com/session.SessionBinding",
                        Filter: filterByUser,
                        Limit:  queryLimit,
                })

        if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
                return ret, nil
        }
        if err != nil {
                return nil, err
        }

        identityBindingRecs, err := r.clientB.GetDataBrokerServiceClient().
                Query(ctx, &databroker.QueryRequest{
                        Type:   "type.googleapis.com/session.IdentityBinding",
                        Filter: filterByUser,
                        Limit:  queryLimit,
                })
        if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
                identityBindingRecs = &databroker.QueryResponse{
                        Records: []*databroker.Record{},
                }
        } else if err != nil {
                return nil, err
        }

        for _, sb := range sessBindingRecs.GetRecords() {
                var sess session.SessionBinding
                if err := sb.GetData().UnmarshalTo(&sess); err != nil {
                        log.Err(err).Ctx(ctx).Msg("GetSessionByUserID: failed to unmarshal session binding")
                        continue
                }
                sessionID := sb.GetId()
                ret[sessionID] = &IdentitySessionPair{
                        SB: &sess,
                }
        }

        for _, ib := range identityBindingRecs.GetRecords() {
                var ident session.IdentityBinding
                if err := ib.GetData().UnmarshalTo(&ident); err != nil {
                        log.Err(err).Ctx(ctx).Msg("GetSessionByUserID: failed to unmarshal identity binding")
                        continue
                }
                sessionID := ib.GetId()
                val, ok := ret[sessionID]
                if !ok {
                        val = &IdentitySessionPair{}
                }
                val.IB = &ident
        }
        return ret, nil
}

1	package code
2
3	import (
4	"context"
5
6	"github.com/cenkalti/backoff/v4"
7	"github.com/rs/zerolog/log"
8	"google.golang.org/grpc/codes"
9	"google.golang.org/grpc/status"
10
11	"github.com/pomerium/pomerium/pkg/grpc/databroker"
12	"github.com/pomerium/pomerium/pkg/grpc/session"
13	)
14
15	type reader struct {
16	clientB databroker.ClientGetter
17	}
18
19	var _ Reader = (*reader)(nil)
20
NEW 21	func NewReader(client databroker.ClientGetter) Reader {	×
22	return &reader{	×
NEW 23	clientB: client,	×
24	}	×
25	}	×
26
27	func (r reader) GetBindingRequest(ctx context.Context, id CodeID) (session.SessionBindingRequest, bool) {	×
28	b := backoff.WithContext(backoff.NewExponentialBackOff(), ctx)	×
29	resp, err := backoff.RetryWithData(func() (*databroker.GetResponse, error) {	×
NEW 30	resp, err := r.clientB.GetDataBrokerServiceClient().	×
NEW 31	Get(ctx, &databroker.GetRequest{	×
NEW 32	Type: "type.googleapis.com/session.SessionBindingRequest",	×
NEW 33	Id: string(id),	×
NEW 34	})	×
35	if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {	×
36	return nil, backoff.Permanent(err)	×
37	}	×
38	if err != nil {	×
39	return nil, err	×
40	}	×
41	return resp, nil	×
42	}, b)
43	if err != nil {	×
44	return nil, false	×
45	}	×
46	s := &session.SessionBindingRequest{}	×
47	if err := resp.GetRecord().GetData().UnmarshalTo(s); err != nil {	×
48	log.Err(err).Ctx(ctx).Msg("GetBindingRequest: failed to unmarshal session binding request")	×
49	return nil, false	×
50	}	×
51	return s, true	×
52	}
53
54	func (r reader) GetSessionByUserID(ctx context.Context, userID string) (map[string]IdentitySessionPair, error) {	×
55	ret := map[string]*IdentitySessionPair{}	×
56	filterByUser := indexedFieldFilter("user_id", userID)	×
57		×
NEW 58	sessBindingRecs, err := r.clientB.GetDataBrokerServiceClient().	×
NEW 59	Query(ctx, &databroker.QueryRequest{	×
NEW 60	Type: "type.googleapis.com/session.SessionBinding",	×
NEW 61	Filter: filterByUser,	×
NEW 62	Limit: queryLimit,	×
NEW 63	})	×
64		×
65	if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {	×
66	return ret, nil	×
67	}	×
68	if err != nil {	×
69	return nil, err	×
70	}	×
71
NEW 72	identityBindingRecs, err := r.clientB.GetDataBrokerServiceClient().	×
NEW 73	Query(ctx, &databroker.QueryRequest{	×
NEW 74	Type: "type.googleapis.com/session.IdentityBinding",	×
NEW 75	Filter: filterByUser,	×
NEW 76	Limit: queryLimit,	×
NEW 77	})	×
78	if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {	×
79	identityBindingRecs = &databroker.QueryResponse{	×
80	Records: []*databroker.Record{},	×
81	}	×
82	} else if err != nil {	×
83	return nil, err	×
84	}	×
85
86	for _, sb := range sessBindingRecs.GetRecords() {	×
87	var sess session.SessionBinding	×
88	if err := sb.GetData().UnmarshalTo(&sess); err != nil {	×
89	log.Err(err).Ctx(ctx).Msg("GetSessionByUserID: failed to unmarshal session binding")	×
90	continue	×
91	}
92	sessionID := sb.GetId()	×
93	ret[sessionID] = &IdentitySessionPair{	×
94	SB: &sess,	×
95	}	×
96	}
97
98	for _, ib := range identityBindingRecs.GetRecords() {	×
99	var ident session.IdentityBinding	×
100	if err := ib.GetData().UnmarshalTo(&ident); err != nil {	×
101	log.Err(err).Ctx(ctx).Msg("GetSessionByUserID: failed to unmarshal identity binding")	×
102	continue	×
103	}
104	sessionID := ib.GetId()	×
105	val, ok := ret[sessionID]	×
106	if !ok {	×
107	val = &IdentitySessionPair{}	×
108	}	×
109	val.IB = &ident	×
110	}
111	return ret, nil	×
112	}

pomerium / pomerium / 19475484620

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