1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "crypto/sha256"
8 "encoding/hex"
9 "fmt"
10 "strings"
11
12 kv "bnema.dev/ermite-kv"
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
20// Key layout of the job repository tokens and of the action resolutions,
21// written by CI.Apply under the lock of their run:
22//
23// ghtoken/<id> CIRepoToken
24// ghtokenhash/<hash> token id
25// ghtokenjob/<run:020d>/<attempt:010d>/<job>/<id> the id: the tokens of a job
26// ghtokenrepo/<repo>/<id> the job key: the tokens that read a repository
27// actres/<run:020d>/<attempt:010d>/<sha256(key)> CIActionResolution
28//
29// Stored tokens use token/ and tokenhash/: the two families never meet, so a
30// job token is never listed, revoked or managed as a stored token.
31const (
32 repoTokenPrefix = "ghtoken/"
33 repoTokenHashPrefix = "ghtokenhash/"
34 repoTokenJobPrefix = "ghtokenjob/"
35 repoTokenRepoPrefix = "ghtokenrepo/"
36 resolutionPrefix = "actres/"
37)
38
39var _ ports.CIRepoTokenStore = (*CI)(nil)
40
41func repoTokenKey(id string) []byte { return []byte(repoTokenPrefix + id) }
42func repoTokenHashKey(h domain.TokenHash) []byte {
43 return []byte(repoTokenHashPrefix + h.String())
44}
45func repoTokenJobPrefixOf(runID int64, attempt int32, jobID string) string {
46 return fmt.Sprintf("%s%020d/%010d/%s/", repoTokenJobPrefix, runID, attempt, jobID)
47}
48func repoTokenJobKey(t domain.CIRepoToken) []byte {
49 return []byte(repoTokenJobPrefixOf(t.RunID, t.RunAttempt, t.JobID) + t.ID)
50}
51func repoTokenRepoPrefixOf(repo domain.RepoName) string {
52 return repoTokenRepoPrefix + repo.String() + "/"
53}
54func repoTokenRepoKey(repo domain.RepoName, id string) []byte {
55 return []byte(repoTokenRepoPrefixOf(repo) + id)
56}
57func resolutionKey(runID int64, attempt int32, key string) []byte {
58 sum := sha256.Sum256([]byte(key))
59 return []byte(fmt.Sprintf("%s%020d/%010d/%s", resolutionPrefix, runID, attempt, hex.EncodeToString(sum[:])))
60}
61
62func encodeRepoToken(t domain.CIRepoToken) []byte {
63 return mustMarshal(&storagepb.CIRepoToken{
64 Id: t.ID, Hash: t.Hash[:], Repo: t.Repo.String(), RunId: t.RunID, RunAttempt: t.RunAttempt, JobId: t.JobID,
65 Archive: t.Archive, ExpiresAt: encodeTime(t.ExpiresAt), CreatedAt: encodeTime(t.CreatedAt), Sha: t.Sha, Permissions: t.Permissions, Oidc: t.OIDC,
66 })
67}
68
69func decodeRepoToken(val []byte) (domain.CIRepoToken, error) {
70 var pb storagepb.CIRepoToken
71 if err := proto.Unmarshal(val, &pb); err != nil {
72 return domain.CIRepoToken{}, fmt.Errorf("decode repo token: %w", err)
73 }
74 repo, err := domain.ParseRepoName(pb.GetRepo())
75 if err != nil {
76 return domain.CIRepoToken{}, fmt.Errorf("decode repo token: %w", err)
77 }
78 t := domain.CIRepoToken{
79 ID: pb.GetId(), Repo: repo, RunID: pb.GetRunId(), RunAttempt: pb.GetRunAttempt(), JobID: pb.GetJobId(),
80 Archive: pb.GetArchive(), ExpiresAt: decodeTime(pb.GetExpiresAt()), CreatedAt: decodeTime(pb.GetCreatedAt()),
81 Sha: pb.GetSha(), Permissions: pb.GetPermissions(), OIDC: pb.GetOidc(),
82 }
83 copy(t.Hash[:], pb.GetHash())
84 return t, nil
85}
86
87// stageRepoTokens stages the creation of the tokens, the deletion of the
88// tokens of the jobs to revoke, and the resolutions to keep.
89func (c *CI) stageRepoTokens(ctx context.Context, b *kv.Batch, w ports.CIWrite, guarded map[domain.RepoName]bool) {
90 snap := c.db.Snapshot(ctx)
91 for _, ref := range w.RevokeJobs {
92 for key, item := range snap.Prefix(ctx, []byte(repoTokenJobPrefixOf(ref.RunID, ref.RunAttempt, ref.JobID))) {
93 b.Delete(append([]byte(nil), key...))
94 id := string(item.Value)
95 if id == "" {
96 id = string(key[strings.LastIndexByte(string(key), '/')+1:])
97 }
98 b.Delete(repoTokenKey(id))
99 if it, ok := snap.Get(ctx, repoTokenKey(id)); ok {
100 if t, err := decodeRepoToken(it.Value); err == nil {
101 b.Delete(repoTokenHashKey(t.Hash))
102 b.Delete(repoTokenRepoKey(t.Repo, id))
103 }
104 }
105 }
106 }
107 for _, t := range w.RepoTokens {
108 b.RequireAbsent(repoTokenKey(t.ID))
109 b.RequireAbsent(repoTokenHashKey(t.Hash))
110 b.Put(repoTokenKey(t.ID), encodeRepoToken(t))
111 b.Put(repoTokenHashKey(t.Hash), []byte(t.ID))
112 b.Put(repoTokenJobKey(t), []byte(t.ID))
113 b.Put(repoTokenRepoKey(t.Repo, t.ID), []byte(t.ID))
114 guarded[t.Repo] = true
115 }
116 for _, r := range w.Resolutions {
117 b.Put(resolutionKey(r.RunID, r.RunAttempt, r.Key), mustMarshal(&storagepb.CIActionResolution{
118 Key: r.Key, NameWithOwner: r.NameWithOwner, ResolvedSha: r.ResolvedSHA, Source: int32(r.Source), //nolint:gosec // 1 or 2
119 }))
120 }
121}
122
123// FindRepoTokenByHash resolves the hash index, then reads the token.
124func (c *CI) FindRepoTokenByHash(ctx context.Context, hash domain.TokenHash) (domain.CIRepoToken, error) {
125 snap := c.db.Snapshot(ctx)
126 idx, ok := snap.Get(ctx, repoTokenHashKey(hash))
127 if !ok {
128 return domain.CIRepoToken{}, domain.ErrNotFound
129 }
130 it, ok := snap.Get(ctx, repoTokenKey(string(idx.Value)))
131 if !ok {
132 return domain.CIRepoToken{}, domain.ErrNotFound
133 }
134 t, err := decodeRepoToken(it.Value)
135 if err != nil {
136 return domain.CIRepoToken{}, err
137 }
138 if t.Hash != hash {
139 return domain.CIRepoToken{}, domain.ErrNotFound
140 }
141 return t, nil
142}
143
144// GetRepoToken reads ghtoken/<id>.
145func (c *CI) GetRepoToken(ctx context.Context, id string) (domain.CIRepoToken, error) {
146 if !validTokenID(id) {
147 return domain.CIRepoToken{}, domain.ErrNotFound
148 }
149 it, ok := c.db.Get(ctx, repoTokenKey(id))
150 if !ok {
151 return domain.CIRepoToken{}, domain.ErrNotFound
152 }
153 return decodeRepoToken(it.Value)
154}
155
156// GetActionResolution reads the resolution kept for the key.
157func (c *CI) GetActionResolution(ctx context.Context, runID int64, attempt int32, key string) (domain.CIActionResolution, error) {
158 it, ok := c.db.Get(ctx, resolutionKey(runID, attempt, key))
159 if !ok {
160 return domain.CIActionResolution{}, domain.ErrNotFound
161 }
162 var pb storagepb.CIActionResolution
163 if err := proto.Unmarshal(it.Value, &pb); err != nil {
164 return domain.CIActionResolution{}, fmt.Errorf("decode action resolution: %w", err)
165 }
166 return domain.CIActionResolution{
167 RunID: runID, RunAttempt: attempt, Key: pb.GetKey(), NameWithOwner: pb.GetNameWithOwner(),
168 ResolvedSHA: pb.GetResolvedSha(), Source: domain.CIActionSource(pb.GetSource()),
169 }, nil
170}
171
172// tokenKeys lists the keys of the token with that id: the record, its hash
173// index, its entry in the list of its job and in the list of its repository.
174func tokenKeys(ctx context.Context, snap *kv.Snapshot, id string) [][]byte {
175 keys := [][]byte{repoTokenKey(id)}
176 if it, ok := snap.Get(ctx, repoTokenKey(id)); ok {
177 if t, err := decodeRepoToken(it.Value); err == nil {
178 keys = append(keys, repoTokenHashKey(t.Hash), repoTokenJobKey(t), repoTokenRepoKey(t.Repo, id))
179 }
180 }
181 return keys
182}
183
184// repoTokenKeys lists the keys of the repository tokens and action
185// resolutions of a run, for its deletion.
186func repoTokenKeys(ctx context.Context, snap *kv.Snapshot, runID int64) [][]byte {
187 var keys [][]byte
188 for key, item := range snap.Prefix(ctx, []byte(fmt.Sprintf("%s%020d/", repoTokenJobPrefix, runID))) {
189 keys = append(keys, append([]byte(nil), key...))
190 keys = append(keys, tokenKeys(ctx, snap, string(item.Value))...)
191 }
192 for key := range snap.Prefix(ctx, []byte(fmt.Sprintf("%s%020d/", resolutionPrefix, runID))) {
193 keys = append(keys, append([]byte(nil), key...))
194 }
195 return keys
196}
197
198// repoReaderKeys lists the keys of every repository token that reads the
199// repository, whichever job holds it: the archive tokens given to the jobs of
200// other repositories go with the repository they read.
201func repoReaderKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
202 var keys [][]byte
203 for key, item := range snap.Prefix(ctx, []byte(repoTokenRepoPrefixOf(repo))) {
204 keys = append(keys, append([]byte(nil), key...))
205 keys = append(keys, tokenKeys(ctx, snap, string(item.Value))...)
206 }
207 return keys
208}
209
210// CountRepoTokens counts the job repository tokens that read the repository
211// (every job's, archive tokens included). It serves tests and diagnostics.
212func (c *CI) CountRepoTokens(ctx context.Context, repo domain.RepoName) (int, error) {
213 n := 0
214 for range c.db.Snapshot(ctx).Prefix(ctx, []byte(repoTokenRepoPrefixOf(repo))) {
215 n++
216 }
217 return n, nil
218}