1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "crypto/rand"
8 "errors"
9 "fmt"
10
11 kv "bnema.dev/ermite-kv"
12 "github.com/bnema/zerowrap"
13 "google.golang.org/protobuf/proto"
14
15 storagepb "bnema.dev/ermite/gen/ermite/storage"
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18)
19
20const (
21 sessionKeyKey = "session/key"
22 sessionEpochKey = "session/epoch"
23
24 // sessionKeyLen is the size of the HS256 secret, in bytes.
25 sessionKeyLen = 32
26)
27
28// Sessions is the ermite-kv backed ports.SessionStore. It shares the database
29// of the Store it comes from.
30type Sessions struct {
31 db *kv.DB
32}
33
34var _ ports.SessionStore = (*Sessions)(nil)
35
36// Sessions returns the session store over the same database. It is valid
37// until the Store is closed.
38func (s *Store) Sessions() *Sessions { return &Sessions{db: s.db} }
39
40// Key reads session/key and generates it when absent. The key is never
41// logged.
42func (s *Sessions) Key(ctx context.Context) ([]byte, error) {
43 for range updateAttempts {
44 if item, ok := s.db.Get(ctx, []byte(sessionKeyKey)); ok {
45 var m storagepb.SessionKey
46 if err := proto.Unmarshal(item.Value, &m); err != nil {
47 return nil, fmt.Errorf("decode session key: %w", err)
48 }
49 if len(m.GetKey()) != sessionKeyLen {
50 return nil, fmt.Errorf("decode session key: %d bytes, want %d", len(m.GetKey()), sessionKeyLen)
51 }
52 return append([]byte(nil), m.GetKey()...), nil
53 }
54 key := make([]byte, sessionKeyLen)
55 if _, err := rand.Read(key); err != nil {
56 return nil, fmt.Errorf("generate session key: %w", err)
57 }
58 var b kv.Batch
59 b.RequireAbsent([]byte(sessionKeyKey))
60 b.Put([]byte(sessionKeyKey), mustMarshal(&storagepb.SessionKey{Key: key}))
61 _, err := s.db.Apply(ctx, &b)
62 if errors.Is(err, kv.ErrConflict) {
63 continue // written meanwhile: read it
64 }
65 if err != nil {
66 return nil, fmt.Errorf("store session key: %w", err)
67 }
68 log := zerowrap.FromCtx(ctx)
69 log.Info().Msg("session key generated")
70 return key, nil
71 }
72 return nil, fmt.Errorf("store session key: %w", domain.ErrConflict)
73}
74
75// Epoch reads session/epoch; a missing key is epoch 0.
76func (s *Sessions) Epoch(ctx context.Context) (uint64, error) {
77 e, _, err := s.readEpoch(ctx)
78 return e, err
79}
80
81func (s *Sessions) readEpoch(ctx context.Context) (epoch, version uint64, err error) {
82 item, ok := s.db.Get(ctx, []byte(sessionEpochKey))
83 if !ok {
84 return 0, 0, nil
85 }
86 var m storagepb.SessionEpoch
87 if err := proto.Unmarshal(item.Value, &m); err != nil {
88 return 0, 0, fmt.Errorf("decode session epoch: %w", err)
89 }
90 return m.GetEpoch(), item.Version, nil
91}
92
93// BumpEpoch increments session/epoch with a version check, so that two
94// concurrent bumps give two different values.
95func (s *Sessions) BumpEpoch(ctx context.Context) (uint64, error) {
96 for range updateAttempts {
97 cur, version, err := s.readEpoch(ctx)
98 if err != nil {
99 return 0, err
100 }
101 var b kv.Batch
102 if version == 0 {
103 b.RequireAbsent([]byte(sessionEpochKey))
104 } else {
105 b.RequireVersion([]byte(sessionEpochKey), version)
106 }
107 b.Put([]byte(sessionEpochKey), mustMarshal(&storagepb.SessionEpoch{Epoch: cur + 1}))
108 _, err = s.db.Apply(ctx, &b)
109 if errors.Is(err, kv.ErrConflict) {
110 continue
111 }
112 if err != nil {
113 return 0, fmt.Errorf("bump session epoch: %w", err)
114 }
115 return cur + 1, nil
116 }
117 return 0, fmt.Errorf("bump session epoch: %w", domain.ErrConflict)
118}