Skip to content

Commit 1649c83

Browse files
rubenvp8510rubenvp8510
committed
Passthrough OAuth bearer token supplied to Query service through to ES storage
Signed-off-by: Ruben Vargas <ruben.vp8510@gmail.com>
1 parent cf752d7 commit 1649c83

14 files changed

+299
-55
lines changed

‎cmd/query/app/flags.go

+10-4
Original file line numberDiff line numberDiff line change
@@ -23,10 +23,11 @@ import (
2323
)
2424

2525
const (
26-
queryPort = "query.port"
27-
queryBasePath = "query.base-path"
28-
queryStaticFiles = "query.static-files"
29-
queryUIConfig = "query.ui-config"
26+
queryPort = "query.port"
27+
queryBasePath = "query.base-path"
28+
queryStaticFiles = "query.static-files"
29+
queryUIConfig = "query.ui-config"
30+
queryTokenPropagation = "query.bearer-token-propagation"
3031
)
3132

3233
// QueryOptions holds configuration for query service
@@ -39,6 +40,8 @@ type QueryOptions struct {
3940
StaticAssets string
4041
// UIConfig is the path to a configuration file for the UI
4142
UIConfig string
43+
// BearerTokenPropagation activate/deactivate bearer token propagation to storage
44+
BearerTokenPropagation bool
4245
}
4346

4447
// AddFlags adds flags for QueryOptions
@@ -47,6 +50,8 @@ func AddFlags(flagSet *flag.FlagSet) {
4750
flagSet.String(queryBasePath, "/", "The base path for all HTTP routes, e.g. /jaeger; useful when running behind a reverse proxy")
4851
flagSet.String(queryStaticFiles, "", "The directory path override for the static assets for the UI")
4952
flagSet.String(queryUIConfig, "", "The path to the UI configuration file in JSON format")
53+
flagSet.Bool(queryTokenPropagation, true, "Allow propagation of bearer token to be used by storage plugins")
54+
5055
}
5156

5257
// InitFromViper initializes QueryOptions with properties from viper
@@ -55,5 +60,6 @@ func (qOpts *QueryOptions) InitFromViper(v *viper.Viper) *QueryOptions {
5560
qOpts.BasePath = v.GetString(queryBasePath)
5661
qOpts.StaticAssets = v.GetString(queryStaticFiles)
5762
qOpts.UIConfig = v.GetString(queryUIConfig)
63+
qOpts.BearerTokenPropagation = v.GetBool(queryTokenPropagation)
5864
return qOpts
5965
}

‎cmd/query/app/server.go

