1
package server
2
 
3
import (
4
	"bytes"
5
	"context"
6
	"errors"
7
	"time"
8
 
9
	"go.miragespace.co/specter/util"
10
)
11
 
12
func (s *Server) maintainSession(ctx context.Context, sess *session) {
13
	ctx, cancel := context.WithCancel(ctx)
14
	defer cancel()
15
	sess.mu.Lock()
16
	if sess.state != active || sess.mode != delegated || connectionDone(sess.conn) {
17
		sess.mu.Unlock()
18
		return
19
	}
20
	sess.revalidateCancel = cancel
21
	sess.revalidateDone = make(chan struct{})
22
	defer close(sess.revalidateDone)
23
	until := sess.authorizedUntil
24
	sess.mu.Unlock()
25
	// Shutdown and expiry must close streams even if storage ignores cancellation.
26
	context.AfterFunc(ctx, func() { sess.close("session maintenance stopped") })
27
	expiry := time.AfterFunc(time.Until(until), func() {
28
		sess.mu.Lock()
29
		expired := sess.state == active && !time.Now().Before(sess.authorizedUntil)
30
		sess.mu.Unlock()
31
		if expired {
32
			sess.close("authorization grace expired")
33
		}
34
	})
35
	defer expiry.Stop()
36
	// Stagger the first check; subsequent checks stay thirty seconds apart.
37
	next := time.NewTimer(util.RandomTimeRange(revalidateInterval))
38
	defer next.Stop()
39
	for {
40
		select {
41
		case <-ctx.Done():
42
			return
43
		case <-sess.conn.Done():
44
			return
45
		case <-next.C:
46
			next.Reset(revalidateInterval)
47
			s.revalidateSession(ctx, sess)
48
			sess.mu.Lock()
49
			until, state := sess.authorizedUntil, sess.state
50
			sess.mu.Unlock()
51
			if state != active {
52
				return
53
			}
54
			expiry.Reset(time.Until(until))
55
		}
56
	}
57
}
58
 
59
func (s *Server) revalidateSession(ctx context.Context, sess *session) {
60
	if ctx.Err() != nil {
61
		return
62
	}
63
	start := time.Now()
64
	sess.mu.Lock()
65
	if sess.state != active {
66
		sess.mu.Unlock()
67
		return
68
	}
69
	expired := !start.Before(sess.authorizedUntil)
70
	sess.mu.Unlock()
71
	if expired {
72
		sess.close("authorization grace expired")
73
		return
74
	}
75
	ctx, cancel := context.WithTimeout(ctx, lookupTimeout)
76
	defer cancel()
77
	rec, err := s.readDelegation(ctx, sess.grantID)
78
	if errors.Is(err, errDelegationNotFound) || errors.Is(err, errDelegationInvalid) {
79
		sess.close(err.Error())
80
		return
81
	}
82
	if err != nil || ctx.Err() != nil {
83
		return
84
	}
85
	expiry := time.Time{}
86
	if rec.GetExpiresAt() != 0 {
87
		expiry = time.Unix(rec.GetExpiresAt(), 0)
88
	}
89
	if rec.GetHostname() != sess.hostname || !bytes.Equal(rec.GetOwner().GetToken(), sess.owner.GetToken()) || !expiry.Equal(sess.grantExpiry) || (!expiry.IsZero() && !time.Now().Before(expiry)) {
90
		sess.close("grant authority changed or expired")
91
		return
92
	}
93
	err = s.verifyDelegatedAuthority(ctx, rec)
94
	if errors.Is(err, errDelegationNotOwned) {
95
		sess.close(err.Error())
96
		return
97
	}
98
	if err != nil {
99
		return
100
	}
101
	if ctx.Err() == nil {
102
		sess.extend(start, expiry)
103
	}
104
}