1// SPDX-License-Identifier: MIT
2
3package e2e
4
5import (
6 "context"
7 "fmt"
8 "os"
9 "path/filepath"
10 "strings"
11 "sync"
12 "testing"
13 "time"
14
15 "bnema.dev/ermite-runner/internal/cachesvc"
16 "bnema.dev/ermite-runner/internal/execstream/execpb"
17 "bnema.dev/ermite-runner/internal/sandbox"
18 "bnema.dev/ermite-runner/internal/sandbox/container"
19 "bnema.dev/ermite-runner/internal/sandbox/disk"
20 "bnema.dev/ermite-runner/internal/sandbox/sandboxtest"
21)
22
23// newCacheDriver is newDriver with the cache service of the runner.
24func newCacheDriver(t *testing.T, image string, svc *cachesvc.Service) sandbox.Driver {
25 t.Helper()
26 dir := t.TempDir()
27 cfg := sandbox.Config{
28 Isolation: sandbox.Container, StateDir: filepath.Join(dir, "state"), Capacity: 2, Image: image,
29 JobCPUs: 2, JobMemoryMiB: 1024, JobDiskMiB: 2048, UIDBase: 900000,
30 Network: netPrefix("10.215.0.0/16"), Executable: buildRunner(t, dir), DNSPort: 0,
31 LockFile: filepath.Join(dir, "host.lock"), JobPids: 4096, Cache: svc,
32 }
33 empty := filepath.Join(dir, "mountinfo")
34 _ = os.WriteFile(empty, nil, 0o644)
35 run := rootlessMountRunner{image: image, mountinfo: empty, mu: &sync.Mutex{}}
36 d, err := container.New(context.Background(), cfg, container.Options{
37 Net: sandboxtest.NewPlumbing(), UnsafeAllowRootlessForTests: true, ConnectTimeout: 90 * time.Second, Run: run,
38 Mounter: &disk.Mounter{Run: run, Mountinfo: empty},
39 })
40 if err != nil {
41 t.Fatal(err)
42 }
43 t.Cleanup(func() { _ = d.Close() })
44 if err := d.Reap(context.Background()); err != nil {
45 t.Fatal(err)
46 }
47 return d
48}
49
50// cacheClient is what @actions/cache does against the v1 API, in python (the
51// image has it): reserve, upload a chunk, commit, look up, download the archive.
52// It talks to 127.0.0.1:9089, the loopback of the container, where the relay of
53// the guest delivers every connection to the runner over the cache socket.
54const cacheClient = `
55import json, sys, urllib.request, urllib.error
56base = "http://127.0.0.1:%d/_apis/artifactcache/"
57tok = "%s"
58def call(method, path, body=None, headers=None):
59 h = {"Authorization": "Bearer " + tok, "Accept": "application/json;api-version=6.0-preview.1"}
60 h.update(headers or {})
61 req = urllib.request.Request(base + path, data=body, method=method, headers=h)
62 try:
63 with urllib.request.urlopen(req, timeout=20) as r:
64 return r.status, r.read()
65 except urllib.error.HTTPError as e:
66 return e.code, e.read()
67data = bytes(range(256)) * 4096 # 1 MiB
68code, body = call("GET", "cache?keys=e2e-key&version=v1")
69print("first lookup", code)
70code, body = call("POST", "caches", json.dumps({"key": "e2e-key", "version": "v1", "cacheSize": len(data)}).encode(), {"Content-Type": "application/json"})
71print("reserve", code)
72cid = json.loads(body)["cacheId"]
73half = len(data) // 2
74for lo, hi in ((half, len(data)), (0, half)): # out of order, like the toolkit may
75 code, _ = call("PATCH", "caches/%%d" %% cid, data[lo:hi], {"Content-Type": "application/octet-stream", "Content-Range": "bytes %%d-%%d/*" %% (lo, hi - 1)})
76 print("chunk", lo, code)
77code, _ = call("POST", "caches/%%d" %% cid, json.dumps({"size": len(data)}).encode(), {"Content-Type": "application/json"})
78print("commit", code)
79code, body = call("GET", "cache?keys=e2e-key&version=v1")
80print("second lookup", code)
81loc = json.loads(body)["archiveLocation"]
82print("archive host", loc.split("/")[2])
83with urllib.request.urlopen(loc, timeout=20) as r:
84 got = r.read()
85print("download", "same" if got == data else "DIFFERENT", len(got))
86code, _ = call("GET", "cache?keys=e2e-key&version=v1", None, {"Authorization": "Bearer nope"})
87print("bad token", code)
88`
89
90// The cache works from inside a container: the service listens on no network,
91// the guest relay turns its loopback port into the cache socket of the
92// container, and a client in the job does a full save and restore through it.
93// A token made for another job is refused on this container's socket.
94func TestContainerCacheOnTheGuestLoopback(t *testing.T) {
95 image := needE2E(t)
96 svc := sandboxtest.NewCache(t)
97 d := newCacheDriver(t, image, svc)
98
99 tok := sandboxtest.CacheToken(svc, "e2e")
100 spec := jobSpec(fmt.Sprintf(`on: push
101jobs:
102 j:
103 runs-on: x
104 steps:
105 - name: cache client
106 shell: python3 {0}
107 run: |
108%s
109 - name: nothing listens on the gateway
110 shell: python3 {0}
111 run: |
112 import socket, struct
113 gw = ""
114 for line in open("/proc/net/route").read().splitlines()[1:]:
115 f = line.split()
116 if f[1] == "00000000":
117 gw = socket.inet_ntoa(struct.pack("<L", int(f[2], 16)))
118 print("default gateway", gw)
119 for port in (8089, 9089):
120 s = socket.socket(); s.settimeout(2)
121 try:
122 s.connect((gw, port)); print("gateway", gw, port, "OPEN")
123 except Exception as e:
124 print("gateway", port, "closed")
125 s.close()
126`, indent(fmt.Sprintf(cacheClient, cachesvc.DefaultPort, tok), 10)))
127 spec.CachePort, spec.CacheToken = cachesvc.DefaultPort, tok
128
129 s := runJob(t, d, spec, nil)
130 all := strings.Join(s.Logs, "\n")
131 if s.Result == nil || s.Result.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS || s.Exit.Code != 0 {
132 t.Fatalf("result %v exit %v\n%s", s.Result, s.Exit, all)
133 }
134 for _, want := range []string{
135 "first lookup 204", "reserve 201", "chunk 524288 204", "chunk 0 204", "commit 204", "second lookup 200",
136 "archive host 127.0.0.1:9089", "download same 1048576", "bad token 403",
137 "default gateway 10.215.", "gateway 8089 closed", "gateway 9089 closed",
138 } {
139 if !strings.Contains(all, want) {
140 t.Errorf("log lacks %q\n%s", want, all)
141 }
142 }
143 // (The log also echoes the script, which contains the word: match the output.)
144 for _, port := range []string{"8089", "9089"} {
145 if strings.Contains(all, "10.215.0.1 "+port+" OPEN") {
146 t.Errorf("the cache port %s answers on the gateway:\n%s", port, all)
147 }
148 }
149}
150
151// A job whose token was made for another job gets 403 on its own socket.
152func TestContainerCacheRefusesAnotherJobsToken(t *testing.T) {
153 image := needE2E(t)
154 svc := sandboxtest.NewCache(t)
155 d := newCacheDriver(t, image, svc)
156 tok := sandboxtest.CacheToken(svc, "someone-else")
157 spec := jobSpec(fmt.Sprintf(`on: push
158jobs:
159 j:
160 runs-on: x
161 steps:
162 - shell: python3 {0}
163 run: |
164%s
165`, indent(fmt.Sprintf(`import urllib.request, urllib.error
166req = urllib.request.Request("http://127.0.0.1:%d/_apis/artifactcache/cache?keys=k&version=v", headers={"Authorization": "Bearer %s"})
167try:
168 urllib.request.urlopen(req, timeout=20); print("status 200")
169except urllib.error.HTTPError as e:
170 print("status", e.code)
171`, cachesvc.DefaultPort, tok), 10)))
172 spec.CachePort, spec.CacheToken = cachesvc.DefaultPort, tok
173 s := runJob(t, d, spec, nil)
174 all := strings.Join(s.Logs, "\n")
175 if s.Result == nil || s.Result.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS || !strings.Contains(all, "status 403") {
176 t.Fatalf("result %v\n%s", s.Result, all)
177 }
178}
179
180// indent prefixes every non-empty line of s with n spaces.
181func indent(s string, n int) string {
182 pad := strings.Repeat(" ", n)
183 lines := strings.Split(strings.TrimSpace(s), "\n")
184 for i, l := range lines {
185 if l != "" {
186 lines[i] = pad + l
187 }
188 }
189 return strings.Join(lines, "\n")
190}