+7-4
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ import (
2121
"strings"
2222

2323
"github.com/gorilla/handlers"
24-
opentracing "github.com/opentracing/opentracing-go"
24+
"github.com/opentracing/opentracing-go"
2525
"github.com/soheilhy/cmux"
2626
"go.uber.org/zap"
2727
"google.golang.org/grpc"
@@ -80,11 +80,14 @@ func createHTTPServer(querySvc *querysvc.QueryService, queryOpts *QueryOptions,
8080

8181
apiHandler.RegisterRoutes(r)
8282
RegisterStaticHandler(r, logger, queryOpts)
83-
compressHandler := handlers.CompressHandler(r)
83+
var handler http.Handler = r
84+
if queryOpts.BearerTokenPropagation {
85+
handler = bearerTokenPropagationHandler(logger, r)
86+
}
87+
handler = handlers.CompressHandler(handler)
8488
recoveryHandler := recoveryhandler.NewRecoveryHandler(logger, true)
85-
8689
return &http.Server{
87-
Handler: recoveryHandler(compressHandler),
90+
Handler: recoveryHandler(handler),
8891
}
8992
}
9093

‎cmd/query/app/server_test.go

+2-1
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,8 @@ func TestServer(t *testing.T) {
4646
querySvc := &querysvc.QueryService{}
4747
tracer := opentracing.NoopTracer{}
4848

49-
server := NewServer(flagsSvc, querySvc, &QueryOptions{Port: ports.QueryAdminHTTP}, tracer)
49+
server := NewServer(flagsSvc, querySvc, &QueryOptions{Port: ports.QueryAdminHTTP,
50+
BearerTokenPropagation: true}, tracer)
5051
assert.NoError(t, server.Start())
5152

5253
// TODO wait for servers to come up and test http and grpc endpoints

‎cmd/query/app/token_propagation_hander_test.go

+84
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
// Copyright (c) 2019 The Jaeger Authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package app
16+
17+
import (
18+
"net/http"
19+
"net/http/httptest"
20+
"sync"
21+
"testing"
22+
23+
"github.com/stretchr/testify/assert"
24+
"go.uber.org/zap"
25+
26+
"github.com/jaegertracing/jaeger/storage/spanstore"
27+
)
28+
29+
30+
func Test_bearTokenPropagationHandler(t *testing.T) {
31+
logger := zap.NewNop()
32+
bearerToken := "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsIm5hbWUiOiJKb2huIERvZSIsImlhdCI"
33+
34+
validTokenHandler := func(stop *sync.WaitGroup) http.HandlerFunc {
35+
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
36+
ctx := r.Context()
37+
token, ok := spanstore.GetBearerToken(ctx)
38+
assert.Equal(t, token, bearerToken)
39+
assert.True(t, ok)
40+
stop.Done()
41+
})
42+
}
43+
44+
emptyHandler := func(stop *sync.WaitGroup) http.HandlerFunc {
45+
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
46+
ctx := r.Context()
47+
token, _ := spanstore.GetBearerToken(ctx)
48+
assert.Empty(t, token, bearerToken)
49+
stop.Done()
50+
})
51+
}
52+
53+
testCases := []struct {
54+
name string
55+
sendHeader bool
56+
header string
57+
handler func(stop *sync.WaitGroup) http.HandlerFunc
58+
}{
59+
{ name:"Bearer token", sendHeader: true, header: "Bearer " + bearerToken, handler:validTokenHandler},
60+
{ name:"Invalid header",sendHeader: true, header: bearerToken, handler:emptyHandler},
61+
{ name:"No header", sendHeader: false, handler:emptyHandler},
62+
{ name:"Basic Auth", sendHeader: true, header: "Basic " + bearerToken, handler:emptyHandler},
63+
{ name:"X-Forwarded-Access-Token", sendHeader: true, header: "Bearer " + bearerToken, handler:validTokenHandler},
64+
}
65+
66+
for _, testCase := range testCases {
67+
t.Run(testCase.name, func(t *testing.T) {
68+
stop := sync.WaitGroup{}
69+
stop.Add(1)
70+
r := bearerTokenPropagationHandler(logger, testCase.handler(&stop))
71+
server := httptest.NewServer(r)
72+
defer server.Close()
73+
req , err := http.NewRequest("GET", server.URL, nil)
74+
assert.Nil(t,err)
75+
if testCase.sendHeader {
76+
req.Header.Add("Authorization", testCase.header)
77+
}
78+
_, err = httpClient.Do(req)
79+
assert.Nil(t, err)
80+
stop.Wait()
81+
})
82+
}
83+
84+
}

‎cmd/query/app/token_propagation_handler.go

+51
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
// Copyright (c) 2019 The Jaeger Authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package app
16+
17+
import (
18+
"net/http"
19+
"strings"
20+
21+
"go.uber.org/zap"
22+
23+
"github.com/jaegertracing/jaeger/storage/spanstore"
24+
)
25+
26+
func bearerTokenPropagationHandler(logger *zap.Logger, h http.Handler) http.Handler {
27+
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
28+
ctx := r.Context()
29+
authHeaderValue := r.Header.Get("Authorization")
30+
// If no Authorization header is present, try with X-Forwarded-Access-Token
31+
if authHeaderValue == "" {
32+
authHeaderValue = r.Header.Get("X-Forwarded-Access-Token")
33+
}
34+
if authHeaderValue != "" {
35+
headerValue := strings.Split(authHeaderValue, " ")
36+
token := ""
37+
if len(headerValue) == 2 {
38+
// Make sure we only capture bearer token , not other types like Basic auth.
39+
if headerValue[0] == "Bearer" {
40+
token = headerValue[1]
41+
}
42+
} else {
43+
logger.Warn("Invalid authorization header, skipping bearer token propagation")
44+
}
45+
h.ServeHTTP(w, r.WithContext(spanstore.ContextWithBearerToken(ctx, token)))
46+
} else {
47+
h.ServeHTTP(w, r.WithContext(ctx))
48+
}
49+
})
50+
51+
}

