git.iko.soy/typeable/oauth2-proxy
on master help sign in
Use session to lock to protect concurrent refreshes
commite2c7ff6ddd874faef4e0644f0709ce39d7a5440f authorKevin Kreitner <kevin.kreitner@real-digital.de> dateTue 19 Jan 2021, 17:28 (6 years ago) parentdc5d2a5
2 files changed+406−35
compare with master
unifiedsplit
pkg/middleware/stored_session.go+98−11 pkg/middleware/stored_session_test.go+308−24
pkg/middleware/stored_session.go+98−11
@@ -14,6 +14,11 @@ import (
1414 "github.com/oauth2-proxy/oauth2-proxy/v7/providers"
1515)
1616
17+const (
18+ SessionLockExpireTime = 5 * time.Second
19+ SessionLockPeekDelay = 50 * time.Millisecond
20+)
21+
1722// StoredSessionLoaderOptions contains all of the requirements to construct
1823// a stored session loader.
1924// All options must be provided.
@@ -91,13 +96,10 @@ func (s *storedSessionLoader) loadSession(next http.Handler) http.Handler {
9196// that is is valid.
9297func (s *storedSessionLoader) getValidatedSession(rw http.ResponseWriter, req *http.Request) (*sessionsapi.SessionState, error) {
9398 session, err := s.store.Load(req)
94− if err != nil {
99+ if err != nil || session == nil {
100+ // No session was found in the storage or error occurred, nothing more to do
95101 return nil, err
96102 }
97− if session == nil {
98− // No session was found in the storage, nothing more to do
99− return nil, nil
100− }
101103
102104 err = s.refreshSessionIfNeeded(rw, req, session)
103105 if err != nil {
@@ -116,12 +118,21 @@ func (s *storedSessionLoader) refreshSessionIfNeeded(rw http.ResponseWriter, req
116118 return nil
117119 }
118120
119− logger.Printf("Refreshing session - User: %s; SessionAge: %s", session.User, session.Age())
120− err := s.refreshSession(rw, req, session)
121+ wasRefreshed, err := s.checkForConcurrentRefresh(session, req)
121122 if err != nil {
122− // If a preemptive refresh fails, we still keep the session
123− // if validateSession succeeds.
124− logger.Errorf("Unable to refresh session: %v", err)
123+ return err
124+ }
125+
126+ // If session was already refreshed via a concurrent request locked skip refreshing,
127+ // because the refreshed session is already loaded from storage
128+ if !wasRefreshed {
129+ logger.Printf("Refreshing session - User: %s; SessionAge: %s", session.User, session.Age())
130+ err = s.refreshSession(rw, req, session)
131+ if err != nil {
132+ // If a preemptive refresh fails, we still keep the session
133+ // if validateSession succeeds.
134+ logger.Errorf("Unable to refresh session: %v", err)
135+ }
125136 }
126137
127138 // Validate all sessions after any Redeem/Refresh operation (fail or success)
@@ -131,6 +142,18 @@ func (s *storedSessionLoader) refreshSessionIfNeeded(rw http.ResponseWriter, req
131142// refreshSession attempts to refresh the session with the provider
132143// and will save the session if it was updated.
133144func (s *storedSessionLoader) refreshSession(rw http.ResponseWriter, req *http.Request, session *sessionsapi.SessionState) error {
145+ err := session.ObtainLock(req.Context(), SessionLockExpireTime)
146+ if err != nil {
147+ logger.Errorf("Unable to obtain lock: %v", err)
148+ return s.handleObtainLockError(req, session)
149+ }
150+ defer func() {
151+ err = session.ReleaseLock(req.Context())
152+ if err != nil {
153+ logger.Errorf("unable to release lock: %v", err)
154+ }
155+ }()
156+
134157 refreshed, err := s.sessionRefresher(req.Context(), session)
135158 if err != nil && !errors.Is(err, providers.ErrNotImplemented) {
136159 return fmt.Errorf("error refreshing tokens: %v", err)
@@ -159,11 +182,75 @@ func (s *storedSessionLoader) refreshSession(rw http.ResponseWriter, req *http.R
159182 err = s.store.Save(rw, req, session)
160183 if err != nil {
161184 logger.PrintAuthf(session.Email, req, logger.AuthError, "error saving session: %v", err)
162− return fmt.Errorf("error saving session: %v", err)
185+ err = fmt.Errorf("error saving session: %v", err)
186+ }
187+ return err
188+}
189+
190+func (s *storedSessionLoader) handleObtainLockError(req *http.Request, session *sessionsapi.SessionState) error {
191+ wasRefreshed, err := s.checkForConcurrentRefresh(session, req)
192+ if err != nil {
193+ logger.Errorf("Unable to wait for obtained lock: %v", err)
194+ return err
195+ }
196+ if !wasRefreshed {
197+ return errors.New("unable to obtain lock and session was also not refreshed via concurrent request")
198+ }
199+ return nil
200+}
201+
202+func (s *storedSessionLoader) updateSessionFromStore(req *http.Request, session *sessionsapi.SessionState) error {
203+ sessionStored, err := s.store.Load(req)
204+ if err != nil {
205+ return fmt.Errorf("unable to load updated session from store: %v", err)
206+ }
207+
208+ if sessionStored == nil {
209+ return fmt.Errorf("no session available to udpate from store")
163210 }
211+ *session = *sessionStored
212+
164213 return nil
165214}
166215
216+func (s *storedSessionLoader) waitForPossibleSessionLock(session *sessionsapi.SessionState, req *http.Request) (bool, error) {
217+ var wasLocked bool
218+ isLocked, err := session.PeekLock(req.Context())
219+ for isLocked {
220+ wasLocked = true
221+ // delay next peek lock
222+ time.Sleep(SessionLockPeekDelay)
223+ isLocked, err = session.PeekLock(req.Context())
224+ }
225+
226+ if err != nil {
227+ return false, err
228+ }
229+
230+ return wasLocked, nil
231+}
232+
233+// checkForConcurrentRefresh returns true if the session is already refreshed via a concurrent request.
234+func (s *storedSessionLoader) checkForConcurrentRefresh(session *sessionsapi.SessionState, req *http.Request) (bool, error) {
235+ wasLocked, err := s.waitForPossibleSessionLock(session, req)
236+ if err != nil {
237+ return false, err
238+ }
239+
240+ refreshed := false
241+ if wasLocked {
242+ logger.Printf("Update session from store instead of refreshing")
243+ err = s.updateSessionFromStore(req, session)
244+ if err != nil {
245+ logger.Errorf("Unable to update session from store: %v", err)
246+ return false, err
247+ }
248+ refreshed = true
249+ }
250+
251+ return refreshed, nil
252+}
253+
167254// validateSession checks whether the session has expired and performs
168255// provider validation on the session.
169256// An error implies the session is not longer valid.
pkg/middleware/stored_session_test.go+308−24
@@ -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,104 @@ import (
1718 . "github.com/onsi/gomega"
1819)
1920
21+type TestLock struct {
22+ Locked bool
23+ WasObtained bool
24+ WasRefreshed bool
25+ WasReleased bool
26+ PeekedCount int
27+ LockedOnPeekCount int
28+ ObtainError error
29+ PeekError error
30+ RefreshError error
31+ ReleaseError error
32+}
33+
34+func (l *TestLock) Obtain(_ context.Context, _ time.Duration) error {
35+ if l.ObtainError != nil {
36+ return l.ObtainError
37+ }
38+ l.Locked = true
39+ l.WasObtained = true
40+ return nil
41+}
42+
43+func (l *TestLock) Peek(_ context.Context) (bool, error) {
44+ if l.PeekError != nil {
45+ return false, l.PeekError
46+ }
47+ locked := l.Locked
48+ l.Locked = false
49+ l.PeekedCount++
50+ // mainly used to test case when peek initially returns false,
51+ // but when trying to obtain lock, it returns true.
52+ if l.LockedOnPeekCount == l.PeekedCount {
53+ return true, nil
54+ }
55+ return locked, nil
56+}
57+
58+func (l *TestLock) Refresh(_ context.Context, _ time.Duration) error {
59+ if l.RefreshError != nil {
60+ return l.ReleaseError
61+ }
62+ l.WasRefreshed = true
63+ return nil
64+}
65+
66+func (l *TestLock) Release(_ context.Context) error {
67+ if l.ReleaseError != nil {
68+ return l.ReleaseError
69+ }
70+ l.Locked = false
71+ l.WasReleased = true
72+ return nil
73+}
74+
75+type LockConc struct {
76+ mu sync.Mutex
77+ lock bool
78+ disablePeek bool
79+}
80+
81+func (l *LockConc) Obtain(_ context.Context, _ time.Duration) error {
82+ l.mu.Lock()
83+ if l.lock {
84+ l.mu.Unlock()
85+ return sessionsapi.ErrLockNotObtained
86+ }
87+ l.lock = true
88+ l.mu.Unlock()
89+ return nil
90+}
91+
92+func (l *LockConc) Peek(_ context.Context) (bool, error) {
93+ var response bool
94+ l.mu.Lock()
95+ if l.disablePeek {
96+ response = false
97+ } else {
98+ response = l.lock
99+ }
100+ l.mu.Unlock()
101+ return response, nil
102+}
103+
104+func (l *LockConc) Refresh(_ context.Context, _ time.Duration) error {
105+ return nil
106+}
107+
108+func (l *LockConc) Release(_ context.Context) error {
109+ l.mu.Lock()
110+ l.lock = false
111+ l.mu.Unlock()
112+ return nil
113+}
114+
20115var _ = Describe("Stored Session Suite", func() {
21116 const (
22117 refresh = "Refresh"
118+ refreshed = "Refreshed"
23119 noRefresh = "NoRefresh"
24120 notImplemented = "NotImplemented"
25121 )
@@ -34,7 +130,7 @@ var _ = Describe("Stored Session Suite", func() {
34130 var defaultRefreshFunc = func(_ context.Context, ss *sessionsapi.SessionState) (bool, error) {
35131 switch ss.RefreshToken {
36132 case refresh:
37− ss.RefreshToken = "Refreshed"
133+ ss.RefreshToken = refreshed
38134 return true, nil
39135 case noRefresh:
40136 return false, nil
@@ -181,6 +277,7 @@ var _ = Describe("Stored Session Suite", func() {
181277 RefreshToken: noRefresh,
182278 CreatedAt: &createdPast,
183279 ExpiresOn: &createdFuture,
280+ Lock: &sessionsapi.NoOpLock{},
184281 },
185282 store: defaultSessionStore,
186283 refreshPeriod: 1 * time.Minute,
@@ -222,6 +319,7 @@ var _ = Describe("Stored Session Suite", func() {
222319 RefreshToken: "Refreshed",
223320 CreatedAt: &now,
224321 ExpiresOn: &createdFuture,
322+ Lock: &sessionsapi.NoOpLock{},
225323 },
226324 store: defaultSessionStore,
227325 refreshPeriod: 1 * time.Minute,
@@ -237,6 +335,7 @@ var _ = Describe("Stored Session Suite", func() {
237335 RefreshToken: "RefreshError",
238336 CreatedAt: &createdPast,
239337 ExpiresOn: &createdFuture,
338+ Lock: &sessionsapi.NoOpLock{},
240339 },
241340 store: defaultSessionStore,
242341 refreshPeriod: 1 * time.Minute,
@@ -266,15 +365,109 @@ var _ = Describe("Stored Session Suite", func() {
266365 validateSession: defaultValidateFunc,
267366 }),
268367 )
368+
369+ type storedSessionLoaderConcurrentTableInput struct {
370+ existingSession *sessionsapi.SessionState
371+ refreshPeriod time.Duration
372+ numConcReqs int
373+ }
374+
375+ DescribeTable("when serving concurrent requests",
376+ func(in storedSessionLoaderConcurrentTableInput) {
377+ lockConc := &LockConc{}
378+
379+ refreshedChan := make(chan bool, in.numConcReqs)
380+ for i := 0; i < in.numConcReqs; i++ {
381+ go func(refreshedChan chan bool, lockConc sessionsapi.Lock) {
382+ existingSession := *in.existingSession // deep copy existingSession state
383+ existingSession.Lock = lockConc
384+ store := &fakeSessionStore{
385+ LoadFunc: func(req *http.Request) (*sessionsapi.SessionState, error) {
386+ return &existingSession, nil
387+ },
388+ SaveFunc: func(http.ResponseWriter, *http.Request, *sessionsapi.SessionState) error {
389+ return nil
390+ },
391+ }
392+
393+ scope := &middlewareapi.RequestScope{
394+ Session: nil,
395+ }
396+
397+ // Set up the request with the request header and a request scope
398+ req := httptest.NewRequest("", "/", nil)
399+ req = middlewareapi.AddRequestScope(req, scope)
400+
401+ rw := httptest.NewRecorder()
402+
403+ sessionRefreshed := false
404+ opts := &StoredSessionLoaderOptions{
405+ SessionStore: store,
406+ RefreshPeriod: in.refreshPeriod,
407+ RefreshSession: func(ctx context.Context, s *sessionsapi.SessionState) (bool, error) {
408+ time.Sleep(10 * time.Millisecond)
409+ sessionRefreshed = true
410+ return true, nil
411+ },
412+ ValidateSession: func(context.Context, *sessionsapi.SessionState) bool {
413+ return true
414+ },
415+ }
416+
417+ handler := NewStoredSessionLoader(opts)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
418+ handler.ServeHTTP(rw, req)
419+
420+ refreshedChan <- sessionRefreshed
421+ }(refreshedChan, lockConc)
422+ }
423+ var refreshedSlice []bool
424+ for i := 0; i < in.numConcReqs; i++ {
425+ refreshedSlice = append(refreshedSlice, <-refreshedChan)
426+ }
427+ sessionRefreshedCount := 0
428+ for _, sessionRefreshed := range refreshedSlice {
429+ if sessionRefreshed {
430+ sessionRefreshedCount++
431+ }
432+ }
433+ Expect(sessionRefreshedCount).To(Equal(1))
434+ },
435+ Entry("with two concurrent requests", storedSessionLoaderConcurrentTableInput{
436+ existingSession: &sessionsapi.SessionState{
437+ RefreshToken: refresh,
438+ CreatedAt: &createdPast,
439+ },
440+ numConcReqs: 2,
441+ refreshPeriod: 1 * time.Minute,
442+ }),
443+ Entry("with 5 concurrent requests", storedSessionLoaderConcurrentTableInput{
444+ existingSession: &sessionsapi.SessionState{
445+ RefreshToken: refresh,
446+ CreatedAt: &createdPast,
447+ },
448+ numConcReqs: 5,
449+ refreshPeriod: 1 * time.Minute,
450+ }),
451+ Entry("with one request", storedSessionLoaderConcurrentTableInput{
452+ existingSession: &sessionsapi.SessionState{
453+ RefreshToken: refresh,
454+ CreatedAt: &createdPast,
455+ },
456+ numConcReqs: 1,
457+ refreshPeriod: 1 * time.Minute,
458+ }),
459+ )
269460 })
270461
271462 Context("refreshSessionIfNeeded", func() {
272463 type refreshSessionIfNeededTableInput struct {
273− refreshPeriod time.Duration
274− session *sessionsapi.SessionState
275− expectedErr error
276− expectRefreshed bool
277− expectValidated bool
464+ refreshPeriod time.Duration
465+ sessionStored bool
466+ session *sessionsapi.SessionState
467+ expectedErr error
468+ expectRefreshed bool
469+ expectValidated bool
470+ expectedLockState TestLock
278471 }
279472
280473 createdPast := time.Now().Add(-5 * time.Minute)
@@ -285,9 +478,18 @@ var _ = Describe("Stored Session Suite", func() {
285478 refreshed := false
286479 validated := false
287480
481+ store := &fakeSessionStore{}
482+ if in.sessionStored {
483+ store = &fakeSessionStore{
484+ LoadFunc: func(req *http.Request) (*sessionsapi.SessionState, error) {
485+ return in.session, nil
486+ },
487+ }
488+ }
489+
288490 s := &storedSessionLoader{
289491 refreshPeriod: in.refreshPeriod,
290− store: &fakeSessionStore{},
492+ store: store,
291493 sessionRefresher: func(_ context.Context, ss *sessionsapi.SessionState) (bool, error) {
292494 refreshed = true
293495 switch ss.RefreshToken {
@@ -316,89 +518,171 @@ var _ = Describe("Stored Session Suite", func() {
316518 }
317519 Expect(refreshed).To(Equal(in.expectRefreshed))
318520 Expect(validated).To(Equal(in.expectValidated))
521+ testLock, ok := in.session.Lock.(*TestLock)
522+ Expect(ok).To(Equal(true))
523+
524+ Expect(testLock).To(Equal(&in.expectedLockState))
319525 },
320526 Entry("when the refresh period is 0, and the session does not need refreshing", refreshSessionIfNeededTableInput{
321527 refreshPeriod: time.Duration(0),
322528 session: &sessionsapi.SessionState{
323529 RefreshToken: refresh,
324530 CreatedAt: &createdFuture,
531+ Lock: &TestLock{},
325532 },
326− expectedErr: nil,
327− expectRefreshed: false,
328− expectValidated: false,
533+ expectedErr: nil,
534+ expectRefreshed: false,
535+ expectValidated: false,
536+ expectedLockState: TestLock{},
329537 }),
330538 Entry("when the refresh period is 0, and the session needs refreshing", refreshSessionIfNeededTableInput{
331539 refreshPeriod: time.Duration(0),
332540 session: &sessionsapi.SessionState{
333541 RefreshToken: refresh,
334542 CreatedAt: &createdPast,
543+ Lock: &TestLock{},
335544 },
336− expectedErr: nil,
337− expectRefreshed: false,
338− expectValidated: false,
545+ expectedErr: nil,
546+ expectRefreshed: false,
547+ expectValidated: false,
548+ expectedLockState: TestLock{},
339549 }),
340550 Entry("when the session does not need refreshing", refreshSessionIfNeededTableInput{
341551 refreshPeriod: 1 * time.Minute,
342552 session: &sessionsapi.SessionState{
343553 RefreshToken: refresh,
344554 CreatedAt: &createdFuture,
555+ Lock: &TestLock{},
345556 },
346− expectedErr: nil,
347− expectRefreshed: false,
348− expectValidated: false,
557+ expectedErr: nil,
558+ expectRefreshed: false,
559+ expectValidated: false,
560+ expectedLockState: TestLock{},
349561 }),
350562 Entry("when the session is refreshed by the provider", refreshSessionIfNeededTableInput{
351563 refreshPeriod: 1 * time.Minute,
352564 session: &sessionsapi.SessionState{
353565 RefreshToken: refresh,
354566 CreatedAt: &createdPast,
567+ Lock: &TestLock{},
355568 },
356569 expectedErr: nil,
357570 expectRefreshed: true,
358571 expectValidated: true,
572+ expectedLockState: TestLock{
573+ Locked: false,
574+ WasObtained: true,
575+ WasReleased: true,
576+ PeekedCount: 1,
577+ },
359578 }),
360− Entry("when the session is not refreshed by the provider", refreshSessionIfNeededTableInput{
579+ Entry("when the session is locked and instead loaded from storage", refreshSessionIfNeededTableInput{
361580 refreshPeriod: 1 * time.Minute,
362581 session: &sessionsapi.SessionState{
363582 RefreshToken: noRefresh,
364583 CreatedAt: &createdPast,
365− ExpiresOn: &createdFuture,
584+ Lock: &TestLock{
585+ Locked: true,
586+ },
366587 },
588+ sessionStored: true,
367589 expectedErr: nil,
368− expectRefreshed: true,
590+ expectRefreshed: false,
369591 expectValidated: true,
592+ expectedLockState: TestLock{
593+ Locked: false,
594+ PeekedCount: 2,
595+ },
370596 }),
371− Entry("when the provider doesn't implement refresh but validation succeeds", refreshSessionIfNeededTableInput{
597+ Entry("when obtaining lock failed, but concurrent request refreshed", refreshSessionIfNeededTableInput{
372598 refreshPeriod: 1 * time.Minute,
373599 session: &sessionsapi.SessionState{
374− RefreshToken: notImplemented,
600+ RefreshToken: noRefresh,
375601 CreatedAt: &createdPast,
602+ Lock: &TestLock{
603+ ObtainError: errors.New("not able to obtain lock"),
604+ LockedOnPeekCount: 2,
605+ },
606+ },
607+ expectedErr: nil,
608+ expectRefreshed: false,
609+ expectValidated: true,
610+ expectedLockState: TestLock{
611+ PeekedCount: 3,
612+ LockedOnPeekCount: 2,
613+ ObtainError: errors.New("not able to obtain lock"),
614+ },
615+ }),
616+ Entry("when obtaining lock failed", refreshSessionIfNeededTableInput{
617+ refreshPeriod: 1 * time.Minute,
618+ session: &sessionsapi.SessionState{
619+ RefreshToken: noRefresh,
620+ CreatedAt: &createdPast,
621+ Lock: &TestLock{
622+ ObtainError: errors.New("not able to obtain lock"),
623+ },
624+ },
625+ expectedErr: nil,
626+ expectRefreshed: false,
627+ expectValidated: true,
628+ expectedLockState: TestLock{
629+ PeekedCount: 2,
630+ ObtainError: errors.New("not able to obtain lock"),
631+ },
632+ }),
633+ Entry("when the session is not refreshed by the provider", refreshSessionIfNeededTableInput{
634+ refreshPeriod: 1 * time.Minute,
635+ session: &sessionsapi.SessionState{
636+ RefreshToken: noRefresh,
637+ CreatedAt: &createdPast,
638+ ExpiresOn: &createdFuture,
639+ Lock: &TestLock{},
376640 },
377641 expectedErr: nil,
378642 expectRefreshed: true,
379643 expectValidated: true,
644+ expectedLockState: TestLock{
645+ Locked: false,
646+ WasObtained: true,
647+ WasReleased: true,
648+ PeekedCount: 1,
649+ },
380650 }),
381− Entry("when the provider refresh fails but validation succeeds", refreshSessionIfNeededTableInput{
651+ Entry("when the provider doesn't implement refresh", refreshSessionIfNeededTableInput{
382652 refreshPeriod: 1 * time.Minute,
383653 session: &sessionsapi.SessionState{
384− RefreshToken: "RefreshError",
654+ RefreshToken: notImplemented,
385655 CreatedAt: &createdPast,
656+ Lock: &TestLock{},
386657 },
387658 expectedErr: nil,
388659 expectRefreshed: true,
389660 expectValidated: true,
661+ expectedLockState: TestLock{
662+ Locked: false,
663+ WasObtained: true,
664+ WasReleased: true,
665+ PeekedCount: 1,
666+ },
390667 }),
391− Entry("when the session is not refreshed by the provider and validation fails", refreshSessionIfNeededTableInput{
668+ Entry("when the session is not refreshed by the provider", refreshSessionIfNeededTableInput{
392669 refreshPeriod: 1 * time.Minute,
393670 session: &sessionsapi.SessionState{
394671 AccessToken: "Invalid",
395672 RefreshToken: noRefresh,
396673 CreatedAt: &createdPast,
397674 ExpiresOn: &createdFuture,
675+ Lock: &TestLock{},
398676 },
399677 expectedErr: errors.New("session is invalid"),
400678 expectRefreshed: true,
401679 expectValidated: true,
680+ expectedLockState: TestLock{
681+ Locked: false,
682+ WasObtained: true,
683+ WasReleased: true,
684+ PeekedCount: 1,
685+ },
402686 }),
403687 )
404688 })