ALL

For reading and verification only. You may not copy, use, modify or redistribute this code. Full notice on the main page.

tunnel/fakecp_test.go

SHA-256: d5969f98b7b57681bbec9e57e34ddb683215d5f50a2279f91ea6ecceb91195ba
324 lines · 9951 bytes · View on GitHub

1package main
2
3import (
4 "crypto/ed25519"
5 "crypto/rand"
6 "io"
7 "net/http"
8 "net/http/httptest"
9 "strconv"
10 "sync"
11 "testing"
12 "time"
13)
14
15// fakeCP is a control plane over real HTTPS (httptest): it authenticates the node with the contracts' own verifier (the Go twin of the TypeScript one
16// the Worker runs), decides authorize with the environment's fakeControl, keeps a feed and publishes the keys. It is what the real client, the JWKS
17// cache and the feed loop are tested against.
18type fakeCP struct {
19 t *testing.T
20 env *testEnv
21 srv *httptest.Server
22
23 nodeKID string
24 nodeKey ed25519.PrivateKey
25
26 mu sync.Mutex
27 nodeKeys map[string]ctNodeKey
28 seen map[string]bool
29 jwks map[string]ed25519.PublicKey // kid -> key, what GET /.well-known/device-jwks.json says
30 down int // when not 0, every endpoint answers with this status
31 hold time.Duration // how long a feed poll with wait is held, in real time, when nothing is new
32 events []FeedEvent
33 head int64
34 floor int64
35 changed chan struct{}
36 bodies [][]byte // the body of every authenticated authorize
37 headers []http.Header // the headers of every authenticated request, to see that nothing but the assertion authenticates
38 refused int // requests whose assertion was refused
39 feedLog []feedCall
40 counts map[string]int
41 inflight int
42 maxIn int
43 delay time.Duration
44 // redirect, when set, answers authorize with a redirect to that URL
45 redirect string
46}
47
48type feedCall struct {
49 since int64
50 wait int64
51 url string
52}
53
54func newFakeCP(t *testing.T, env *testEnv) *fakeCP {
55 t.Helper()
56 pub, priv, err := ed25519.GenerateKey(rand.Reader)
57 if err != nil {
58 t.Fatal(err)
59 }
60 cp := &fakeCP{
61 t: t, env: env, nodeKID: "node-t1-1", nodeKey: priv,
62 nodeKeys: map[string]ctNodeKey{}, seen: map[string]bool{}, jwks: map[string]ed25519.PublicKey{},
63 changed: make(chan struct{}), counts: map[string]int{},
64 }
65 cp.nodeKeys[cp.nodeKID] = ctNodeKey{node: testNodeID, x: b64Encode(pub)}
66 cp.jwks[testKID] = env.cpKey.Public().(ed25519.PublicKey)
67 cp.srv = httptest.NewTLSServer(http.HandlerFunc(cp.handle))
68 t.Cleanup(cp.srv.Close)
69 return cp
70}
71
72func (cp *fakeCP) url() string { return cp.srv.URL }
73
74// control is the real client of the node, pointed at this control plane.
75func (cp *fakeCP) control() *HTTPControl {
76 hc, err := NewHTTPControl(HTTPControlConfig{BaseURL: cp.srv.URL, NodeID: testNodeID, KID: cp.nodeKID, Key: cp.nodeKey, Client: cp.srv.Client(), Now: cp.env.now, Version: "test"})
77 if err != nil {
78 cp.t.Fatal(err)
79 }
80 return hc
81}
82
83func (cp *fakeCP) setDown(status int) {
84 cp.mu.Lock()
85 cp.down = status
86 cp.mu.Unlock()
87}
88
89// addEvent appends a feed event (epoch is the next one) and wakes the polls that are held.
90func (cp *fakeCP) addEvent(kind string, d *testDevice) int64 {
91 cp.mu.Lock()
92 defer cp.mu.Unlock()
93 cp.head++
94 e := FeedEvent{Epoch: cp.head, Type: kind, At: cp.env.now().Unix()}
95 if d != nil {
96 e.DeviceID, e.Label = d.id, d.label
97 }
98 cp.events = append(cp.events, e)
99 close(cp.changed)
100 cp.changed = make(chan struct{})
101 return e.Epoch
102}
103
104func (cp *fakeCP) callCount(name string) int {
105 cp.mu.Lock()
106 defer cp.mu.Unlock()
107 return cp.counts[name]
108}
109
110func (cp *fakeCP) maxInflight() int {
111 cp.mu.Lock()
112 defer cp.mu.Unlock()
113 return cp.maxIn
114}
115
116func (cp *fakeCP) feedCalls() []feedCall {
117 cp.mu.Lock()
118 defer cp.mu.Unlock()
119 return append([]feedCall(nil), cp.feedLog...)
120}
121
122func (cp *fakeCP) reply(w http.ResponseWriter, status int, body map[string]any) {
123 w.Header().Set("content-type", "application/json")
124 w.WriteHeader(status)
125 _, _ = io.WriteString(w, canonicalJSON(body))
126}
127
128func (cp *fakeCP) refuse(w http.ResponseWriter, status int, word string, extra map[string]any) {
129 body := map[string]any{"ok": false, "error": word}
130 for k, v := range extra {
131 body[k] = v
132 }
133 cp.reply(w, status, body)
134}
135
136func (cp *fakeCP) handle(w http.ResponseWriter, r *http.Request) {
137 cp.mu.Lock()
138 down := cp.down
139 cp.counts[r.URL.Path]++
140 cp.mu.Unlock()
141 if down != 0 {
142 cp.refuse(w, down, "end_to_end_unavailable", map[string]any{"retry_after_s": 15})
143 return
144 }
145 switch {
146 case r.URL.Path == jwksPath && r.Method == http.MethodGet:
147 cp.serveJWKS(w)
148 case r.URL.Path == authorizePath && r.Method == http.MethodPost:
149 cp.serveAuthorize(w, r)
150 case r.URL.Path == feedPath && r.Method == http.MethodGet:
151 cp.serveFeed(w, r)
152 default:
153 cp.refuse(w, 404, "not_found", nil)
154 }
155}
156
157func (cp *fakeCP) serveJWKS(w http.ResponseWriter) {
158 cp.mu.Lock()
159 list := []any{}
160 for kid, key := range cp.jwks {
161 list = append(list, map[string]any{"kty": "OKP", "crv": "Ed25519", "x": b64Encode(key), "kid": kid, "alg": "EdDSA", "use": "sig", "key_ops": []any{"verify"}})
162 }
163 cp.mu.Unlock()
164 cp.reply(w, 200, map[string]any{"keys": list})
165}
166
167// authenticate checks the node's assertion for exactly this request with the contracts' verifier. It answers 401 and reports false when it fails.
168func (cp *fakeCP) authenticate(w http.ResponseWriter, r *http.Request, body []byte) bool {
169 cp.mu.Lock()
170 defer cp.mu.Unlock()
171 result := ctVerifyNodeAssertion(r.Header.Get("x-all-node-assertion"), epochSeconds(cp.env.now()), cp.nodeKeys, r.Method, "https://"+r.Host+r.URL.RequestURI(), body, 5, cp.seen)
172 if !result.ok {
173 cp.refused++
174 cp.reply(w, 401, map[string]any{"ok": false, "error": "unauthorized"})
175 return false
176 }
177 cp.seen[result.jti] = true
178 cp.headers = append(cp.headers, r.Header.Clone())
179 return true
180}
181
182func (cp *fakeCP) serveAuthorize(w http.ResponseWriter, r *http.Request) {
183 body, _ := io.ReadAll(io.LimitReader(r.Body, 1<<20))
184 if cp.redirect != "" {
185 http.Redirect(w, r, cp.redirect, http.StatusFound)
186 return
187 }
188 if !cp.authenticate(w, r, body) {
189 return
190 }
191 cp.mu.Lock()
192 cp.bodies = append(cp.bodies, body)
193 cp.mu.Unlock()
194 doc, err := parseJSON(string(body))
195 if err != nil || len(ctSchemaSet.validate("authorize-request.schema.json", doc)) > 0 {
196 cp.refuse(w, 400, "invalid_authorize_request", nil)
197 return
198 }
199 m := doc.(map[string]any)
200 req := AuthorizeRequest{DeviceID: m["device_id"].(string), SessionJTI: m["session_jti"].(string), Reason: m["reason"].(string), Node: m["node"].(string)}
201 cp.mu.Lock()
202 cp.inflight++
203 cp.maxIn = max(cp.maxIn, cp.inflight)
204 delay := cp.delay
205 cp.mu.Unlock()
206 defer func() {
207 cp.mu.Lock()
208 cp.inflight--
209 cp.mu.Unlock()
210 }()
211 if delay > 0 {
212 select {
213 case <-time.After(delay):
214 case <-r.Context().Done():
215 return
216 }
217 }
218 resp, aerr := cp.env.cp.Authorize(r.Context(), req)
219 if aerr != nil {
220 if ce, ok := aerr.(*ControlError); ok {
221 extra := map[string]any{}
222 if ce.Code == codeLicenseInactive {
223 extra["status"] = "trial_ended"
224 }
225 cp.refuse(w, ce.Status, ce.Code, extra)
226 return
227 }
228 cp.refuse(w, 503, "end_to_end_unavailable", nil)
229 return
230 }
231 cp.reply(w, 200, resp.toMap())
232}
233
234// replyFeed answers a feed poll after checking the answer against the contracts' own rules (schema, strictly increasing epochs, scopes): the fake must
235// never be more lenient than the real control plane, or the node would be tested against something that does not exist.
236func (cp *fakeCP) replyFeed(w http.ResponseWriter, since int64, body map[string]any) {
237 if doc, err := parseJSON(canonicalJSON(body)); err != nil {
238 cp.t.Errorf("the fake feed made a body that is not JSON: %v", err)
239 } else if _, reset := body["reset"]; !reset {
240 if errs := ctCheckFeed(doc, float64(since)); len(errs) > 0 {
241 cp.t.Errorf("the fake feed broke the contract: %v", errs)
242 }
243 }
244 cp.reply(w, 200, body)
245}
246
247func (cp *fakeCP) serveFeed(w http.ResponseWriter, r *http.Request) {
248 if !cp.authenticate(w, r, nil) {
249 return
250 }
251 since, err := strconv.ParseInt(r.URL.Query().Get("since"), 10, 64)
252 if err != nil {
253 cp.refuse(w, 400, "invalid_feed_request", nil)
254 return
255 }
256 wait, _ := strconv.ParseInt(r.URL.Query().Get("wait"), 10, 64)
257 cp.mu.Lock()
258 cp.feedLog = append(cp.feedLog, feedCall{since: since, wait: wait, url: r.URL.RequestURI()})
259 hold := cp.hold
260 cp.mu.Unlock()
261
262 deadline := time.Now().Add(min(hold, time.Duration(wait)*time.Second))
263 for {
264 cp.mu.Lock()
265 if since > cp.head || since < cp.floor {
266 head := cp.head
267 cp.mu.Unlock()
268 cp.replyFeed(w, since, map[string]any{"ok": true, "v": 1, "epoch": head, "server_time": cp.env.now().Unix(), "reset": true, "events": []any{}})
269 return
270 }
271 var out []any
272 for _, e := range cp.events {
273 if e.Epoch > since {
274 var device, label any
275 if !e.global() {
276 device, label = e.DeviceID, e.Label
277 }
278 out = append(out, map[string]any{"epoch": e.Epoch, "type": e.Type, "device_id": device, "label": label, "at": e.At})
279 }
280 }
281 changed, head := cp.changed, cp.head
282 cp.mu.Unlock()
283 if len(out) > 0 || !time.Now().Before(deadline) {
284 if out == nil {
285 out = []any{}
286 }
287 cp.replyFeed(w, since, map[string]any{"ok": true, "v": 1, "epoch": head, "server_time": cp.env.now().Unix(), "events": out})
288 return
289 }
290 select {
291 case <-changed:
292 case <-time.After(time.Until(deadline)):
293 case <-r.Context().Done():
294 return
295 }
296 }
297}
298
299// httpNode starts a node whose control plane, key set and feed are the real clients talking to a fakeCP over HTTPS, on the fake clock when clk is set.
300func httpNode(t *testing.T, clk *fakeClock, mutate func(*Config)) (*testEnv, *fakeCP) {
301 t.Helper()
302 var cp *fakeCP
303 e := newEnv(t, clk, func(e *testEnv, cfg *Config) {
304 cp = newFakeCP(t, e)
305 control := cp.control()
306 health := &cpHealth{}
307 clock := newClock(e.now, cfg.AfterFunc)
308 jwks := NewJWKSCache(JWKSConfig{Fetch: control.FetchJWKS, Clock: clock, Log: cfg.Logger, MinReload: time.Millisecond, Observe: func(ok bool) {
309 if ok {
310 health.good(clock.Now())
311 } else {
312 health.bad(clock.Now())
313 }
314 }})
315 cfg.Keys, cfg.Control, cfg.Feed, cfg.Health = jwks, control, control, health
316 cfg.Background = append(cfg.Background, jwks.Run)
317 cfg.FeedStartWait = 5 * time.Second
318 cfg.MinRenewGap = 0
319 if mutate != nil {
320 mutate(cfg)
321 }
322 })
323 return e, cp
324}