‎cmd/query/main.go

+4-2
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ import (
3838
"github.com/jaegertracing/jaeger/plugin/storage"
3939
"github.com/jaegertracing/jaeger/ports"
4040
istorage "github.com/jaegertracing/jaeger/storage"
41+
"github.com/jaegertracing/jaeger/storage/spanstore"
4142
storageMetrics "github.com/jaegertracing/jaeger/storage/spanstore/metrics"
4243
)
4344

@@ -78,7 +79,9 @@ func main() {
7879
}
7980
defer closer.Close()
8081
opentracing.SetGlobalTracer(tracer)
81-
82+
queryOpts := new(app.QueryOptions).InitFromViper(v)
83+
// TODO: Need to figure out set enable/disable propagation on storage plugins.
84+
v.Set(spanstore.StoragePropagationKey, queryOpts.BearerTokenPropagation)
8285
storageFactory.InitFromViper(v)
8386
if err := storageFactory.Initialize(baseFactory, logger); err != nil {
8487
logger.Fatal("Failed to init storage factory", zap.Error(err))
@@ -98,7 +101,6 @@ func main() {
98101
dependencyReader,
99102
*queryServiceOptions)
100103

101-
queryOpts := new(app.QueryOptions).InitFromViper(v)
102104
server := app.NewServer(svc, queryService, queryOpts, tracer)
103105

104106
if err := server.Start(); err != nil {

‎pkg/es/config/config.go

+55-30
Original file line numberDiff line numberDiff line change
@@ -32,32 +32,34 @@ import (
3232

3333
"github.com/jaegertracing/jaeger/pkg/es"
3434
"github.com/jaegertracing/jaeger/pkg/es/wrapper"
35+
"github.com/jaegertracing/jaeger/storage/spanstore"
3536
storageMetrics "github.com/jaegertracing/jaeger/storage/spanstore/metrics"
3637
)
3738

3839
// Configuration describes the configuration properties needed to connect to an ElasticSearch cluster
3940
type Configuration struct {
40-
Servers []string
41-
Username string
42-
Password string
43-
TokenFilePath string
44-
Sniffer bool // https://github.com/olivere/elastic/wiki/Sniffing
45-
MaxNumSpans int // defines maximum number of spans to fetch from storage per query
46-
MaxSpanAge time.Duration `yaml:"max_span_age"` // configures the maximum lookback on span reads
47-
NumShards int64 `yaml:"shards"`
48-
NumReplicas int64 `yaml:"replicas"`
49-
Timeout time.Duration `validate:"min=500"`
50-
BulkSize int
51-
BulkWorkers int
52-
BulkActions int
53-
BulkFlushInterval time.Duration
54-
IndexPrefix string
55-
TagsFilePath string
56-
AllTagsAsFields bool
57-
TagDotReplacement string
58-
Enabled bool
59-
TLS TLSConfig
60-
UseReadWriteAliases bool
41+
Servers []string
42+
Username string
43+
Password string
44+
TokenFilePath string
45+
AllowTokenFromContext bool
46+
Sniffer bool // https://github.com/olivere/elastic/wiki/Sniffing
47+
MaxNumSpans int // defines maximum number of spans to fetch from storage per query
48+
MaxSpanAge time.Duration `yaml:"max_span_age"` // configures the maximum lookback on span reads
49+
NumShards int64 `yaml:"shards"`
50+
NumReplicas int64 `yaml:"replicas"`
51+
Timeout time.Duration `validate:"min=500"`
52+
BulkSize int
53+
BulkWorkers int
54+
BulkActions int
55+
BulkFlushInterval time.Duration
56+
IndexPrefix string
57+
TagsFilePath string
58+
AllTagsAsFields bool
59+
TagDotReplacement string
60+
Enabled bool
61+
TLS TLSConfig
62+
UseReadWriteAliases bool
6163
}
6264

6365
// TLSConfig describes the configuration properties to connect tls enabled ElasticSearch cluster
@@ -90,7 +92,7 @@ func (c *Configuration) NewClient(logger *zap.Logger, metricsFactory metrics.Fac
9092
if len(c.Servers) < 1 {
9193
return nil, errors.New("No servers specified")
9294
}
93-
options, err := c.getConfigOptions()
95+
options, err := c.getConfigOptions(logger)
9496
if err != nil {
9597
return nil, err
9698
}
@@ -247,8 +249,13 @@ func (c *Configuration) IsEnabled() bool {
247249
}
248250

249251
// getConfigOptions wraps the configs to feed to the ElasticSearch client init
250-
func (c *Configuration) getConfigOptions() ([]elastic.ClientOptionFunc, error) {
251-
options := []elastic.ClientOptionFunc{elastic.SetURL(c.Servers...), elastic.SetSniff(c.Sniffer)}
252+
func (c *Configuration) getConfigOptions(logger *zap.Logger) ([]elastic.ClientOptionFunc, error) {
253+
254+
options := []elastic.ClientOptionFunc{elastic.SetURL(c.Servers...), elastic.SetSniff(c.Sniffer),
255+
// Disable health check when token from context is allowed, this is because at this time
256+
// we don' have a valid token to do the check ad if we don't disable the check the service that
257+
// uses this won't start.
258+
elastic.SetHealthcheck(!c.AllowTokenFromContext)}
252259
httpClient := &http.Client{
253260
Timeout: c.Timeout,
254261
}
@@ -271,14 +278,24 @@ func (c *Configuration) getConfigOptions() ([]elastic.ClientOptionFunc, error) {
271278
}
272279
httpTransport.TLSClientConfig = &tls.Config{RootCAs: ca}
273280
}
281+
282+
token := ""
274283
if c.TokenFilePath != "" {
275-
token, err := loadToken(c.TokenFilePath)
284+
if c.AllowTokenFromContext {
285+
logger.Warn("Token file and token propagation are both enabled, token from file won't be used")
286+
}
287+
tokenFromFile, err := loadToken(c.TokenFilePath)
276288
if err != nil {
277289
return nil, err
278290
}
291+
token = tokenFromFile
292+
}
293+
294+
if token != "" || c.AllowTokenFromContext {
279295
httpClient.Transport = &tokenAuthTransport{
280-
token: token,
281-
wrapped: httpTransport,
296+
token: token,
297+
allowOverrideFromCtx: c.AllowTokenFromContext,
298+
wrapped: httpTransport,
282299
}
283300
} else {
284301
httpClient.Transport = httpTransport
@@ -329,12 +346,20 @@ func (tlsConfig *TLSConfig) loadPrivateKey() (*tls.Certificate, error) {
329346

330347
// TokenAuthTransport
331348
type tokenAuthTransport struct {
332-
token string
333-
wrapped *http.Transport
349+
token string
350+
allowOverrideFromCtx bool
351+
wrapped *http.Transport
334352
}
335353

336354
func (tr *tokenAuthTransport) RoundTrip(r *http.Request) (*http.Response, error) {
337-
r.Header.Set("Authorization", "Bearer "+tr.token)
355+
token := tr.token
356+
if tr.allowOverrideFromCtx {
357+
headerToken, _ := spanstore.GetBearerToken(r.Context())
358+
if headerToken != "" {
359+
token = headerToken
360+
}
361+
}
362+
r.Header.Set("Authorization", "Bearer "+token)
338363
return tr.wrapped.RoundTrip(r)
339364
}
340365

‎plugin/storage/es/options.go

+3
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import (
2222
"github.com/spf13/viper"
2323

2424
"github.com/jaegertracing/jaeger/pkg/es/config"
25+
"github.com/jaegertracing/jaeger/storage/spanstore"
2526
)
2627

2728
const (
@@ -255,6 +256,8 @@ func initFromViper(cfg *namespaceConfig, v *viper.Viper) {
255256
cfg.TagDotReplacement = v.GetString(cfg.namespace + suffixTagDeDotChar)
256257
cfg.UseReadWriteAliases = v.GetBool(cfg.namespace + suffixReadAlias)
257258
cfg.Enabled = v.GetBool(cfg.namespace + suffixEnabled)
259+
// TODO: Need to figure out a better way for do this.
260+
cfg.AllowTokenFromContext = v.GetBool(spanstore.StoragePropagationKey)
258261
}
259262

260263
// GetPrimary returns primary configuration.

0 commit comments

Comments
 (0)