@@ -6,6 +6,7 @@ import (
66 "fmt"
77 "net/http"
88 "net/http/httptest"
9+ "sync"
910 "time"
1011
1112 middlewareapi "github.com/oauth2-proxy/oauth2-proxy/v7/pkg/apis/middleware"
@@ -17,9 +18,74 @@ import (
1718 . "github.com/onsi/gomega"
1819)
1920
21+type testLock struct {
22+ locked bool
23+ obtainOnAttempt int
24+ obtainAttempts int
25+ obtainError error
26+}
27+
28+func (l *testLock) Obtain(_ context.Context, _ time.Duration) error {
29+ l.obtainAttempts++
30+ if l.obtainAttempts < l.obtainOnAttempt {
31+ return sessionsapi.ErrLockNotObtained
32+ }
33+ if l.obtainError != nil {
34+ return l.obtainError
35+ }
36+ l.locked = true
37+ return nil
38+}
39+
40+func (l *testLock) Peek(_ context.Context) (bool, error) {
41+ return l.locked, nil
42+}
43+
44+func (l *testLock) Refresh(_ context.Context, _ time.Duration) error {
45+ return nil
46+}
47+
48+func (l *testLock) Release(_ context.Context) error {
49+ l.locked = false
50+ return nil
51+}
52+
53+type testLockConcurrent struct {
54+ mu sync.RWMutex
55+ locked bool
56+}
57+
58+func (l *testLockConcurrent) Obtain(_ context.Context, _ time.Duration) error {
59+ l.mu.Lock()
60+ defer l.mu.Unlock()
61+ if l.locked {
62+ return sessionsapi.ErrLockNotObtained
63+ }
64+ l.locked = true
65+ return nil
66+}
67+
68+func (l *testLockConcurrent) Peek(_ context.Context) (bool, error) {
69+ l.mu.RLock()
70+ defer l.mu.RUnlock()
71+ return l.locked, nil
72+}
73+
74+func (l *testLockConcurrent) Refresh(_ context.Context, _ time.Duration) error {
75+ return nil
76+}
77+
78+func (l *testLockConcurrent) Release(_ context.Context) error {
79+ l.mu.Lock()
80+ defer l.mu.Unlock()
81+ l.locked = false
82+ return nil
83+}
84+
2085var _ = Describe("Stored Session Suite", func() {
2186 const (
2287 refresh = "Refresh"
88+ refreshed = "Refreshed"
2389 noRefresh = "NoRefresh"
2490 notImplemented = "NotImplemented"
2591 )
@@ -34,7 +100,7 @@ var _ = Describe("Stored Session Suite", func() {
34100 var defaultRefreshFunc = func(_ context.Context, ss *sessionsapi.SessionState) (bool, error) {
35101 switch ss.RefreshToken {
36102 case refresh:
37− ss.RefreshToken = "Refreshed"
103+ ss.RefreshToken = refreshed
38104 return true, nil
39105 case noRefresh:
40106 return false, nil
@@ -181,6 +247,7 @@ var _ = Describe("Stored Session Suite", func() {
181247 RefreshToken: noRefresh,
182248 CreatedAt: &createdPast,
183249 ExpiresOn: &createdFuture,
250+ Lock: &sessionsapi.NoOpLock{},
184251 },
185252 store: defaultSessionStore,
186253 refreshPeriod: 1 * time.Minute,
@@ -222,6 +289,7 @@ var _ = Describe("Stored Session Suite", func() {
222289 RefreshToken: "Refreshed",
223290 CreatedAt: &now,
224291 ExpiresOn: &createdFuture,
292+ Lock: &sessionsapi.NoOpLock{},
225293 },
226294 store: defaultSessionStore,
227295 refreshPeriod: 1 * time.Minute,
@@ -237,6 +305,7 @@ var _ = Describe("Stored Session Suite", func() {
237305 RefreshToken: "RefreshError",
238306 CreatedAt: &createdPast,
239307 ExpiresOn: &createdFuture,
308+ Lock: &sessionsapi.NoOpLock{},
240309 },
241310 store: defaultSessionStore,
242311 refreshPeriod: 1 * time.Minute,
@@ -266,15 +335,116 @@ var _ = Describe("Stored Session Suite", func() {
266335 validateSession: defaultValidateFunc,
267336 }),
268337 )
338+
339+ type storedSessionLoaderConcurrentTableInput struct {
340+ existingSession *sessionsapi.SessionState
341+ refreshPeriod time.Duration
342+ numConcReqs int
343+ }
344+
345+ DescribeTable("when serving concurrent requests",
346+ func(in storedSessionLoaderConcurrentTableInput) {
347+ lockConc := &testLockConcurrent{}
348+
349+ lock := &sync.RWMutex{}
350+ existingSession := *in.existingSession // deep copy existingSession state
351+ existingSession.Lock = lockConc
352+ store := &fakeSessionStore{
353+ LoadFunc: func(req *http.Request) (*sessionsapi.SessionState, error) {
354+ lock.RLock()
355+ defer lock.RUnlock()
356+ session := existingSession
357+ return &session, nil
358+ },
359+ SaveFunc: func(_ http.ResponseWriter, _ *http.Request, session *sessionsapi.SessionState) error {
360+ lock.Lock()
361+ defer lock.Unlock()
362+ existingSession = *session
363+ return nil
364+ },
365+ }
366+
367+ refreshedChan := make(chan bool, in.numConcReqs)
368+ for i := 0; i < in.numConcReqs; i++ {
369+ go func(refreshedChan chan bool, lockConc sessionsapi.Lock) {
370+ scope := &middlewareapi.RequestScope{
371+ Session: nil,
372+ }
373+
374+ // Set up the request with the request header and a request scope
375+ req := httptest.NewRequest("", "/", nil)
376+ req = middlewareapi.AddRequestScope(req, scope)
377+
378+ rw := httptest.NewRecorder()
379+
380+ sessionRefreshed := false
381+ opts := &StoredSessionLoaderOptions{
382+ SessionStore: store,
383+ RefreshPeriod: in.refreshPeriod,
384+ RefreshSession: func(ctx context.Context, s *sessionsapi.SessionState) (bool, error) {
385+ time.Sleep(10 * time.Millisecond)
386+ sessionRefreshed = true
387+ return true, nil
388+ },
389+ ValidateSession: func(context.Context, *sessionsapi.SessionState) bool {
390+ return true
391+ },
392+ }
393+
394+ handler := NewStoredSessionLoader(opts)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
395+ handler.ServeHTTP(rw, req)
396+
397+ refreshedChan <- sessionRefreshed
398+ }(refreshedChan, lockConc)
399+ }
400+ var refreshedSlice []bool
401+ for i := 0; i < in.numConcReqs; i++ {
402+ refreshedSlice = append(refreshedSlice, <-refreshedChan)
403+ }
404+ sessionRefreshedCount := 0
405+ for _, sessionRefreshed := range refreshedSlice {
406+ if sessionRefreshed {
407+ sessionRefreshedCount++
408+ }
409+ }
410+ Expect(sessionRefreshedCount).To(Equal(1))
411+ },
412+ Entry("with two concurrent requests", storedSessionLoaderConcurrentTableInput{
413+ existingSession: &sessionsapi.SessionState{
414+ RefreshToken: refresh,
415+ CreatedAt: &createdPast,
416+ },
417+ numConcReqs: 2,
418+ refreshPeriod: 1 * time.Minute,
419+ }),
420+ Entry("with 5 concurrent requests", storedSessionLoaderConcurrentTableInput{
421+ existingSession: &sessionsapi.SessionState{
422+ RefreshToken: refresh,
423+ CreatedAt: &createdPast,
424+ },
425+ numConcReqs: 5,
426+ refreshPeriod: 1 * time.Minute,
427+ }),
428+ Entry("with one request", storedSessionLoaderConcurrentTableInput{
429+ existingSession: &sessionsapi.SessionState{
430+ RefreshToken: refresh,
431+ CreatedAt: &createdPast,
432+ },
433+ numConcReqs: 1,
434+ refreshPeriod: 1 * time.Minute,
435+ }),
436+ )
269437 })
270438
271439 Context("refreshSessionIfNeeded", func() {
272440 type refreshSessionIfNeededTableInput struct {
273− refreshPeriod time.Duration
274− session *sessionsapi.SessionState
275− expectedErr error
276− expectRefreshed bool
277− expectValidated bool
441+ refreshPeriod time.Duration
442+ session *sessionsapi.SessionState
443+ concurrentSessionRefresh bool
444+ expectedErr error
445+ expectRefreshed bool
446+ expectValidated bool
447+ expectedLockObtained bool
278448 }
279449
280450 createdPast := time.Now().Add(-5 * time.Minute)
@@ -285,9 +455,28 @@ var _ = Describe("Stored Session Suite", func() {
285455 refreshed := false
286456 validated := false
287457
458+ session := &sessionsapi.SessionState{}
459+ *session = *in.session
460+ if in.concurrentSessionRefresh {
461+ // Update the session that Load returns.
462+ // This simulates a concurrent refresh in the background.
463+ session.CreatedAt = &createdFuture
464+ }
465+ store := &fakeSessionStore{
466+ LoadFunc: func(req *http.Request) (*sessionsapi.SessionState, error) {
467+ // Loading the session from the provider creates a new lock
468+ session.Lock = &testLock{}
469+ return session, nil
470+ },
471+ SaveFunc: func(_ http.ResponseWriter, _ *http.Request, s *sessionsapi.SessionState) error {
472+ *session = *s
473+ return nil
474+ },
475+ }
476+
288477 s := &storedSessionLoader{
289478 refreshPeriod: in.refreshPeriod,
290− store: &fakeSessionStore{},
479+ store: store,
291480 sessionRefresher: func(_ context.Context, ss *sessionsapi.SessionState) (bool, error) {
292481 refreshed = true
293482 switch ss.RefreshToken {
@@ -316,89 +505,129 @@ var _ = Describe("Stored Session Suite", func() {
316505 }
317506 Expect(refreshed).To(Equal(in.expectRefreshed))
318507 Expect(validated).To(Equal(in.expectValidated))
508+ testLock, ok := in.session.Lock.(*testLock)
509+ Expect(ok).To(Equal(true))
510+
511+ if in.expectedLockObtained {
512+ Expect(testLock.obtainAttempts).Should(BeNumerically(">", 0), "Expected at least one attempt at obtaining the session lock")
513+ }
514+ Expect(testLock.locked).To(BeFalse(), "Expected lock should always be released")
319515 },
320516 Entry("when the refresh period is 0, and the session does not need refreshing", refreshSessionIfNeededTableInput{
321517 refreshPeriod: time.Duration(0),
322518 session: &sessionsapi.SessionState{
323519 RefreshToken: refresh,
324520 CreatedAt: &createdFuture,
521+ Lock: &testLock{},
325522 },
326− expectedErr: nil,
327− expectRefreshed: false,
328− expectValidated: false,
523+ expectedErr: nil,
524+ expectRefreshed: false,
525+ expectValidated: false,
526+ expectedLockObtained: false,
329527 }),
330528 Entry("when the refresh period is 0, and the session needs refreshing", refreshSessionIfNeededTableInput{
331529 refreshPeriod: time.Duration(0),
332530 session: &sessionsapi.SessionState{
333531 RefreshToken: refresh,
334532 CreatedAt: &createdPast,
533+ Lock: &testLock{},
335534 },
336− expectedErr: nil,
337− expectRefreshed: false,
338− expectValidated: false,
535+ expectedErr: nil,
536+ expectRefreshed: false,
537+ expectValidated: false,
538+ expectedLockObtained: false,
339539 }),
340540 Entry("when the session does not need refreshing", refreshSessionIfNeededTableInput{
341541 refreshPeriod: 1 * time.Minute,
342542 session: &sessionsapi.SessionState{
343543 RefreshToken: refresh,
344544 CreatedAt: &createdFuture,
545+ Lock: &testLock{},
345546 },
346− expectedErr: nil,
347− expectRefreshed: false,
348− expectValidated: false,
547+ expectedErr: nil,
548+ expectRefreshed: false,
549+ expectValidated: false,
550+ expectedLockObtained: false,
349551 }),
350552 Entry("when the session is refreshed by the provider", refreshSessionIfNeededTableInput{
351553 refreshPeriod: 1 * time.Minute,
352554 session: &sessionsapi.SessionState{
353555 RefreshToken: refresh,
354556 CreatedAt: &createdPast,
557+ Lock: &testLock{},
355558 },
356− expectedErr: nil,
357− expectRefreshed: true,
358− expectValidated: true,
559+ expectedErr: nil,
560+ expectRefreshed: true,
561+ expectValidated: true,
562+ expectedLockObtained: true,
359563 }),
360− Entry("when the session is not refreshed by the provider", refreshSessionIfNeededTableInput{
564+ Entry("when obtaining lock failed, but concurrent request refreshed", refreshSessionIfNeededTableInput{
361565 refreshPeriod: 1 * time.Minute,
362566 session: &sessionsapi.SessionState{
363567 RefreshToken: noRefresh,
364568 CreatedAt: &createdPast,
365− ExpiresOn: &createdFuture,
569+ Lock: &testLock{
570+ obtainOnAttempt: 4,
571+ },
366572 },
367− expectedErr: nil,
368− expectRefreshed: true,
369− expectValidated: true,
573+ concurrentSessionRefresh: true,
574+ expectedErr: nil,
575+ expectRefreshed: false,
576+ expectValidated: false,
577+ expectedLockObtained: true,
370578 }),
371− Entry("when the provider doesn't implement refresh but validation succeeds", refreshSessionIfNeededTableInput{
579+ Entry("when obtaining lock failed with a valid session", refreshSessionIfNeededTableInput{
372580 refreshPeriod: 1 * time.Minute,
373581 session: &sessionsapi.SessionState{
374− RefreshToken: notImplemented,
582+ RefreshToken: noRefresh,
583+ CreatedAt: &createdPast,
584+ Lock: &testLock{
585+ obtainError: sessionsapi.ErrLockNotObtained,
586+ },
587+ },
588+ expectedErr: errors.New("timeout obtaining session lock"),
589+ expectRefreshed: false,
590+ expectValidated: false,
591+ expectedLockObtained: true,
592+ }),
593+ Entry("when the session is not refreshed by the provider", refreshSessionIfNeededTableInput{
594+ refreshPeriod: 1 * time.Minute,
595+ session: &sessionsapi.SessionState{
596+ RefreshToken: noRefresh,
375597 CreatedAt: &createdPast,
598+ ExpiresOn: &createdFuture,
599+ Lock: &testLock{},
376600 },
377− expectedErr: nil,
378− expectRefreshed: true,
379− expectValidated: true,
601+ expectedErr: nil,
602+ expectRefreshed: true,
603+ expectValidated: true,
604+ expectedLockObtained: true,
380605 }),
381− Entry("when the provider refresh fails but validation succeeds", refreshSessionIfNeededTableInput{
606+ Entry("when the provider doesn't implement refresh", refreshSessionIfNeededTableInput{
382607 refreshPeriod: 1 * time.Minute,
383608 session: &sessionsapi.SessionState{
384− RefreshToken: "RefreshError",
609+ RefreshToken: notImplemented,
385610 CreatedAt: &createdPast,
611+ Lock: &testLock{},
386612 },
387− expectedErr: nil,
388− expectRefreshed: true,
389− expectValidated: true,
613+ expectedErr: nil,
614+ expectRefreshed: true,
615+ expectValidated: true,
616+ expectedLockObtained: true,
390617 }),
391− Entry("when the session is not refreshed by the provider and validation fails", refreshSessionIfNeededTableInput{
618+ Entry("when the session is not refreshed by the provider", refreshSessionIfNeededTableInput{
392619 refreshPeriod: 1 * time.Minute,
393620 session: &sessionsapi.SessionState{
394621 AccessToken: "Invalid",
395622 RefreshToken: noRefresh,
396623 CreatedAt: &createdPast,
397624 ExpiresOn: &createdFuture,
625+ Lock: &testLock{},
398626 },
399− expectedErr: errors.New("session is invalid"),
400− expectRefreshed: true,
401− expectValidated: true,
627+ expectedErr: errors.New("session is invalid"),
628+ expectRefreshed: true,
629+ expectValidated: true,
630+ expectedLockObtained: true,
402631 }),
403632 )
404633 })