1// SPDX-License-Identifier: MIT
2
3package cachesvc
4
5import (
6 "context"
7 "encoding/json"
8 "io"
9 "net"
10 "net/http"
11 "net/url"
12 "path/filepath"
13 "strings"
14 "testing"
15 "time"
16)
17
18// channel serves svc on a Unix socket as the channel of sandbox id and returns
19// an HTTP client that can reach that socket only.
20func channel(t *testing.T, svc *Service, id, job string, opts ...ServeOption) *http.Client {
21 t.Helper()
22 sock := filepath.Join(t.TempDir(), "c")
23 ln, err := net.Listen("unix", sock)
24 if err != nil {
25 t.Fatal(err)
26 }
27 if err := svc.Serve(id, ln, job, opts...); err != nil {
28 t.Fatal(err)
29 }
30 t.Cleanup(func() { svc.Stop(id) })
31 return &http.Client{Timeout: 10 * time.Second, Transport: &http.Transport{
32 DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
33 var d net.Dialer
34 return d.DialContext(ctx, "unix", sock)
35 },
36 DisableKeepAlives: true,
37 }}
38}
39
40func lookupStatus(t *testing.T, c *http.Client, token string) int {
41 t.Helper()
42 req, err := http.NewRequest("GET", "http://cache"+APIPath+"cache?keys=k&version="+v1, nil)
43 if err != nil {
44 t.Fatal(err)
45 }
46 if token != "" {
47 req.Header.Set("Authorization", "Bearer "+token)
48 }
49 resp, err := c.Do(req)
50 if err != nil {
51 t.Fatal(err)
52 }
53 _ = resp.Body.Close()
54 return resp.StatusCode
55}
56
57// A channel is the listener of one sandbox. The token rules are the service's
58// (a missing or foreign token is a 403, a valid one is served), and on top of
59// them a channel serves the job of its sandbox only: the token of another job,
60// even a valid one, gets nothing, so a token that leaves its sandbox is useless
61// on every other sandbox's channel.
62func TestChannelServesItsOwnJobOnly(t *testing.T) {
63 r := newRig(t, 1<<20)
64 a := channel(t, r.svc, "er0000000a", "job-a")
65 b := channel(t, r.svc, "er0000000b", "job-b")
66 tokA := r.job("job-a", repoA, main_, main_).token
67 tokB := r.job("job-b", repoA, main_, main_).token
68
69 if got := lookupStatus(t, a, tokA); got != http.StatusNoContent {
70 t.Fatalf("A's token on A's channel: %d", got)
71 }
72 if got := lookupStatus(t, b, tokB); got != http.StatusNoContent {
73 t.Fatalf("B's token on B's channel: %d", got)
74 }
75 if got := lookupStatus(t, b, tokA); got != http.StatusForbidden {
76 t.Fatalf("A's token on B's channel: %d", got)
77 }
78 if got := lookupStatus(t, a, tokB); got != http.StatusForbidden {
79 t.Fatalf("B's token on A's channel: %d", got)
80 }
81 if got := lookupStatus(t, a, ""); got != http.StatusForbidden {
82 t.Fatalf("no token: %d", got)
83 }
84 if got := lookupStatus(t, a, "forged.token"); got != http.StatusForbidden {
85 t.Fatalf("forged token: %d", got)
86 }
87}
88
89// archiveStatus requests the archive at loc (a lookup's archiveLocation) on the
90// channel c, with the given method, and returns the status and the body.
91func archiveStatus(t *testing.T, c *http.Client, method, loc string) (int, string) {
92 t.Helper()
93 u, err := url.Parse(loc)
94 if err != nil {
95 t.Fatal(err)
96 }
97 req, err := http.NewRequest(method, "http://cache"+u.RequestURI(), nil)
98 if err != nil {
99 t.Fatal(err)
100 }
101 resp, err := c.Do(req)
102 if err != nil {
103 t.Fatal(err)
104 }
105 defer func() { _ = resp.Body.Close() }()
106 b, _ := io.ReadAll(resp.Body)
107 return resp.StatusCode, string(b)
108}
109
110// The signed URL of an archive is bound to the job it was issued to: it works on
111// the channel of that job and is refused, for GET and HEAD, on the channel of
112// any other, even for a job of the same repository and scope.
113func TestChannelArchiveURLBoundToJob(t *testing.T) {
114 r := newRig(t, 1<<20)
115 a := channel(t, r.svc, "er0000000a", "job-a")
116 b := channel(t, r.svc, "er0000000b", "job-b")
117 r.job("job-a", repoA, main_, main_).save("k", v1, []byte("secret-a"))
118
119 req, _ := http.NewRequest("GET", "http://cache"+APIPath+"cache?keys=k&version="+v1, nil)
120 req.Header.Set("Authorization", "Bearer "+r.job("job-a", repoA, main_, main_).token)
121 resp, err := a.Do(req)
122 if err != nil {
123 t.Fatal(err)
124 }
125 var e struct {
126 ArchiveLocation string `json:"archiveLocation"`
127 }
128 err = json.NewDecoder(resp.Body).Decode(&e)
129 _ = resp.Body.Close()
130 if err != nil || resp.StatusCode != http.StatusOK {
131 t.Fatalf("lookup: %d %v", resp.StatusCode, err)
132 }
133
134 if code, body := archiveStatus(t, a, "GET", e.ArchiveLocation); code != 200 || body != "secret-a" {
135 t.Fatalf("GET on the issuing channel: %d %q", code, body)
136 }
137 if code, _ := archiveStatus(t, a, "HEAD", e.ArchiveLocation); code != 200 {
138 t.Fatalf("HEAD on the issuing channel: %d", code)
139 }
140 if code, body := archiveStatus(t, b, "GET", e.ArchiveLocation); code != http.StatusForbidden || strings.Contains(body, "secret") {
141 t.Fatalf("GET on another job's channel: %d %q", code, body)
142 }
143 if code, _ := archiveStatus(t, b, "HEAD", e.ArchiveLocation); code != http.StatusForbidden {
144 t.Fatalf("HEAD on another job's channel: %d", code)
145 }
146 // Rewriting the job in the URL does not help: the signature covers it.
147 u, _ := url.Parse(e.ArchiveLocation)
148 q := u.Query()
149 q.Set("job", "job-b")
150 u.RawQuery = q.Encode()
151 if code, _ := archiveStatus(t, b, "GET", u.String()); code != http.StatusForbidden {
152 t.Fatalf("URL with the job rewritten: %d", code)
153 }
154}
155
156// The scope rules do not depend on the channel: a read-only job cannot reserve,
157// a job writes to its own ref and never reads another repository's entries.
158func TestChannelKeepsScopeRules(t *testing.T) {
159 r := newRig(t, 1<<20)
160 w := channel(t, r.svc, "er0000000a", "job-w")
161 ro := channel(t, r.svc, "er0000000b", "job-ro")
162 other := channel(t, r.svc, "er0000000c", "job-o")
163 r.job("job-w", repoA, main_, main_).save("k", v1, []byte("data"))
164
165 check := func(c *http.Client, jobID, repo, write string, wantHit bool) {
166 t.Helper()
167 tok := r.svc.Token(jobID, Scope{Repository: repo, WriteRef: write, ReadRefs: []string{main_}}, time.Hour)
168 got := lookupStatus(t, c, tok)
169 if wantHit != (got == http.StatusOK) || (!wantHit && got != http.StatusNoContent) {
170 t.Fatalf("%s: lookup %d, want hit=%v", jobID, got, wantHit)
171 }
172 }
173 check(w, "job-w", repoA, main_, true)
174 check(ro, "job-ro", repoA, "", true)
175 check(other, "job-o", "bnema/other", main_, false)
176
177 roTok := r.svc.Token("job-ro", Scope{Repository: repoA, ReadRefs: []string{main_}}, time.Hour)
178 req, _ := http.NewRequest("POST", "http://cache"+APIPath+"caches", nil)
179 req.Header.Set("Authorization", "Bearer "+roTok)
180 resp, err := ro.Do(req)
181 if err != nil {
182 t.Fatal(err)
183 }
184 _ = resp.Body.Close()
185 if resp.StatusCode != http.StatusForbidden {
186 t.Fatalf("a read-only job reserved: %d", resp.StatusCode)
187 }
188}
189
190// The connection caps apply to a channel as to any listener, and Stop closes
191// the listener and the open connections of that sandbox only.
192func TestChannelCapsAndStop(t *testing.T) {
193 r := newRigWith(t, Config{Max: 1 << 20, MinCharge: 1, ConnsPerListener: 1})
194 sockA, sockB := filepath.Join(t.TempDir(), "a"), filepath.Join(t.TempDir(), "b")
195 for id, sock := range map[string]string{"er0000000a": sockA, "er0000000b": sockB} {
196 ln, err := net.Listen("unix", sock)
197 if err != nil {
198 t.Fatal(err)
199 }
200 if err := r.svc.Serve(id, ln, "j-"+id); err != nil {
201 t.Fatal(err)
202 }
203 }
204 held, err := net.Dial("unix", sockA)
205 if err != nil {
206 t.Fatal(err)
207 }
208 defer func() { _ = held.Close() }()
209 // The one slot of A is taken once the server has accepted `held`: a second
210 // connection is then closed at once. B has its own share.
211 waitUntil(t, func() bool {
212 c, err := net.Dial("unix", sockA)
213 if err != nil {
214 return false
215 }
216 defer func() { _ = c.Close() }()
217 _ = c.SetReadDeadline(time.Now().Add(time.Second))
218 _, err = c.Read(make([]byte, 1))
219 return err != nil && !isTimeout(err)
220 })
221 other, err := net.Dial("unix", sockB)
222 if err != nil {
223 t.Fatal(err)
224 }
225 defer func() { _ = other.Close() }()
226
227 r.svc.Stop("er0000000a")
228 _ = held.SetReadDeadline(time.Now().Add(5 * time.Second))
229 if _, err := held.Read(make([]byte, 1)); err == nil || isTimeout(err) {
230 t.Fatalf("the open connection of a stopped channel: %v", err)
231 }
232 if c, err := net.Dial("unix", sockA); err == nil {
233 _ = c.Close()
234 t.Fatal("a stopped channel still accepts")
235 }
236 // B is not affected.
237 _ = other.SetReadDeadline(time.Now().Add(200 * time.Millisecond))
238 if _, err := other.Read(make([]byte, 1)); err == nil || !isTimeout(err) {
239 t.Fatalf("B's connection after A's stop: %v", err)
240 }
241 r.svc.Stop("er0000000a") // idempotent, unknown ids ignored
242 r.svc.Stop("unknown")
243}
244
245// One channel per sandbox id, none once the service is closed, and CloseAll
246// closes the listeners it was given.
247func TestChannelServeErrorsAndCloseAll(t *testing.T) {
248 r := newRig(t, 1<<20)
249 sock := filepath.Join(t.TempDir(), "s")
250 ln, err := net.Listen("unix", sock)
251 if err != nil {
252 t.Fatal(err)
253 }
254 if err := r.svc.Serve("er0000000a", ln, "j"); err != nil {
255 t.Fatal(err)
256 }
257 dup, err := net.Listen("unix", filepath.Join(t.TempDir(), "d"))
258 if err != nil {
259 t.Fatal(err)
260 }
261 if err := r.svc.Serve("er0000000a", dup, "j"); err == nil {
262 t.Fatal("two channels for one sandbox")
263 }
264 if _, err := dup.Accept(); err == nil {
265 t.Fatal("the refused listener was left open")
266 }
267 if err := r.svc.CloseAll(); err != nil {
268 t.Fatal(err)
269 }
270 if c, err := net.Dial("unix", sock); err == nil {
271 _ = c.Close()
272 t.Fatal("CloseAll left a channel open")
273 }
274 late, err := net.Listen("unix", filepath.Join(t.TempDir(), "l"))
275 if err != nil {
276 t.Fatal(err)
277 }
278 if err := r.svc.Serve("er0000000b", late, "j"); err == nil {
279 t.Fatal("Serve after CloseAll")
280 }
281 if _, err := late.Accept(); err == nil {
282 t.Fatal("the listener of a refused Serve was left open")
283 }
284}
285
286func isTimeout(err error) bool {
287 ne, ok := err.(net.Error)
288 return ok && ne.Timeout()
289}
290
291func waitUntil(t *testing.T, ok func() bool) {
292 t.Helper()
293 for end := time.Now().Add(10 * time.Second); time.Now().Before(end); time.Sleep(20 * time.Millisecond) {
294 if ok() {
295 return
296 }
297 }
298 t.Fatal("condition not reached in 10s")
299}