1// SPDX-License-Identifier: MIT
2
3package guest
4
5import (
6 "context"
7 "errors"
8 "io"
9 "net"
10 "net/http"
11 "path/filepath"
12 "strconv"
13 "testing"
14 "time"
15
16 "bnema.dev/ermite-runner/internal/execstream/execpb"
17)
18
19func TestCacheURL(t *testing.T) {
20 for _, tc := range []struct {
21 port uint32
22 want string
23 }{
24 // The toolkit appends "_apis/artifactcache/…" to this: the slash is needed.
25 {9089, "http://127.0.0.1:9089/"},
26 {0, ""},
27 {70000, ""},
28 } {
29 if got := cacheURL(tc.port); got != tc.want {
30 t.Errorf("cacheURL(%d) = %q, want %q", tc.port, got, tc.want)
31 }
32 }
33}
34
35// noBridge is a sandbox without a docker daemon.
36func noBridge() (net.IP, error) { return nil, errors.New("no docker0") }
37
38// serveCache serves the cache channel next to the executor target in dir; it
39// answers with the Authorization header it got.
40func serveCache(t *testing.T, dir string) Target {
41 t.Helper()
42 ln, err := net.Listen("unix", filepath.Join(dir, "cache.sock"))
43 if err != nil {
44 t.Fatal(err)
45 }
46 srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
47 _, _ = io.WriteString(w, "cache says "+r.Header.Get("Authorization"))
48 })}
49 go func() { _ = srv.Serve(ln) }()
50 t.Cleanup(func() { _ = srv.Close() })
51 return Target{Network: "unix", Addr: filepath.Join(dir, "exec.sock")}
52}
53
54// get fetches url with the token and returns the body.
55func get(t *testing.T, url string) string {
56 t.Helper()
57 req, _ := http.NewRequest("GET", url+"_apis/artifactcache/cache", nil)
58 req.Header.Set("Authorization", "Bearer tok")
59 cl := &http.Client{Timeout: 5 * time.Second, Transport: &http.Transport{DisableKeepAlives: true}}
60 resp, err := cl.Do(req)
61 if err != nil {
62 t.Fatal(err)
63 }
64 body, _ := io.ReadAll(resp.Body)
65 _ = resp.Body.Close()
66 return string(body)
67}
68
69// freePort returns a loopback port nobody listens on.
70func freePort(t *testing.T) uint32 {
71 t.Helper()
72 ln, err := net.Listen("tcp", "127.0.0.1:0")
73 if err != nil {
74 t.Skipf("loopback sockets are not allowed here: %v", err)
75 }
76 defer func() { _ = ln.Close() }()
77 return uint32(ln.Addr().(*net.TCPAddr).Port)
78}
79
80// A job with a cache service gets the relay on its loopback port, and the URL
81// and token for the actions; the relay reaches the cache channel of the host,
82// the sibling of the executor target, and ends with the job.
83func TestStartCacheRelay(t *testing.T) {
84 dir := t.TempDir()
85 execSock := filepath.Join(dir, "exec.sock")
86 ln, err := net.Listen("unix", filepath.Join(dir, "cache.sock"))
87 if err != nil {
88 t.Fatal(err)
89 }
90 srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
91 _, _ = io.WriteString(w, "cache says "+r.Header.Get("Authorization"))
92 })}
93 go func() { _ = srv.Serve(ln) }()
94 t.Cleanup(func() { _ = srv.Close() })
95
96 port := freePort(t)
97 spec := &execpb.JobSpec{CachePort: port, CacheToken: "tok", CacheMode: "read"}
98 got := startCacheRelay(context.Background(), Target{Network: "unix", Addr: execSock}, spec, noBridge)
99 if got.relay == nil || got.token != "tok" || got.mode != "read" || got.url != cacheURL(port) ||
100 got.bridge != nil || got.containerURL != "" || got.results || got.cacheOff {
101 t.Fatalf("relay %+v", got)
102 }
103
104 req, _ := http.NewRequest("GET", got.url+"_apis/artifactcache/cache", nil)
105 req.Header.Set("Authorization", "Bearer tok")
106 cl := &http.Client{Timeout: 5 * time.Second, Transport: &http.Transport{DisableKeepAlives: true}}
107 resp, err := cl.Do(req)
108 if err != nil {
109 t.Fatal(err)
110 }
111 body, _ := io.ReadAll(resp.Body)
112 _ = resp.Body.Close()
113 if string(body) != "cache says Bearer tok" {
114 t.Fatalf("got %q", body)
115 }
116
117 got.close()
118 if c, err := net.DialTimeout("tcp", got.relay.Addr().String(), time.Second); err == nil {
119 _ = c.Close()
120 t.Fatal("the relay outlives the job")
121 }
122}
123
124// No cache service, no relay and no variables; a port that is taken leaves the
125// job without a cache instead of failing it.
126func TestStartCacheRelayWithoutServiceOrPort(t *testing.T) {
127 target := Target{Network: "unix", Addr: filepath.Join(t.TempDir(), "exec.sock")}
128 for _, spec := range []*execpb.JobSpec{{}, {CachePort: 9089}, {CacheToken: "tok"}} {
129 if got := startCacheRelay(context.Background(), target, spec, noBridge); got != (cacheRelay{}) {
130 t.Errorf("%+v: relay %+v", spec, got)
131 }
132 }
133 busy, err := net.Listen("tcp", "127.0.0.1:0")
134 if err != nil {
135 t.Skipf("loopback sockets are not allowed here: %v", err)
136 }
137 defer func() { _ = busy.Close() }()
138 spec := &execpb.JobSpec{CachePort: uint32(busy.Addr().(*net.TCPAddr).Port), CacheToken: "tok"}
139 if got := startCacheRelay(context.Background(), target, spec, noBridge); got != (cacheRelay{}) {
140 t.Errorf("taken port: relay %+v", got)
141 }
142}
143
144// The relay also listens on the address of the docker bridge, on the same
145// port, for the steps that run in a container; their URL has that address.
146// 127.0.0.2 stands for docker0 (a loopback address that is bindable on Linux).
147func TestStartCacheRelayOnBridge(t *testing.T) {
148 target := serveCache(t, t.TempDir())
149 port := freePort(t)
150 probe, err := net.Listen("tcp", net.JoinHostPort("127.0.0.2", strconv.FormatUint(uint64(port), 10)))
151 if err != nil {
152 t.Skipf("127.0.0.2 is not bindable here: %v", err)
153 }
154 _ = probe.Close()
155 bridge := func() (net.IP, error) { return net.ParseIP("127.0.0.2"), nil }
156 got := startCacheRelay(context.Background(), target, &execpb.JobSpec{CachePort: port, CacheToken: "tok", CacheMode: "write"}, bridge)
157 defer got.close()
158 want := "http://127.0.0.2:" + strconv.FormatUint(uint64(port), 10) + "/"
159 if got.bridge == nil || got.containerURL != want || got.url != cacheURL(port) {
160 t.Fatalf("relay %+v", got)
161 }
162 for _, u := range []string{got.url, got.containerURL} {
163 if body := get(t, u); body != "cache says Bearer tok" {
164 t.Errorf("%s: got %q", u, body)
165 }
166 }
167 got.close()
168 if c, err := net.DialTimeout("tcp", got.bridge.Addr().String(), time.Second); err == nil {
169 _ = c.Close()
170 t.Fatal("the bridge relay outlives the job")
171 }
172}
173
174// A bridge address that cannot be listened on leaves the loopback relay, and
175// no URL for the containers.
176func TestStartCacheRelayBridgeFails(t *testing.T) {
177 target := serveCache(t, t.TempDir())
178 port := freePort(t)
179 // 192.0.2.1 (TEST-NET-1) is not an address of this machine.
180 bridge := func() (net.IP, error) { return net.ParseIP("192.0.2.1"), nil }
181 got := startCacheRelay(context.Background(), target, &execpb.JobSpec{CachePort: port, CacheToken: "tok"}, bridge)
182 defer got.close()
183 if got.relay == nil || got.bridge != nil || got.containerURL != "" {
184 t.Fatalf("relay %+v", got)
185 }
186}
187
188// JobSpec.results: the channel serves the results service; without a cache
189// mode it serves nothing else, and the cache variables are off.
190func TestStartCacheRelayResults(t *testing.T) {
191 target := serveCache(t, t.TempDir())
192 for _, tc := range []struct {
193 mode string
194 results, cacheOff bool
195 }{{"write", true, false}, {"", true, true}, {"read", false, false}, {"", false, false}} {
196 spec := &execpb.JobSpec{CachePort: freePort(t), CacheToken: "tok", CacheMode: tc.mode, Results: tc.results}
197 got := startCacheRelay(context.Background(), target, spec, noBridge)
198 got.close()
199 if got.relay == nil || got.results != tc.results || got.cacheOff != tc.cacheOff {
200 t.Errorf("%+v: relay %+v", tc, got)
201 }
202 }
203}
204
205// docker0IPv4 without the interface is an error, not an address.
206func TestDocker0IPv4(t *testing.T) {
207 ip, err := docker0IPv4()
208 if _, ierr := net.InterfaceByName(BridgeInterface); ierr != nil {
209 if err == nil {
210 t.Errorf("no %s, got %v", BridgeInterface, ip)
211 }
212 return
213 }
214 if err == nil && ip.To4() == nil {
215 t.Errorf("not IPv4: %v", ip)
216 }
217}