For reading and verification only. You may not copy, use, modify or redistribute this code. Full notice on the main page.
tunnel/jwks.go
| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/ed25519" |
| 6 | "encoding/base64" |
| 7 | "errors" |
| 8 | "log/slog" |
| 9 | "regexp" |
| 10 | "sync" |
| 11 | "time" |
| 12 | ) |
| 13 | |
| 14 | // The control plane's published keys are the root of trust for the sessions a node verifies offline. The node reads them through KeySource: "is this |
| 15 | // kid a key of the control plane, and which one". What fetches, caches and refreshes them is JWKSCache below (issuer pinned, 300 s cache, reload on an |
| 16 | // unknown kid, at most 16 keys, fail closed after 30 minutes without a refresh). |
| 17 | |
| 18 | // KeySource gives the public Ed25519 key of a control-plane kid. A source that cannot vouch for its keys any more (stale beyond the policy's window) |
| 19 | // answers false for every kid: the node then refuses every session, it never trusts an old set. |
| 20 | type KeySource interface { |
| 21 | Key(kid string) (ed25519.PublicKey, bool) |
| 22 | } |
| 23 | |
| 24 | // keyHealth is what a KeySource may add: whether it can vouch for its keys right now. When it cannot, a session whose kid it does not know is refused |
| 25 | // as control_unavailable (try again later) and not as auth (your credentials are wrong). |
| 26 | type keyHealth interface { |
| 27 | Healthy() bool |
| 28 | } |
| 29 | |
| 30 | // StaticKeys is a fixed key set: what the tests and the test-node binary use, and what a real source returns once it has a fresh set. |
| 31 | type StaticKeys map[string]ed25519.PublicKey |
| 32 | |
| 33 | func (k StaticKeys) Key(kid string) (ed25519.PublicKey, bool) { |
| 34 | key, ok := k[kid] |
| 35 | return key, ok |
| 36 | } |
| 37 | |
| 38 | // SwapKeys is a key set that can be replaced while the node runs (a rotation, or a refresh that returned a new set). It is safe for concurrent use. |
| 39 | type SwapKeys struct { |
| 40 | mu sync.RWMutex |
| 41 | keys StaticKeys |
| 42 | } |
| 43 | |
| 44 | func NewSwapKeys(keys StaticKeys) *SwapKeys { return &SwapKeys{keys: keys} } |
| 45 | |
| 46 | // Set replaces the whole set; a nil or empty set refuses everything. |
| 47 | func (s *SwapKeys) Set(keys StaticKeys) { |
| 48 | s.mu.Lock() |
| 49 | s.keys = keys |
| 50 | s.mu.Unlock() |
| 51 | } |
| 52 | |
| 53 | func (s *SwapKeys) Key(kid string) (ed25519.PublicKey, bool) { |
| 54 | s.mu.RLock() |
| 55 | defer s.mu.RUnlock() |
| 56 | return s.keys.Key(kid) |
| 57 | } |
| 58 | |
| 59 | var errBadJWKS = errors.New("bad_jwks") |
| 60 | |
| 61 | var reB64Charset = regexp.MustCompile(`^[A-Za-z0-9_-]+$`) |
| 62 | |
| 63 | // b64Decode is the strict base64url of the contracts: no padding and the canonical encoding (a text that decodes to bytes that encode differently |
| 64 | // is refused, so one token has one text). |
| 65 | func b64Decode(text string) ([]byte, error) { |
| 66 | if !reB64Charset.MatchString(text) { |
| 67 | return nil, errors.New("base64url") |
| 68 | } |
| 69 | raw, err := base64.RawURLEncoding.Strict().DecodeString(text) |
| 70 | if err != nil || base64.RawURLEncoding.EncodeToString(raw) != text { |
| 71 | return nil, errors.New("base64url") |
| 72 | } |
| 73 | return raw, nil |
| 74 | } |
| 75 | |
| 76 | func b64Encode(raw []byte) string { return base64.RawURLEncoding.EncodeToString(raw) } |
| 77 | |
| 78 | // publicKeyX decodes the x of an Ed25519 JWK: 43 canonical base64url characters that are 32 bytes. |
| 79 | func publicKeyX(x string) (ed25519.PublicKey, bool) { |
| 80 | if len(x) != 43 { |
| 81 | return nil, false |
| 82 | } |
| 83 | raw, err := b64Decode(x) |
| 84 | if err != nil || len(raw) != ed25519.PublicKeySize { |
| 85 | return nil, false |
| 86 | } |
| 87 | return ed25519.PublicKey(raw), true |
| 88 | } |
| 89 | |
| 90 | // ParseJWKS reads the control plane's GET /.well-known/device-jwks.json (jwks.schema.json): one to max_keys Ed25519 keys, each with its kid, alg |
| 91 | // EdDSA, use sig and key_ops [verify], no private part, no repeated kid. A set the schema refuses is bad_jwks and nothing of it is used. |
| 92 | func ParseJWKS(doc []byte) (StaticKeys, error) { |
| 93 | value, err := parseJSON(string(doc)) |
| 94 | if err != nil { |
| 95 | return nil, errBadJWKS |
| 96 | } |
| 97 | top, ok := value.(map[string]any) |
| 98 | if !ok || len(top) != 1 { |
| 99 | return nil, errBadJWKS |
| 100 | } |
| 101 | list, ok := top["keys"].([]any) |
| 102 | if !ok || len(list) < 1 || len(list) > policyData.JWKS.MaxKeys { |
| 103 | return nil, errBadJWKS |
| 104 | } |
| 105 | keys := StaticKeys{} |
| 106 | for _, item := range list { |
| 107 | m, ok := item.(map[string]any) |
| 108 | if !ok || len(m) != 7 { |
| 109 | return nil, errBadJWKS |
| 110 | } |
| 111 | kid, _ := m["kid"].(string) |
| 112 | x, _ := m["x"].(string) |
| 113 | ops, _ := m["key_ops"].([]any) |
| 114 | if m["kty"] != "OKP" || m["crv"] != "Ed25519" || m["alg"] != "EdDSA" || m["use"] != "sig" || !reKID.MatchString(kid) || |
| 115 | len(ops) != 1 || ops[0] != "verify" { |
| 116 | return nil, errBadJWKS |
| 117 | } |
| 118 | key, ok := publicKeyX(x) |
| 119 | if !ok { |
| 120 | return nil, errBadJWKS |
| 121 | } |
| 122 | if _, dup := keys[kid]; dup { |
| 123 | return nil, errBadJWKS |
| 124 | } |
| 125 | keys[kid] = key |
| 126 | } |
| 127 | return keys, nil |
| 128 | } |
| 129 | |
| 130 | // ---------------------------------------------------------------------------------------------------------------------------- the cache |
| 131 | |
| 132 | // JWKSConfig configures a JWKSCache. Every window defaults to the policy's. |
| 133 | type JWKSConfig struct { |
| 134 | // Fetch reads the control plane's GET /.well-known/device-jwks.json. The URL is the configured control plane's and nothing else: the issuer is |
| 135 | // pinned by the caller, an answer cannot name another place to read the keys from. |
| 136 | Fetch func(ctx context.Context) ([]byte, error) |
| 137 | Clock clock |
| 138 | TTL time.Duration // how long a fetched set is used before the next refresh |
| 139 | StaleAfter time.Duration // beyond this without a successful refresh the cache vouches for nothing |
| 140 | MinReload time.Duration // between two fetches that an unknown kid (or a stale set) asked for: a flood of random kids is one fetch, not thousands |
| 141 | FetchTimeout time.Duration |
| 142 | Observe func(ok bool) // every fetch's outcome (the node's reachability record) |
| 143 | Log *slog.Logger |
| 144 | } |
| 145 | |
| 146 | // JWKSCache is the node's KeySource for the control plane's keys. It fetches the set at start and every TTL (a failure is retried with the policy's |
| 147 | // backoff and the old set keeps serving), reloads when a session names a kid it does not hold (so a rotation reaches a node at the next RENEW, with at |
| 148 | // most one fetch per MinReload however many sessions ask), keeps at most the policy's number of keys (ParseJWKS refuses a bigger set whole), replaces |
| 149 | // the whole set on every successful fetch (a key the control plane dropped stops being trusted), and FAILS CLOSED: after StaleAfter without a refresh |
| 150 | // it answers false for every kid, an old set is never trusted. |
| 151 | type JWKSCache struct { |
| 152 | cfg JWKSConfig |
| 153 | |
| 154 | mu sync.Mutex |
| 155 | keys StaticKeys |
| 156 | fetchedAt time.Time // the last successful fetch |
| 157 | attemptedAt time.Time // the last fetch of any outcome that was asked for by a lookup |
| 158 | flight chan struct{} |
| 159 | failures int |
| 160 | } |
| 161 | |
| 162 | func NewJWKSCache(cfg JWKSConfig) *JWKSCache { |
| 163 | if cfg.TTL <= 0 { |
| 164 | cfg.TTL = secs(policyData.JWKS.CacheS) |
| 165 | } |
| 166 | if cfg.StaleAfter <= 0 { |
| 167 | cfg.StaleAfter = secs(policyData.JWKS.StaleFailClosedS) |
| 168 | } |
| 169 | if cfg.MinReload <= 0 { |
| 170 | cfg.MinReload = minReload |
| 171 | } |
| 172 | if cfg.FetchTimeout <= 0 { |
| 173 | cfg.FetchTimeout = fetchTimeout |
| 174 | } |
| 175 | if cfg.Clock.now == nil { |
| 176 | cfg.Clock = newClock(nil, nil) |
| 177 | } |
| 178 | if cfg.Log == nil { |
| 179 | cfg.Log = slog.New(slog.DiscardHandler) |
| 180 | } |
| 181 | return &JWKSCache{cfg: cfg} |
| 182 | } |
| 183 | |
| 184 | // fetchTimeout bounds one fetch of the keys (a lookup that triggers one waits at most this long). |
| 185 | const fetchTimeout = 5 * time.Second |
| 186 | |
| 187 | // minReload is the default gap between fetches that lookups ask for. |
| 188 | const minReload = 10 * time.Second |
| 189 | |
| 190 | func (c *JWKSCache) fresh(now time.Time) bool { |
| 191 | return !c.fetchedAt.IsZero() && now.Sub(c.fetchedAt) <= c.cfg.StaleAfter |
| 192 | } |
| 193 | |
| 194 | // Healthy reports whether the cache can vouch for its keys: it has a set that is not older than StaleAfter. |
| 195 | func (c *JWKSCache) Healthy() bool { |
| 196 | c.mu.Lock() |
| 197 | defer c.mu.Unlock() |
| 198 | return c.fresh(c.cfg.Clock.Now()) |
| 199 | } |
| 200 | |
| 201 | // Age is how long ago the last successful fetch was (a negative duration when there has not been one): the monitor's "age of the JWKS". |
| 202 | func (c *JWKSCache) Age() time.Duration { |
| 203 | c.mu.Lock() |
| 204 | defer c.mu.Unlock() |
| 205 | if c.fetchedAt.IsZero() { |
| 206 | return -1 |
| 207 | } |
| 208 | return c.cfg.Clock.Now().Sub(c.fetchedAt) |
| 209 | } |
| 210 | |
| 211 | // Key is the KeySource lookup. |
| 212 | func (c *JWKSCache) Key(kid string) (ed25519.PublicKey, bool) { |
| 213 | now := c.cfg.Clock.Now() |
| 214 | c.mu.Lock() |
| 215 | fresh := c.fresh(now) |
| 216 | key, known := c.keys[kid] |
| 217 | due := !c.fetchedAt.IsZero() && now.Sub(c.fetchedAt) >= c.cfg.TTL |
| 218 | askAllowed := c.attemptedAt.IsZero() || now.Sub(c.attemptedAt) >= c.cfg.MinReload |
| 219 | if due && fresh && known && askAllowed { |
| 220 | // The attempt is stamped here, with the reading this lookup took, not by the goroutine whenever it gets to run: the gap to the next attempt |
| 221 | // must not depend on scheduling, and a burst of lookups past the TTL starts one refresh, not one each. |
| 222 | c.attemptedAt = now |
| 223 | } |
| 224 | c.mu.Unlock() |
| 225 | |
| 226 | switch { |
| 227 | case fresh && known && !due: |
| 228 | return key, true |
| 229 | case fresh && known: |
| 230 | // Past its TTL but still vouched for: serve it and refresh behind the lookup (the periodic loop is the main path; this covers a busy node |
| 231 | // whose loop is in a backoff). |
| 232 | if askAllowed { |
| 233 | go func() { _ = c.reload(context.Background(), false) }() |
| 234 | } |
| 235 | return key, true |
| 236 | case !askAllowed: |
| 237 | return nil, false |
| 238 | } |
| 239 | // An unknown kid, or a set that is no longer vouched for: one fetch (every concurrent lookup waits for the same one), then the answer. |
| 240 | _ = c.reload(context.Background(), true) |
| 241 | c.mu.Lock() |
| 242 | defer c.mu.Unlock() |
| 243 | if !c.fresh(c.cfg.Clock.Now()) { |
| 244 | return nil, false |
| 245 | } |
| 246 | key, known = c.keys[kid] |
| 247 | return key, known |
| 248 | } |
| 249 | |
| 250 | // Refresh fetches the set now (single flight) and replaces the cached one when the answer is a valid set. It returns the fetch's error. |
| 251 | func (c *JWKSCache) Refresh(ctx context.Context) error { return c.reload(ctx, false) } |
| 252 | |
| 253 | func (c *JWKSCache) reload(ctx context.Context, fromLookup bool) error { |
| 254 | c.mu.Lock() |
| 255 | if wait := c.flight; wait != nil { |
| 256 | c.mu.Unlock() |
| 257 | select { |
| 258 | case <-wait: |
| 259 | case <-ctx.Done(): |
| 260 | return ctx.Err() |
| 261 | } |
| 262 | c.mu.Lock() |
| 263 | defer c.mu.Unlock() |
| 264 | if c.fresh(c.cfg.Clock.Now()) { |
| 265 | return nil |
| 266 | } |
| 267 | return errors.New("jwks refresh failed") |
| 268 | } |
| 269 | wait := make(chan struct{}) |
| 270 | c.flight = wait |
| 271 | if fromLookup { |
| 272 | c.attemptedAt = c.cfg.Clock.Now() |
| 273 | } |
| 274 | c.mu.Unlock() |
| 275 | |
| 276 | fctx, cancel := context.WithTimeout(ctx, c.cfg.FetchTimeout) |
| 277 | doc, err := c.cfg.Fetch(fctx) |
| 278 | cancel() |
| 279 | var keys StaticKeys |
| 280 | if err == nil { |
| 281 | keys, err = ParseJWKS(doc) |
| 282 | } |
| 283 | now := c.cfg.Clock.Now() |
| 284 | c.mu.Lock() |
| 285 | if err == nil { |
| 286 | c.keys, c.fetchedAt, c.failures = keys, now, 0 |
| 287 | } else { |
| 288 | c.failures++ |
| 289 | } |
| 290 | c.flight = nil |
| 291 | close(wait) |
| 292 | c.mu.Unlock() |
| 293 | if c.cfg.Observe != nil { |
| 294 | c.cfg.Observe(err == nil) |
| 295 | } |
| 296 | if err != nil { |
| 297 | c.cfg.Log.Warn("jwks_refresh_failed", "reason", errorWord(err)) |
| 298 | } |
| 299 | return err |
| 300 | } |
| 301 | |
| 302 | // errorWord is a short, address-free word for a failure: nothing from the network goes to a log. |
| 303 | func errorWord(err error) string { |
| 304 | switch { |
| 305 | case errors.Is(err, errBadJWKS): |
| 306 | return "bad_jwks" |
| 307 | case errors.Is(err, context.DeadlineExceeded): |
| 308 | return "timeout" |
| 309 | case errors.Is(err, context.Canceled): |
| 310 | return "canceled" |
| 311 | } |
| 312 | var ce *ControlError |
| 313 | if errors.As(err, &ce) { |
| 314 | return ce.Code |
| 315 | } |
| 316 | return "unreachable" |
| 317 | } |
| 318 | |
| 319 | // Run keeps the set fresh until ctx ends: a fetch now, then one every TTL (spread a little, so two nodes do not call together), and after a failure the |
| 320 | // policy's backoff. |
| 321 | func (c *JWKSCache) Run(ctx context.Context) { |
| 322 | for { |
| 323 | err := c.Refresh(ctx) |
| 324 | c.mu.Lock() |
| 325 | failures := c.failures |
| 326 | c.mu.Unlock() |
| 327 | wait := jitter(c.cfg.TTL, 10) |
| 328 | if err != nil { |
| 329 | wait = backoff(failures) |
| 330 | } |
| 331 | if !c.cfg.Clock.sleep(ctx, wait) { |
| 332 | return |
| 333 | } |
| 334 | } |
| 335 | } |