Mercurial > gemma

comparison pkg/auth/store.go @ 495:5c08afd15ce7 metamorph-for-all

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Give file containing the session store a more suited name.
author Sascha L. Teichmann <sascha.teichmann@intevation.de>
date Fri, 24 Aug 2018 14:28:05 +0200
parents pkg/auth/pool.go@8a0737aa6ab6
children 22e1bf563a04
comparison
equal deleted inserted replaced
493:8a0737aa6ab6 495:5c08afd15ce7
1 package auth
2
3 import (
4 "bytes"
5 "log"
6 "time"
7
8 bolt "github.com/coreos/bbolt"
9 )
10
11 // Sessions is the global connection pool.
12 var Sessions *SessionStore
13
14 type SessionStore struct {
15 storage *bolt.DB
16 conns map[string]*Connection
17 cmds chan func(*SessionStore)
18 }
19
20 var sessionsBucket = []byte("sessions")
21
22 func NewSessionStore(filename string) (*SessionStore, error) {
23
24 pcp := &SessionStore{
25 conns: make(map[string]*Connection),
26 cmds: make(chan func(*SessionStore)),
27 }
28 if err := pcp.openStorage(filename); err != nil {
29 return nil, err
30 }
31 go pcp.run()
32 return pcp, nil
33 }
34
35 // openStorage opens a storage file.
36 func (pcp *SessionStore) openStorage(filename string) error {
37
38 // No file, nothing to restore/persist.
39 if filename == "" {
40 return nil
41 }
42
43 db, err := bolt.Open(filename, 0600, nil)
44 if err != nil {
45 return err
46 }
47
48 err = db.Update(func(tx *bolt.Tx) error {
49 b, err := tx.CreateBucketIfNotExists(sessionsBucket)
50 if err != nil {
51 return err
52 }
53
54 // pre-load sessions
55 c := b.Cursor()
56
57 for k, v := c.First(); k != nil; k, v = c.Next() {
58 var conn Connection
59 if err := conn.deserialize(bytes.NewReader(v)); err != nil {
60 return err
61 }
62 pcp.conns[string(k)] = &conn
63 }
64
65 return nil
66 })
67
68 if err != nil {
69 db.Close()
70 return err
71 }
72
73 pcp.storage = db
74 return nil
75 }
76
77 func (pcp *SessionStore) run() {
78 for {
79 select {
80 case cmd := <-pcp.cmds:
81 cmd(pcp)
82 case <-time.After(time.Minute * 5):
83 pcp.cleanToken()
84 }
85 }
86 }
87
88 func (pcp *SessionStore) cleanToken() {
89 now := time.Now()
90 for token, con := range pcp.conns {
91 expires := time.Unix(con.session.ExpiresAt, 0)
92 if expires.Before(now) {
93 delete(pcp.conns, token)
94 pcp.remove(token)
95 }
96 }
97 }
98
99 func (pcp *SessionStore) remove(token string) {
100 if pcp.storage == nil {
101 return
102 }
103 err := pcp.storage.Update(func(tx *bolt.Tx) error {
104 b := tx.Bucket(sessionsBucket)
105 return b.Delete([]byte(token))
106 })
107 if err != nil {
108 log.Printf("error: %v\n", err)
109 }
110 }
111
112 func (pcp *SessionStore) Delete(token string) bool {
113 res := make(chan bool)
114 pcp.cmds <- func(pcp *SessionStore) {
115 if _, found := pcp.conns[token]; !found {
116 res <- false
117 return
118 }
119 delete(pcp.conns, token)
120 pcp.remove(token)
121 res <- true
122 }
123 return <-res
124 }
125
126 func (pcp *SessionStore) store(token string, con *Connection) {
127 if pcp.storage == nil {
128 return
129 }
130 err := pcp.storage.Update(func(tx *bolt.Tx) error {
131 b := tx.Bucket(sessionsBucket)
132 var buf bytes.Buffer
133 if err := con.serialize(&buf); err != nil {
134 return err
135 }
136 return b.Put([]byte(token), buf.Bytes())
137 })
138 if err != nil {
139 log.Printf("error: %v\n", err)
140 }
141 }
142
143 func (pcp *SessionStore) Add(token string, session *Session) *Connection {
144 res := make(chan *Connection)
145
146 pcp.cmds <- func(cp *SessionStore) {
147 con := pcp.conns[token]
148 if con == nil {
149 con = &Connection{}
150 pcp.conns[token] = con
151 }
152 con.set(session)
153 pcp.store(token, con)
154 res <- con
155 }
156
157 con := <-res
158 return con
159 }
160
161 func (pcp *SessionStore) Renew(token string) (string, error) {
162
163 type result struct {
164 newToken string
165 err error
166 }
167
168 resCh := make(chan result)
169
170 pcp.cmds <- func(cp *SessionStore) {
171 con := pcp.conns[token]
172 if con == nil {
173 resCh <- result{err: ErrNoSuchToken}
174 } else {
175 delete(pcp.conns, token)
176 pcp.remove(token)
177 newToken := GenerateSessionKey()
178 // TODO: Ensure that this is not racy!
179 con.session.ExpiresAt = time.Now().Add(maxTokenValid).Unix()
180 pcp.conns[newToken] = con
181 pcp.store(newToken, con)
182 resCh <- result{newToken: newToken}
183 }
184 }
185
186 r := <-resCh
187 return r.newToken, r.err
188 }
189
190 func (pcp *SessionStore) Do(token string) (*Session, error) {
191
192 type result struct {
193 session *Session
194 err error
195 }
196
197 res := make(chan result)
198
199 pcp.cmds <- func(pcp *SessionStore) {
200 con := pcp.conns[token]
201 if con == nil {
202 res <- result{err: ErrNoSuchToken}
203 return
204 }
205 con.touch()
206 pcp.store(token, con)
207
208 res <- result{session: con.session}
209 }
210
211 r := <-res
212
213 if r.err != nil {
214 return nil, r.err
215 }
216
217 return r.session, nil
218 }
219
220 func (pcp *SessionStore) Session(token string) *Session {
221 res := make(chan *Session)
222 pcp.cmds <- func(pcp *SessionStore) {
223 con := pcp.conns[token]
224 if con == nil {
225 res <- nil
226 } else {
227 con.touch()
228 pcp.store(token, con)
229 res <- con.session
230 }
231 }
232 return <-res
233 }
234
235 func (pcp *SessionStore) Logout(user string) {
236 pcp.cmds <- func(pcp *SessionStore) {
237 for token, con := range pcp.conns {
238 if con.session.User == user {
239 delete(pcp.conns, token)
240 pcp.remove(token)
241 }
242 }
243 }
244 }
245
246 func (pcp *SessionStore) Shutdown() error {
247 if db := pcp.storage; db != nil {
248 log.Println("info: shutdown persistent connection pool.")
249 pcp.storage = nil
250 return db.Close()
251 }
252 log.Println("info: shutdown in-memory connection pool.")
253 return nil
254 }