diff --git a/docs/changes/unreleased/1928-scoped-client-relay.md b/docs/changes/unreleased/1928-scoped-client-relay.md new file mode 100644 index 0000000000..c0d98d3213 --- /dev/null +++ b/docs/changes/unreleased/1928-scoped-client-relay.md @@ -0,0 +1,11 @@ +--- +kind: added +title: Scoped clients keep sealed slots and a signed queue +pr: 1928 +surface: [remote, docs] +invalidates: + - "The relay manual said the hosted Worker does not support scoped clients. The Go relay and the hosted Worker both register a client device, store its sealed slots, and accept a signed queue." + - "A device on the relay was a full device. A full device can now register a scoped client that cannot read the directory, the blob store, or another client's slots and queue." +--- + +Registration happens before the certificate is released. The relay cannot read sealed slots or calls. An executor must verify the retained request proof before it runs one. diff --git a/docs/relay-clients.md b/docs/relay-clients.md new file mode 100644 index 0000000000..6ef2b47ce9 --- /dev/null +++ b/docs/relay-clients.md @@ -0,0 +1,102 @@ +# Scoped clients, slots and queue + +## Authentication and pairing + +All routes use the existing `Codeaf-Identity`, `Codeaf-Cert`, `Codeaf-Time` +and `Codeaf-Sig` headers and `codeaf-req-v1` signing bytes (method, request +URI including query, request time, SHA-256 of the exact body). Identity namespaces +come exclusively from verified identity keys. Request skew is at most five minutes. +Errors are JSON `{"err":"code"}`. All replies include `Codeaf-Now` (Unix ms). +A store must be enabled. JSON bodies use UTF-8; sealed bytes use padded standard +base64. Unknown JSON fields and trailing JSON are rejected on fabric routes. + +A full device pairs a client with `PUT /v1/clients/{device}` and body +`{"cert":,"methods":["method.name"],"name":"sealed name"}`. +The certificate must verify under the request's identity public key and name the +path device. The client generates its own Ed25519 key; the full device signs only +its public key with the existing certificate format. **Register before releasing +the certificate to the client**: legacy certificates carry no cryptographic scope. +The pairing grant contains only the certificate, identity public key and public +connection information; never cell keys, dedup secrets, identity private keys or +vault data. The relay accepts no grant or secret fields on this route. + +Registration returns 204 and creates a directory device record with `kind: client` +and `methods`. At most 64 distinct method names, each 1–128 ASCII letters, digits, +underscore, dot or hyphen; an empty list is a read-only client. Repeat registration +returns 409 `exists`; scope changes require revocation and a new key. A full +existing device cannot be converted to a client. Old records without kind remain +full devices. `GET /v1/dir/list` by a full device includes the client records. +Clients cannot access directory, identity or blob-store routes (401 unauthorized), +including self-updates, approvals, leases and the vault. A client can read only its +own slots and append only its own queue. The method allowlist is checked at append; +the executing box must also check it before execution. + +`POST /v1/dir/devices/{device}/revoke` is the existing full-device revoke. +A successful revoke deletes that client's slots, queue and sequence state. Its +certificate stays revoked across restarts; requests return 401 `revoked`. + +## Slots + +`PUT /v1/clients/{device}/slots/{name}` by any full device of the same identity +replaces one slot. Body: `{"sealed":"base64"}`. The receiver must be an active +client. Names are 1–128 ASCII letters, digits, underscore, dot or hyphen. Maximum +64 slots per client and 65536 decoded bytes per slot (nonempty). +Success: 204. Writers are responsible for sealing to the current client key and +publishing only chats whose lease they hold; the relay cannot inspect sealed data. + +`GET /v1/clients/{device}/slots/{name}` by that client returns 200 +`{"sealed":"base64","writer":"dev_…","at":1700000000000}`. +The relay sets writer and Unix-ms time. Slots expire at age 24 hours, including +across restart, and expired slots free capacity. Full devices cannot read slots. + +## Queue + +`POST /v1/clients/{device}/queue` by that client accepts +`{"number":1,"at":1700000000000,"method":"method.name","target":"any","sealed":"base64"}`. +Numbers start at 1 and increase by exactly one per client, even after ack or +expiry. Integers must fit signed 64 bits. `at` is Unix ms, no more than five +minutes in the future and strictly younger than 24 hours. Each nonempty sealed +entry is at most 4096 decoded bytes; at most 128 unacknowledged entries per +client. Target is `any` or an active registered full device id in this identity. +The signed HTTP request authenticates all fields. Success: 204. Invalid numbers +(replay or gap) return 409 `sequence`; stale timestamps return 400 `stale`. +Denied methods return 403 `method_denied`. Expired entries free capacity but +never reset numbering. + +`GET /v1/clients/{device}/queue` by a full device returns 200 `{"entries":[]}`, +ordered by number, with fields above plus `signature`, `cert`, `identity`, +`request_time`, `request_uri`, and `body` (base64 exact original JSON bytes). +These preserve the original request proof for independent box verification. +Only entries targeted at the caller, unclaimed `any` entries, or `any` entries +claimed by the caller are listed. A list does not claim or change a chat lease. + +`POST /v1/clients/{device}/queue/{number}/claim` by a full device atomically +claims an `any` entry; success 204, same claimer repeat 204, another claimer +409 `claimed`. A targeted entry is claimable only by its target. Claims persist +until ack or expiry; the relay never automatically reassigns work after a crash. +An executor must claim before running. This prevents competing boxes executing +the same entry; durable executor-side idempotency is still required after a crash. + +`DELETE /v1/clients/{device}/queue/{number}` by the claim owner deletes the entry +and returns 204. Ack without owning the claim returns 409 `claimed`. Missing or +expired entries return 404 `not_found`. A box must recheck age, signature, method +scope, routing and any required biometric flag inside the sealed call, and log +execution. The relay cannot verify encrypted call contents or biometric evidence. + +## Refusals, persistence and implementations + +Malformed bodies, names, targets or empty sealed values: 400 `bad_request`. +Oversized HTTP bodies (maximum 96 KiB), decoded blobs or method lists: 413 +`too_large`. Slot/queue capacity: 429 `full`. Unknown client/slot/entry: 404 +`not_found`. Wrong role or identity: 401 `unauthorized`; revoked certificate: +401 `revoked`; request skew: 401 `skew`; replaced identity: 410 `rotated`; +deleted identity: 410 `gone`. Unsupported methods: 405. Internal storage failure: +500 `internal`. Nonempty bodies on bodyless routes: 400 `bad_request`. + +Records, sequence watermarks, proof and claims survive relay restarts. Fabric +mutations and revocation serialize within a relay process. One running Go relay +owns a store directory, as with its in-memory directory revocation feed. +Expiry is evaluated on reads and writes with the relay clock. Identity retirement +erases fabric data with the existing directory and blobs. + +The Go relay and the Worker implement this contract. diff --git a/internal/clusterstate/apply.go b/internal/clusterstate/apply.go index e6a11a6563..7ab4d63083 100644 --- a/internal/clusterstate/apply.go +++ b/internal/clusterstate/apply.go @@ -166,6 +166,9 @@ func Dependencies(c Command) []RecordKey { if m.PutDevice != nil { keys = append(keys, DeviceKey(m.PutDevice.ID)) } + if m.RegisterClient != nil { + keys = append(keys, DeviceKey(m.RegisterClient.Device), AuthorityKey(m.RegisterClient.Device)) + } if m.Revoke != nil { keys = append(keys, DeviceKey(m.Revoke.Device)) } @@ -412,6 +415,10 @@ func evaluateCommand(c Command, meta ApplicationMeta, read ReadRecord, paged boo out.Result.Code = Revoked return out, nil } + if actor.Kind == "client" { + out.Result.Code = Unauthorized + return out, nil + } // Worker incarnation and scope precede dedup so an old successful result // cannot restart activity after a process replacement, expiry or revocation. if auth.Worker != nil { @@ -517,7 +524,7 @@ func evaluateCommand(c Command, meta ApplicationMeta, read ReadRecord, paged boo if Decode(d, &seen) != nil { return out, ErrDependency } - if !seen.Revoked { + if !seen.Revoked && seen.Kind != "client" { seen.LastSeen = out.Now put(DeviceKey(c.Change.DeviceSeen.Device), seen) } @@ -741,9 +748,43 @@ func applyMutation(c Command, now int64, identity IdentityRec, rows map[RecordKe return OK case m.PutDevice != nil: v := m.PutDevice + if v.Device.Kind != "" || len(v.Device.Methods) > 0 { + return BadRequest + } old, had := device(v.ID) put(DeviceKey(v.ID), Stamped(old, had, v.Device, now)) return OK + case m.RegisterClient != nil: + v := m.RegisterClient + if !validClientDeviceID(v.Device) || len(v.Name) > MaxClientName || len(v.Methods) > MaxClientMethods { + return BadRequest + } + seen := map[string]bool{} + for _, method := range v.Methods { + if !validFabricName(method) || seen[method] { + return BadRequest + } + seen[method] = true + } + if _, had := device(v.Device); had || rows[AuthorityKey(v.Device)] != nil || v.Device == c.Actor { + return Exists + } + if PersonalCertificate(identity, v.Device, v.Cert) != nil { + return BadRequest + } + var methods []string + if len(v.Methods) > 0 { + methods = append([]string(nil), v.Methods...) + } + put(DeviceKey(v.Device), Device{ + V: 1, + Kind: "client", + Methods: methods, + Name: v.Name, + AddedBy: identity.Identity, + Created: now, + }) + return OK case m.Revoke != nil: v := m.Revoke old, had := device(v.Device) @@ -843,6 +884,9 @@ func applyMutation(c Command, now int64, identity IdentityRec, rows map[RecordKe if old.Revoked { return Revoked } + if old.Kind == "client" || v.Record.Kind != "" || len(v.Record.Methods) > 0 { + return BadRequest + } public, e := JoinCertificate(identity, request, v.Cert) if e != nil || validGrant(v.Grant) == false { return BadRequest diff --git a/internal/clusterstate/effects.go b/internal/clusterstate/effects.go index 324dc0bcfb..a4c02a3e44 100644 --- a/internal/clusterstate/effects.go +++ b/internal/clusterstate/effects.go @@ -121,6 +121,9 @@ func applyEffect(c Command, now int64, rows map[RecordKey][]byte, put func(Recor if d.Revoked { return Revoked } + if d.Kind == "client" { + return Unauthorized + } if Decode(rows[AuthorityKey(r.Device)], &auth) != nil || !auth.ValidRole() || auth.Cluster != c.Cluster || !auth.Personal && auth.Worker == nil { return Unauthorized } diff --git a/internal/clusterstate/liveness.go b/internal/clusterstate/liveness.go index c57e62458f..f4d44367cd 100644 --- a/internal/clusterstate/liveness.go +++ b/internal/clusterstate/liveness.go @@ -75,6 +75,9 @@ func renewHolds(c Command, now int64, rows map[RecordKey][]byte, put func(Record if actor.Revoked { return Revoked } + if actor.Kind == "client" { + return Unauthorized + } if auth.Worker != nil { var identity IdentityRec if Decode(rows[IdentityKey()], &identity) != nil { diff --git a/internal/clusterstate/personal_host.go b/internal/clusterstate/personal_host.go index b2c4747dca..18bff62e99 100644 --- a/internal/clusterstate/personal_host.go +++ b/internal/clusterstate/personal_host.go @@ -26,7 +26,7 @@ func personalAllowed(cluster, actor, certificate string, read ReadRecord) (Code, } if raw != nil { var d Device - if Decode(raw, &d) != nil || d.Revoked { + if Decode(raw, &d) != nil || d.Revoked || d.Kind == "client" { return Unauthorized, nil } } diff --git a/internal/clusterstate/records.go b/internal/clusterstate/records.go index 6d41447e7c..20d4eacc08 100644 --- a/internal/clusterstate/records.go +++ b/internal/clusterstate/records.go @@ -55,6 +55,12 @@ type Device struct { // Codeaf-Chat-Notes. Outbox []Note `json:"outbox,omitempty"` Took []string `json:"took,omitempty"` + + // Kind is "" for a full device and "client" for a scoped client that may + // only read its slots and append to its call queue. Methods lists the queue + // methods a scoped client may invoke. + Kind string `json:"kind,omitempty"` + Methods []string `json:"methods,omitempty"` } // Note is one message waiting in a device's outbox for whichever computer diff --git a/internal/clusterstate/rules.go b/internal/clusterstate/rules.go index 39de24f24f..22462e3d41 100644 --- a/internal/clusterstate/rules.go +++ b/internal/clusterstate/rules.go @@ -135,6 +135,7 @@ func RevokeOf(target Device, caller, targetID string) (Device, error) { // relay-owned fields are never taken from the client: Revoked and LastSeen stay // as stored, and Created is the time of the first write. func Stamped(old Device, had bool, d Device, now int64) Device { + d.Kind, d.Methods = old.Kind, old.Methods d.Revoked, d.LastSeen, d.Created = old.Revoked, old.LastSeen, old.Created if !had { d.Created = now diff --git a/internal/clusterstate/types.go b/internal/clusterstate/types.go index 629e80c1ea..cea2e3c6a8 100644 --- a/internal/clusterstate/types.go +++ b/internal/clusterstate/types.go @@ -81,6 +81,7 @@ type Mutation struct { Release *ReleaseCell `json:"release,omitempty"` Archive *CellID `json:"archive,omitempty"` PutDevice *PutDevice `json:"put_device,omitempty"` + RegisterClient *ClientRegistration `json:"register_client,omitempty"` Revoke *DeviceID `json:"revoke,omitempty"` SetVault *VaultCAS `json:"set_vault,omitempty"` AdvanceTime *AdvanceTime `json:"advance_time,omitempty"` @@ -143,6 +144,15 @@ type PutDevice struct { Device Device `json:"device"` } +// ClientRegistration records a scoped client device and its allowed queue +// methods in the replicated directory. Only a full device may submit it. +type ClientRegistration struct { + Device string `json:"device"` + Cert string `json:"cert"` + Name string `json:"name,omitempty"` + Methods []string `json:"methods,omitempty"` +} + type VaultCAS struct { Old string `json:"old"` New string `json:"new"` diff --git a/internal/clusterstate/v1.go b/internal/clusterstate/v1.go index 38a033c43e..51e205b7b4 100644 --- a/internal/clusterstate/v1.go +++ b/internal/clusterstate/v1.go @@ -34,7 +34,37 @@ type v1Certificate struct { Signature string `json:"sig"` } -const V1SkewMillis int64 = 5 * 60 * 1000 +const ( + V1SkewMillis int64 = 5 * 60 * 1000 + MaxClientMethods = 64 + MaxClientName = 256 +) + +func validClientDeviceID(s string) bool { + if len(s) != 36 || !strings.HasPrefix(s, "dev_") { + return false + } + for i := 4; i < len(s); i++ { + c := s[i] + if !(c >= '0' && c <= '9' || c >= 'a' && c <= 'f') { + return false + } + } + return true +} + +func validFabricName(s string) bool { + if len(s) == 0 || len(s) > 128 { + return false + } + for i := 0; i < len(s); i++ { + c := s[i] + if !(c >= 'A' && c <= 'Z' || c >= 'a' && c <= 'z' || c >= '0' && c <= '9' || c == '_' || c == '.' || c == '-') { + return false + } + } + return true +} // V1Command translates only the request the device actually signed. It never // reads a new holder to manufacture an expectation or signs on the device's behalf. @@ -262,6 +292,58 @@ func (p V1Request) Mutation() (Mutation, error) { } return Mutation{Rotate: &RotateRequest{Op: v.Op, GraceMS: v.GraceMS}}, nil } + if strings.HasPrefix(path, "/v1/clients/") { + s := strings.Split(strings.TrimPrefix(path, "/v1/clients/"), "/") + if len(s) != 1 || p.Method != "PUT" { + return bad() + } + id, err := v1Segment(s[0]) + if err != nil || !validClientDeviceID(id) { + return bad() + } + var in struct { + Cert *v1Certificate `json:"cert"` + Methods []string `json:"methods"` + Name string `json:"name,omitempty"` + } + if !decode(&in, false) || in.Cert == nil || in.Methods == nil || len(in.Methods) > MaxClientMethods || len(in.Name) > MaxClientName { + return bad() + } + seen := map[string]bool{} + for _, m := range in.Methods { + if !validFabricName(m) || seen[m] { + return bad() + } + seen[m] = true + } + certRaw, err := json.Marshal(in.Cert) + if err != nil { + return bad() + } + cert := base64.RawURLEncoding.EncodeToString(certRaw) + if len(p.IdentityKey) == ed25519.PublicKeySize { + ident := IdentityRec{V: 1, Identity: IdentityID(p.IdentityKey), SigningKey: p.IdentityKey} + if PersonalCertificate(ident, id, cert) != nil { + return bad() + } + } else { + pub, pubErr := hex.DecodeString(in.Cert.Device) + sig, sigErr := hex.DecodeString(in.Cert.Signature) + if pubErr != nil || sigErr != nil || len(pub) != ed25519.PublicKeySize || len(sig) != ed25519.SignatureSize || in.Cert.V != 1 || in.Cert.Created < 0 || in.Cert.Created > int64(MaxSafeInteger) || DeviceIDOfKey(pub) != id { + return bad() + } + } + var methods []string + if len(in.Methods) > 0 { + methods = append([]string(nil), in.Methods...) + } + return Mutation{RegisterClient: &ClientRegistration{ + Device: id, + Cert: cert, + Name: in.Name, + Methods: methods, + }}, nil + } s := strings.Split(strings.TrimPrefix(path, "/v1/dir/"), "/") if len(s) < 2 || len(s) > 3 || path == strings.TrimPrefix(path, "/v1/dir/") { return bad() @@ -277,7 +359,7 @@ func (p V1Request) Mutation() (Mutation, error) { if s[0] == "devices" { if p.Method == "PUT" && verb == "" { var v Device - if !decode(&v, false) { + if !decode(&v, false) || v.Kind != "" || len(v.Methods) > 0 { return bad() } return Mutation{PutDevice: &PutDevice{ID: id, Device: v}}, nil @@ -297,7 +379,7 @@ func (p V1Request) Mutation() (Mutation, error) { Cert string `json:"cert"` Grant string `json:"grant"` } - if !decode(&a, false) { + if !decode(&a, false) || a.Device.Kind != "" || len(a.Device.Methods) > 0 { return bad() } var cert v1Certificate @@ -405,6 +487,11 @@ func normalizeV1Mutation(m Mutation) Mutation { v.Publish.FrameHints = boundedFrameHints(v.Publish.FrameHints) m.Publish = &v } + if m.RegisterClient != nil && m.RegisterClient.Methods != nil && len(m.RegisterClient.Methods) == 0 { + v := *m.RegisterClient + v.Methods = nil + m.RegisterClient = &v + } return m } diff --git a/internal/clusterstate/v1_stage.go b/internal/clusterstate/v1_stage.go index 5025129751..59436215b8 100644 --- a/internal/clusterstate/v1_stage.go +++ b/internal/clusterstate/v1_stage.go @@ -182,6 +182,9 @@ func evaluateV1Stage(c Command, meta ApplicationMeta, read ReadRecord) (Evaluati out.Result.Code = Revoked return out, nil } + if device.Kind == "client" { + return out, nil + } } path, err := c.V1.route() if err != nil || !strings.HasPrefix(path, "/v1/") { diff --git a/internal/clusterstate/v1_test.go b/internal/clusterstate/v1_test.go index f06789bec6..95aa7b7023 100644 --- a/internal/clusterstate/v1_test.go +++ b/internal/clusterstate/v1_test.go @@ -2,7 +2,10 @@ package clusterstate_test import ( "crypto/ed25519" + "encoding/base64" + "reflect" "strconv" + "strings" "testing" cs "github.com/Agent-Field/codeaf/internal/clusterstate" @@ -41,3 +44,137 @@ func TestR25V1EmptyFramePlansKeepOriginalAuthority(t *testing.T) { } } } + +func TestV1RegisterClientAndScopedClientRefusal(t *testing.T) { + a := newApplication(t) + root := actorKey("root").Public().(ed25519.PublicKey) + fullPriv := actorKey("full") + fullID := cs.DeviceIDOfKey(fullPriv.Public().(ed25519.PublicKey)) + fullCert := joinCert(t, "full") + + clientPriv := actorKey("client") + clientID := cs.DeviceIDOfKey(clientPriv.Public().(ed25519.PublicKey)) + clientCertB64 := joinCert(t, "client") + clientCertJSON, err := base64.RawURLEncoding.DecodeString(clientCertB64) + if err != nil { + t.Fatal(err) + } + + v1Eval := func(priv ed25519.PrivateKey, cert, method, uri string, body []byte, now int64) cs.Result { + t.Helper() + proof := cs.V1Request{ + Method: method, + URI: uri, + IdentityKey: root, + Certificate: cert, + At: strconv.FormatInt(now, 10), + Body: body, + } + proof.Signature = ed25519.Sign(priv, cs.V1SigningMessage(proof.Method, proof.URI, proof.At, body)) + cmd, err := cs.V1Command("test", now, proof) + if err != nil { + t.Fatalf("V1Command(%s %s): %v", method, uri, err) + } + return a.evaluate(cmd).Result + } + + // Self-registration via PUT /v1/dir/devices/{id} must refuse kind or methods. + for _, badBody := range []string{ + `{"V":1,"name":"n","added_by":"","revoked":false,"caps":{"os":"linux","arch":"amd64","sandbox":null,"container":null,"gpu":null,"cow":"none"},"kind":"client"}`, + `{"V":1,"name":"n","added_by":"","revoked":false,"caps":{"os":"linux","arch":"amd64","sandbox":null,"container":null,"gpu":null,"cow":"none"},"methods":["chat.read"]}`, + } { + proof := cs.V1Request{Method: "PUT", URI: "/v1/dir/devices/" + fullID, IdentityKey: root, Certificate: fullCert, At: "1000", Body: []byte(badBody)} + proof.Signature = ed25519.Sign(fullPriv, cs.V1SigningMessage(proof.Method, proof.URI, proof.At, proof.Body)) + if _, err := cs.V1Command("test", 1000, proof); err == nil { + t.Fatalf("expected PUT /v1/dir/devices with kind/methods to fail V1Command, got nil") + } + } + if r := a.run("a", 1000, cs.Mutation{PutDevice: &cs.PutDevice{ID: "a", Device: cs.Device{V: 1, Kind: "client"}}}); r.Code != cs.BadRequest { + t.Fatalf("PutDevice with Kind=client: got %s, want bad_request", r.Code) + } + if r := a.run("a", 1001, cs.Mutation{PutDevice: &cs.PutDevice{ID: "a", Device: cs.Device{V: 1, Methods: []string{"m"}}}}); r.Code != cs.BadRequest { + t.Fatalf("PutDevice with Methods: got %s, want bad_request", r.Code) + } + + // Stamped must preserve stored Kind and Methods and ignore caller values. + stamped := cs.Stamped(cs.Device{V: 1, Kind: "client", Methods: []string{"chat.read"}}, true, cs.Device{V: 1, Kind: "", Methods: []string{"evil"}}, 2000) + if stamped.Kind != "client" || !reflect.DeepEqual(stamped.Methods, []string{"chat.read"}) { + t.Fatalf("Stamped did not preserve Kind/Methods: %+v", stamped) + } + + // Invalid PUT /v1/clients/{device} payloads must fail V1Command. + for _, tc := range []struct { + uri string + body string + }{ + {"/v1/clients/not-a-device", `{"cert":` + string(clientCertJSON) + `,"methods":[]}`}, + {"/v1/clients/" + fullID, `{"cert":` + string(clientCertJSON) + `,"methods":[]}`}, + {"/v1/clients/" + clientID, `{"cert":` + string(clientCertJSON) + `,"methods":["bad method"]}`}, + {"/v1/clients/" + clientID, `{"cert":` + string(clientCertJSON) + `,"methods":["dup","dup"]}`}, + {"/v1/clients/" + clientID, `{"cert":` + string(clientCertJSON) + `,"methods":[],"name":"` + strings.Repeat("n", cs.MaxClientName+1) + `"}`}, + } { + proof := cs.V1Request{Method: "PUT", URI: tc.uri, IdentityKey: root, Certificate: fullCert, At: "1002", Body: []byte(tc.body)} + proof.Signature = ed25519.Sign(fullPriv, cs.V1SigningMessage(proof.Method, proof.URI, proof.At, proof.Body)) + if _, err := cs.V1Command("test", 1002, proof); err == nil { + t.Fatalf("expected invalid client registration (%s) to fail V1Command", tc.uri) + } + } + + // A full device cannot register itself as a client. + fullCertJSON, _ := base64.RawURLEncoding.DecodeString(fullCert) + if r := v1Eval(fullPriv, fullCert, "PUT", "/v1/clients/"+fullID, []byte(`{"cert":`+string(fullCertJSON)+`,"methods":[]}`), 1003); r.Code != cs.Exists { + t.Fatalf("self client registration: got %s, want exists", r.Code) + } + + // Valid client registration succeeds and commits Kind="client" and Methods. + regBody := []byte(`{"cert":` + string(clientCertJSON) + `,"methods":["chat.read","chat.send"],"name":"sealed-phone"}`) + if r := v1Eval(fullPriv, fullCert, "PUT", "/v1/clients/"+clientID, regBody, 1004); r.Code != cs.OK { + t.Fatalf("RegisterClient: got %s, want ok", r.Code) + } + var stored cs.Device + if err := cs.Decode(a.rows[cs.DeviceKey(clientID)], &stored); err != nil { + t.Fatal(err) + } + if stored.Kind != "client" || !reflect.DeepEqual(stored.Methods, []string{"chat.read", "chat.send"}) || stored.Name != "sealed-phone" || stored.AddedBy != cs.IdentityID(root) || stored.Created != 1004 { + t.Fatalf("unexpected stored client device: %+v", stored) + } + + // Re-registering the same client returns Exists. + if r := v1Eval(fullPriv, fullCert, "PUT", "/v1/clients/"+clientID, regBody, 1005); r.Code != cs.Exists { + t.Fatalf("duplicate RegisterClient: got %s, want exists", r.Code) + } + + // A scoped client is refused every directory mutation (both v1 and v2). + otherPriv := actorKey("other") + otherID := cs.DeviceIDOfKey(otherPriv.Public().(ed25519.PublicKey)) + otherCertJSON, _ := base64.RawURLEncoding.DecodeString(joinCert(t, "other")) + for _, tc := range []struct { + method string + uri string + body string + }{ + {"PUT", "/v1/dir/devices/" + clientID, `{"V":1,"name":"n","added_by":"","revoked":false,"caps":{"os":"linux","arch":"amd64","sandbox":null,"container":null,"gpu":null,"cow":"none"}}`}, + {"POST", "/v1/dir/vault", `{"old":"","new":"v1"}`}, + {"POST", "/v1/dir/cells/chat", `{"head":"h"}`}, + {"POST", "/v1/dir/devices/" + fullID + "/revoke", ``}, + {"PUT", "/v1/clients/" + otherID, `{"cert":` + string(otherCertJSON) + `,"methods":[]}`}, + } { + if r := v1Eval(clientPriv, clientCertB64, tc.method, tc.uri, []byte(tc.body), 1006); r.Code != cs.Unauthorized { + t.Fatalf("scoped client %s %s: got %s, want unauthorized", tc.method, tc.uri, r.Code) + } + } + + // Revoking the client marks Revoked=true while keeping Kind="client", and re-registration is still Exists. + if r := v1Eval(fullPriv, fullCert, "POST", "/v1/dir/devices/"+clientID+"/revoke", nil, 1007); r.Code != cs.OK { + t.Fatalf("revoke client: got %s, want ok", r.Code) + } + if err := cs.Decode(a.rows[cs.DeviceKey(clientID)], &stored); err != nil { + t.Fatal(err) + } + if !stored.Revoked || stored.Kind != "client" { + t.Fatalf("revoked client record: %+v", stored) + } + if r := v1Eval(fullPriv, fullCert, "PUT", "/v1/clients/"+clientID, regBody, 1008); r.Code != cs.Exists { + t.Fatalf("re-register revoked client: got %s, want exists", r.Code) + } +} diff --git a/internal/directory/handler.go b/internal/directory/handler.go index ff8e74eff0..862a383034 100644 --- a/internal/directory/handler.go +++ b/internal/directory/handler.go @@ -205,6 +205,9 @@ func putDevice(c call, d Device) (any, error) { if c.id() != c.device { return nil, ErrUnauthorized } + if d.Kind != "" || len(d.Methods) > 0 { + return nil, errBadRequest + } return nothing(c.cl.PutDevice(c.ctx(), c.id(), d)) } diff --git a/internal/directory/records.go b/internal/directory/records.go index b05c701fcc..43a151bf46 100644 --- a/internal/directory/records.go +++ b/internal/directory/records.go @@ -39,6 +39,10 @@ import "github.com/Agent-Field/codeaf/internal/clusterstate" // remain usable without linking the directory transports and stores. type IdentityRec = clusterstate.IdentityRec type Caps = clusterstate.Caps + +// Device is one device's row. Name and Outbox are opaque sealed bytes; the +// relay checks only their size. Kind is "" for a full device and "client" for +// a scoped client that may only read its slots and append to its call queue. type Device = clusterstate.Device type Lease = clusterstate.Lease diff --git a/internal/manual/chat/relay-what-it-stores-and-sees.md b/internal/manual/chat/relay-what-it-stores-and-sees.md index 0eaf67d8b3..783265cf6d 100644 --- a/internal/manual/chat/relay-what-it-stores-and-sees.md +++ b/internal/manual/chat/relay-what-it-stores-and-sees.md @@ -69,3 +69,43 @@ replaces them, and a publish without hints clears the previous hints. The prototype is off by default. Every member of the relay cluster must run a build that understands these fields before any computer sends them. Older members refuse unknown fields; existing requests omit the fields and keep their encoding. + +## Scoped clients, slots and queued calls + +The Go relay supports a client device kind with its own key pair and a list of +allowed methods. Full devices register the client before handing it its signed +certificate. A client receives no cell key, dedup secret, identity private key or +vault. It cannot read the directory or the blob store, change leases, pair another +device or change its own permissions. Revocation stops that certificate across +relay restarts. This is a relay protocol capability; the chat pairing commands do +not yet create scoped clients. The hosted Worker supports the same protocol. + +The HTTP contract is `docs/relay-clients.md` in the source tree. An empty method +list allows offline reads only; adding methods requires a new paired key. + +## Client slots and offline snapshots + +A full device can replace a named sealed slot for one scoped client of the same +identity. Only that client reads the slot. The relay adds the writer device and +write time; readers can show when a snapshot was last refreshed. Each client has +at most 64 slots, each at most 64 KiB, expiring after 24 hours. Revoking the client +deletes its slots. The relay sees ciphertext and slot names, so names must not +contain private content. Slots persist across relay restarts. The publishing box +must seal to the client's current key and respect chat lease ownership. + +## Client queue, offline calls and replay protection + +Scoped clients can append sealed calls for an allowed method. Each call carries +a signed sequence number, time and target box, or `any`. Numbers must increase +by one, including after acknowledgement or expiry. Each client can hold 128 +unacknowledged calls, each at most 4 KiB; calls expire at 24 hours. The relay +rejects replays, gaps, stale calls and methods outside the pairing's allowlist. +A client with no allowed methods cannot queue calls. + +Full devices list work and claim an entry before executing it. An `any` claim +has one winner and survives restart; another box cannot acknowledge it. The +winning box acknowledges to delete the entry. Queue operations never acquire a +chat lease. Revoking a client deletes its queue. An executor must verify the +retained original request proof, permissions, age and any required biometric +confirmation inside the encrypted call, use durable execution idempotency and +log executed calls. The relay cannot read the call or verify biometric evidence. diff --git a/internal/raftabi/hosted.go b/internal/raftabi/hosted.go index 0f523b3da6..1b1dda5096 100644 --- a/internal/raftabi/hosted.go +++ b/internal/raftabi/hosted.go @@ -132,9 +132,11 @@ func legacyProjectionJSON(value any) ([]byte, error) { // revocation and rotation; socket presence supplies none of that authority. func hostedAuthority(d *decoder) ([]byte, error) { identity, device, actor, cert, write := d.blob(), d.blob(), d.str(), d.str(), d.u32() - if d.end() != nil || write > 1 { + if d.end() != nil || write > 3 { return nil, errFrame } + isWrite := write&1 != 0 + allowClient := write&2 != 0 var root cs.IdentityRec var rec cs.Device if cs.Decode(identity, &root) != nil || len(device) > 0 && cs.Decode(device, &rec) != nil { @@ -145,7 +147,9 @@ func hostedAuthority(d *decoder) ([]byte, error) { code = cs.Unauthorized } else if rec.Revoked { code = cs.Revoked - } else if write == 1 && root.Rotation != nil { + } else if rec.Kind == "client" && !allowClient { + code = cs.Unauthorized + } else if isWrite && root.Rotation != nil { code = cs.Rotated } e := encoder{} diff --git a/internal/raftdirectory/v1.go b/internal/raftdirectory/v1.go index c235af1c65..e72c2d393c 100644 --- a/internal/raftdirectory/v1.go +++ b/internal/raftdirectory/v1.go @@ -121,6 +121,9 @@ func authorizeV1Read(ctx context.Context, s *raftstore.SQLite, app raftstore.App if device.Revoked { return directory.ErrRevoked } + if device.Kind == "client" { + return directory.ErrUnauthorized + } } return nil } diff --git a/internal/relayconf/fabric_test.go b/internal/relayconf/fabric_test.go new file mode 100644 index 0000000000..be8fd14d36 --- /dev/null +++ b/internal/relayconf/fabric_test.go @@ -0,0 +1,16 @@ +//go:build relayurl + +package relayconf + +import ( + "testing" + + "github.com/Agent-Field/codeaf/internal/relayserve/fabrictest" +) + +func TestFabricConformance(t *testing.T) { + base := baseURL(t) + fabrictest.Run(t, func(t *testing.T) *fabrictest.Rig { + return &fabrictest.Rig{URL: func() string { return base }, Client: httpClient, Now: wall} + }) +} diff --git a/internal/relayserve/fabric.go b/internal/relayserve/fabric.go new file mode 100644 index 0000000000..ff879daa46 --- /dev/null +++ b/internal/relayserve/fabric.go @@ -0,0 +1,257 @@ +package relayserve + +import ( + "bytes" + "crypto/ed25519" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "io" + "net/http" + "regexp" + "strings" + "sync" + + "github.com/Agent-Field/codeaf/internal/directory" + "github.com/Agent-Field/codeaf/internal/identity" + "github.com/Agent-Field/codeaf/internal/reqsign" + "github.com/Agent-Field/codeaf/internal/wireauth" +) + +// Fabric limits bound the opaque data held for each client. +const ( + MaxFabricBody = 96 << 10 + MaxClientMethods = 64 +) + +var fabricName = regexp.MustCompile(`^[A-Za-z0-9_.-]{1,128}$`) +var fabricDevice = regexp.MustCompile(`^dev_[0-9a-f]{32}$`) + +// fabric serializes client registration, data mutations and directory writes. +// A store belongs to one relay process, like the directory's revocation feed. +type fabric struct { + ns *namespaces + auth wireauth.Authenticate + mu sync.Mutex +} + +type fabricError struct { + status int + code string +} + +func (e fabricError) Error() string { return e.code } +func badFabric(status int, code string) error { return fabricError{status, code} } + +func fabricReply(w http.ResponseWriter, out any, err error) { + if err != nil { + e := fabricError{500, "internal"} + if !errors.As(err, &e) { + switch { + case errors.Is(err, wireauth.ErrRevoked), errors.Is(err, directory.ErrRevoked): + e = fabricError{401, "revoked"} + case errors.Is(err, wireauth.ErrSkew): + e = fabricError{401, "skew"} + case errors.Is(err, wireauth.ErrUnauthorized), errors.Is(err, directory.ErrUnauthorized): + e = fabricError{401, "unauthorized"} + case errors.Is(err, wireauth.ErrGone): + e = fabricError{410, "gone"} + case errors.Is(err, directory.ErrNotFound): + e = fabricError{404, "not_found"} + case errors.Is(err, directory.ErrExists): + e = fabricError{409, "exists"} + case errors.Is(err, directory.ErrSelfRevoke): + e = fabricError{400, "self_revoke"} + case errors.Is(err, directory.ErrBadRequest): + e = fabricError{400, "bad_request"} + case errors.Is(err, directory.ErrFull): + e = fabricError{507, "full"} + case errors.Is(err, wireauth.ErrRotated), errors.Is(err, directory.ErrRotated): + e = fabricError{410, "rotated"} + case errors.Is(err, wireauth.ErrUnavailable), errors.Is(err, directory.ErrUnreachable): + e = fabricError{503, "unreachable"} + } + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(e.status) + _ = json.NewEncoder(w).Encode(map[string]string{"err": e.code}) + return + } + if out == nil { + w.WriteHeader(http.StatusNoContent) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(out) +} + +func decodeFabric(body []byte, out any) error { + dec := json.NewDecoder(bytes.NewReader(body)) + dec.DisallowUnknownFields() + if dec.Decode(out) != nil { + return badFabric(400, "bad_request") + } + if dec.Decode(new(any)) != io.EOF { + return badFabric(400, "bad_request") + } + return nil +} + +// directoryHandler keeps revocation and a fabric mutation from crossing. +// Watch upgrades stay outside the lock because they can last for hours. +func (f *fabric) directoryHandler(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + next.ServeHTTP(w, r) + return + } + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, directory.MaxBody)) + if err != nil { + fabricReply(w, nil, badFabric(413, "too_large")) + return + } + r.Body = io.NopCloser(bytes.NewReader(body)) + reply := &fabricResponse{headers: make(http.Header)} + func() { f.mu.Lock(); defer f.mu.Unlock(); next.ServeHTTP(reply, r) }() + for key, values := range reply.headers { + w.Header()[key] = values + } + if reply.status != 0 { + w.WriteHeader(reply.status) + } + if reply.body.Len() != 0 { + _, _ = w.Write(reply.body.Bytes()) + } + }) +} + +// fabricResponse keeps network writes outside the mutation lock. Directory +// writes have bounded JSON answers; streaming watch replies never pass here. +type fabricResponse struct { + headers http.Header + status int + body bytes.Buffer +} + +func (w *fabricResponse) Header() http.Header { return w.headers } +func (w *fabricResponse) WriteHeader(status int) { + if w.status == 0 { + w.status = status + } +} +func (w *fabricResponse) Write(p []byte) (int, error) { + if w.status == 0 { + w.status = 200 + } + return w.body.Write(p) +} + +type fabricCall struct { + f *fabric + r *http.Request + body []byte + id, dev string + ns *nativeNamespace + listing directory.Listing +} + +type fabricRoute func(fabricCall) (any, error) + +func (f *fabric) handler() http.Handler { + mux := http.NewServeMux() + f.mountQueue(mux) + mux.HandleFunc("PUT /v1/clients/{device}", f.serve(registerClient)) + mux.HandleFunc("PUT /v1/clients/{device}/slots/{name}", f.serve(putSlot)) + mux.HandleFunc("GET /v1/clients/{device}/slots/{name}", f.serve(getSlot)) + return mux +} + +func (f *fabric) serve(route fabricRoute) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + isDir := strings.HasPrefix(r.URL.Path, "/v1/dir/") + if isDir { + w.Header().Set(directory.SavedTurnTimeHeader, "1") + w.Header().Set(directory.HeadTakeHeader, "1") + w.Header().Set(directory.UploadProgressHeader, "1") + w.Header().Set(directory.WorkingBeatsHeader, "1") + w.Header().Set(directory.ChatNotesHeader, "1") + w.Header().Set(directory.ChatFoldersHeader, "1") + } + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, MaxFabricBody)) + if err != nil { + fabricReply(w, nil, badFabric(413, "too_large")) + return + } + out, err := func() (any, error) { + f.mu.Lock() + defer f.mu.Unlock() + c, err := f.admit(r, body) + if err != nil { + return nil, err + } + return route(c) + }() + if isDir && err == nil { + w.Header().Set(directory.ConditionalTakeHeader, "1") + } + fabricReply(w, out, err) + } +} + +func (f *fabric) admit(r *http.Request, body []byte) (fabricCall, error) { + c := fabricCall{f: f, r: r, body: body} + var err error + c.id, c.dev, err = f.auth(r, body) + if err != nil { + return c, err + } + if !fabricDevice.MatchString(r.PathValue("device")) { + return c, badFabric(400, "bad_request") + } + c.ns, err = f.ns.activate(r.Context(), c.id, nil) + if err != nil { + return c, err + } + c.listing, err = c.ns.fabricListing(r.Context()) + return c, err +} + +func registerClient(c fabricCall) (any, error) { + if c.listing.Devices[c.dev].Kind == "client" { + return nil, wireauth.ErrUnauthorized + } + if c.listing.Identity.Rotation != nil { + return nil, wireauth.ErrRotated + } + var in struct { + Cert identity.Cert `json:"cert"` + Methods []string `json:"methods"` + Name string `json:"name"` + } + if err := decodeFabric(c.body, &in); err != nil { + return nil, err + } + if len(in.Methods) > MaxClientMethods || len(in.Name) > 256 { + return nil, badFabric(413, "too_large") + } + seen := map[string]bool{} + for _, method := range in.Methods { + if !fabricName.MatchString(method) || seen[method] { + return nil, badFabric(400, "bad_request") + } + seen[method] = true + } + pub, keyErr := hex.DecodeString(in.Cert.Device) + if keyErr != nil || len(pub) != ed25519.PublicKeySize { + return nil, badFabric(400, "bad_request") + } + key, err := base64.RawURLEncoding.DecodeString(c.r.Header.Get(reqsign.HeaderIdentity)) + if err != nil || len(key) != ed25519.PublicKeySize || in.Cert.Verify(key) != nil || in.Cert.DeviceID() != c.r.PathValue("device") { + return nil, badFabric(400, "bad_request") + } + if _, exists := c.listing.Devices[in.Cert.DeviceID()]; exists { + return nil, badFabric(409, "exists") + } + return nil, c.ns.commitFabricV1(c.r.Context(), c.dev, c.r, c.body) +} diff --git a/internal/relayserve/fabric_test.go b/internal/relayserve/fabric_test.go new file mode 100644 index 0000000000..ee7fbb747c --- /dev/null +++ b/internal/relayserve/fabric_test.go @@ -0,0 +1,74 @@ +package relayserve_test + +import ( + "github.com/Agent-Field/codeaf/internal/directory" + "github.com/Agent-Field/codeaf/internal/relayserve/fabrictest" + "github.com/Agent-Field/codeaf/internal/reqsign" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestFabricConformance(t *testing.T) { + fabrictest.Run(t, func(t *testing.T) *fabrictest.Rig { + r := newRig(t) + return &fabrictest.Rig{URL: func() string { return r.srv.URL }, Client: r.srv.Client(), Now: r.clock.Now, Advance: r.clock.Advance, Restart: r.restart} + }) +} + +// A body that has not arrived must not prevent another device from revoking. +func TestFabricSlowBodyDoesNotBlockRevocation(t *testing.T) { + for _, route := range []struct{ method, path string }{{"PUT", "/v1/clients/dev_00000000000000000000000000000000"}, {"POST", "/v1/dir/vault"}} { + t.Run(route.path, func(t *testing.T) { + r := newRig(t) + full, client, _ := twoOfOne(t, r) + body := &pausedBody{entered: make(chan struct{}), release: make(chan struct{})} + req := httptest.NewRequest(route.method, route.path, body) + done := make(chan struct{}) + go func() { defer close(done); r.svc.Handler.ServeHTTP(httptest.NewRecorder(), req) }() + <-body.entered + defer func() { close(body.release); <-done }() + hc := &http.Client{Timeout: 2 * time.Second} + dir := directory.NewHTTP(r.srv.URL, reqsign.SignFor(full, r.clock.Now), hc) + if err := dir.Revoke(bg, client.dev.ID()); err != nil { + t.Fatalf("revoke while anonymous body is unfinished: %v", err) + } + }) + } +} + +type pausedBody struct{ entered, release chan struct{} } + +func (b *pausedBody) Read([]byte) (int, error) { close(b.entered); <-b.release; return 0, io.EOF } +func (*pausedBody) Close() error { return nil } + +func TestFabricSlowResponseDoesNotBlockRevocation(t *testing.T) { + r := newRig(t) + full, client, _ := twoOfOne(t, r) + req := httptest.NewRequest("PUT", "/v1/clients/"+client.dev.ID(), nil) + // An invalid request still produces a JSON refusal that a reader can stall. + writer := &pausedResponse{header: make(http.Header), entered: make(chan struct{}), release: make(chan struct{})} + done := make(chan struct{}) + go func() { defer close(done); r.svc.Handler.ServeHTTP(writer, req) }() + <-writer.entered + defer func() { close(writer.release); <-done }() + dir := directory.NewHTTP(r.srv.URL, reqsign.SignFor(full, r.clock.Now), &http.Client{Timeout: 2 * time.Second}) + if err := dir.Revoke(bg, client.dev.ID()); err != nil { + t.Fatalf("revoke while response reader is stalled: %v", err) + } +} + +type pausedResponse struct { + header http.Header + entered, release chan struct{} +} + +func (w *pausedResponse) Header() http.Header { return w.header } +func (*pausedResponse) WriteHeader(int) {} +func (w *pausedResponse) Write(p []byte) (int, error) { + close(w.entered) + <-w.release + return len(p), nil +} diff --git a/internal/relayserve/fabricqueue.go b/internal/relayserve/fabricqueue.go new file mode 100644 index 0000000000..da9ee51a02 --- /dev/null +++ b/internal/relayserve/fabricqueue.go @@ -0,0 +1,152 @@ +package relayserve + +import ( + "net/http" + "slices" + "strconv" + + "github.com/Agent-Field/codeaf/internal/reqsign" + "github.com/Agent-Field/codeaf/internal/wireauth" +) + +// Queue limits bound both the number of calls and each sealed payload. +const ( + MaxQueueBytes = 4096 + MaxClientQueue = 128 +) + +type queuedCall struct { + Number int64 `json:"number"` + At int64 `json:"at"` + Method string `json:"method"` + Target string `json:"target"` + Sealed []byte `json:"sealed"` +} + +type queueEntry struct { + queuedCall + ClaimedBy string `json:"claimed_by,omitempty"` + Signature string `json:"signature"` + Cert string `json:"cert"` + Identity string `json:"identity"` + RequestTime string `json:"request_time"` + RequestURI string `json:"request_uri"` + Body []byte `json:"body"` +} + +func appendQueue(c fabricCall) (any, error) { + if err := c.target(true); err != nil { + return nil, err + } + var in queuedCall + if err := decodeFabric(c.body, &in); err != nil { + return nil, err + } + if err := c.validateEntry(in); err != nil { + return nil, err + } + s, err := c.load() + if err != nil { + return nil, err + } + if in.Number <= 0 || in.Number != s.LastNumber+1 { + return nil, badFabric(409, "sequence") + } + if len(s.Queue) >= MaxClientQueue { + return nil, badFabric(429, "full") + } + h := c.r.Header + s.Queue = append(s.Queue, queueEntry{queuedCall: in, Signature: h.Get(reqsign.HeaderSig), Cert: h.Get(reqsign.HeaderCert), Identity: h.Get(reqsign.HeaderIdentity), RequestTime: h.Get(reqsign.HeaderTime), RequestURI: c.r.URL.RequestURI(), Body: c.body}) + s.LastNumber = in.Number + return nil, c.save(s) +} + +func (c fabricCall) validateEntry(in queuedCall) error { + if len(in.Sealed) > MaxQueueBytes { + return badFabric(413, "too_large") + } + if len(in.Sealed) == 0 || !fabricName.MatchString(in.Method) { + return badFabric(400, "bad_request") + } + if !slices.Contains(c.listing.Devices[c.dev].Methods, in.Method) { + return badFabric(403, "method_denied") + } + if in.Target != "any" { + d, ok := c.listing.Devices[in.Target] + if !fabricDevice.MatchString(in.Target) || !ok || d.Kind == "client" || d.Revoked { + return badFabric(400, "bad_request") + } + } + now := c.f.ns.now().UnixMilli() + if in.At <= now-FabricMaxAge.Milliseconds() || in.At > now+reqsign.Skew.Milliseconds() { + return badFabric(400, "stale") + } + return nil +} + +func listQueue(c fabricCall) (any, error) { + if err := c.target(false); err != nil { + return nil, err + } + if len(c.body) != 0 { + return nil, badFabric(400, "bad_request") + } + s, err := c.load() + if err != nil { + return nil, err + } + entries := []queueEntry{} + for _, e := range s.Queue { + if (e.Target == "any" || e.Target == c.dev) && (e.ClaimedBy == "" || e.ClaimedBy == c.dev) { + entries = append(entries, e) + } + } + return struct { + Entries []queueEntry `json:"entries"` + }{entries}, nil +} + +func claimQueue(c fabricCall) (any, error) { return changeQueue(c, false) } +func ackQueue(c fabricCall) (any, error) { return changeQueue(c, true) } + +func changeQueue(c fabricCall, ack bool) (any, error) { + if err := c.target(false); err != nil { + return nil, err + } + if len(c.body) != 0 { + return nil, badFabric(400, "bad_request") + } + text := c.r.PathValue("number") + number, err := strconv.ParseInt(text, 10, 64) + if err != nil || number <= 0 || strconv.FormatInt(number, 10) != text { + return nil, badFabric(400, "bad_request") + } + s, err := c.load() + if err != nil { + return nil, err + } + index := slices.IndexFunc(s.Queue, func(e queueEntry) bool { return e.Number == number }) + if index < 0 { + return nil, badFabric(404, "not_found") + } + e := &s.Queue[index] + if e.Target != "any" && e.Target != c.dev { + return nil, wireauth.ErrUnauthorized + } + if (e.ClaimedBy != "" && e.ClaimedBy != c.dev) || (ack && e.ClaimedBy != c.dev) { + return nil, badFabric(409, "claimed") + } + if ack { + s.Queue = slices.Delete(s.Queue, index, index+1) + } else { + e.ClaimedBy = c.dev + } + return nil, c.save(s) +} + +func (f *fabric) mountQueue(mux *http.ServeMux) { + mux.HandleFunc("POST /v1/clients/{device}/queue", f.serve(appendQueue)) + mux.HandleFunc("GET /v1/clients/{device}/queue", f.serve(listQueue)) + mux.HandleFunc("POST /v1/clients/{device}/queue/{number}/claim", f.serve(claimQueue)) + mux.HandleFunc("DELETE /v1/clients/{device}/queue/{number}", f.serve(ackQueue)) +} diff --git a/internal/relayserve/fabricslots.go b/internal/relayserve/fabricslots.go new file mode 100644 index 0000000000..bc41b6e38c --- /dev/null +++ b/internal/relayserve/fabricslots.go @@ -0,0 +1,210 @@ +package relayserve + +import ( + "context" + "encoding/json" + "errors" + "os" + "path/filepath" + "slices" + "strings" + "time" + + "github.com/Agent-Field/codeaf/internal/fsreplace" + "github.com/Agent-Field/codeaf/internal/wireauth" +) + +// Slot limits are the same for every client and implementation. +const ( + MaxSlotBytes = 64 << 10 + MaxClientSlots = 64 + FabricMaxAge = 24 * time.Hour +) + +type fabricSlot struct { + Sealed []byte `json:"sealed"` + Writer string `json:"writer"` + At int64 `json:"at"` +} + +// clientState is replaced durably as one file; readers never see half a write. +// Slots and queue share the same lifetime and revocation boundary. +type clientState struct { + Slots map[string]fabricSlot `json:"slots"` + Queue []queueEntry `json:"queue"` + LastNumber int64 `json:"last_number"` +} + +func (c fabricCall) statePath() string { + return filepath.Join(c.f.ns.root, "fabric", c.id, c.r.PathValue("device")+".json") +} + +func (c fabricCall) load() (clientState, error) { + s := clientState{Slots: map[string]fabricSlot{}} + raw, err := os.ReadFile(c.statePath()) + if errors.Is(err, os.ErrNotExist) { + return s, nil + } + if err != nil { + return s, err + } + if err := json.Unmarshal(raw, &s); err != nil { + return s, err + } + now := c.f.ns.now().UnixMilli() + for name, slot := range s.Slots { + if now-slot.At >= FabricMaxAge.Milliseconds() { + delete(s.Slots, name) + } + } + s.Queue = slices.DeleteFunc(s.Queue, func(e queueEntry) bool { return e.At <= now-FabricMaxAge.Milliseconds() }) + return s, nil +} + +func (c fabricCall) save(s clientState) error { + raw, err := json.Marshal(s) + if err != nil { + return err + } + path := c.statePath() + if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { + return err + } + file, err := os.CreateTemp(filepath.Dir(path), ".client-*") + if err != nil { + return err + } + defer os.Remove(file.Name()) + _, werr := file.Write(raw) + err = errors.Join(werr, file.Sync(), file.Close()) + if err != nil { + return err + } + if err := fsreplace.Rename(file.Name(), path); err != nil { + return err + } + dir, err := os.Open(filepath.Dir(path)) + if err != nil { + return err + } + return errors.Join(dir.Sync(), dir.Close()) +} + +func (c fabricCall) target(client bool) error { + callerClient := c.listing.Devices[c.dev].Kind == "client" + if callerClient != client || (client && c.dev != c.r.PathValue("device")) { + return wireauth.ErrUnauthorized + } + target, ok := c.listing.Devices[c.r.PathValue("device")] + if !ok || target.Kind != "client" || target.Revoked { + return badFabric(404, "not_found") + } + if c.r.Method != "GET" && c.listing.Identity.Rotation != nil { + return wireauth.ErrRotated + } + return nil +} + +func putSlot(c fabricCall) (any, error) { + if err := c.target(false); err != nil { + return nil, err + } + name := c.r.PathValue("name") + if !fabricName.MatchString(name) { + return nil, badFabric(400, "bad_request") + } + var in struct { + Sealed []byte `json:"sealed"` + } + if err := decodeFabric(c.body, &in); err != nil { + return nil, err + } + if len(in.Sealed) > MaxSlotBytes { + return nil, badFabric(413, "too_large") + } + if len(in.Sealed) == 0 { + return nil, badFabric(400, "bad_request") + } + s, err := c.load() + if err != nil { + return nil, err + } + if _, exists := s.Slots[name]; !exists && len(s.Slots) >= MaxClientSlots { + return nil, badFabric(429, "full") + } + s.Slots[name] = fabricSlot{Sealed: in.Sealed, Writer: c.dev, At: c.f.ns.now().UnixMilli()} + return nil, c.save(s) +} + +func getSlot(c fabricCall) (any, error) { + if err := c.target(true); err != nil { + return nil, err + } + if len(c.body) != 0 || !fabricName.MatchString(c.r.PathValue("name")) { + return nil, badFabric(400, "bad_request") + } + s, err := c.load() + if err != nil { + return nil, err + } + slot, ok := s.Slots[c.r.PathValue("name")] + if !ok { + return nil, badFabric(404, "not_found") + } + return slot, nil +} + +func revokeClient(c fabricCall) (any, error) { + if c.listing.Devices[c.dev].Kind == "client" { + return nil, wireauth.ErrUnauthorized + } + if len(c.body) != 0 { + return nil, badFabric(400, "bad_request") + } + if err := c.ns.commitFabricV1(c.r.Context(), c.dev, c.r, c.body); err != nil { + return nil, err + } + err := os.Remove(c.statePath()) + if errors.Is(err, os.ErrNotExist) { + err = nil + } + return nil, err +} + +// cleanRevoked completes deletion if the process stopped between committing a +// revoke and removing its client file. The sweep already holds the fabric lock. +func (f *fabric) cleanRevoked() error { + ids, err := f.ns.known() + if err != nil { + return err + } + for _, id := range ids { + if f.ns.gone(id) { + continue + } + entries, err := os.ReadDir(filepath.Join(f.ns.root, "fabric", id)) + if errors.Is(err, os.ErrNotExist) || len(entries) == 0 { + continue + } + if err != nil { + return err + } + d, err := f.ns.activate(context.Background(), id, nil) + if err != nil { + return err + } + l, err := d.fabricListing(context.Background()) + if err != nil { + return err + } + for _, entry := range entries { + dev, ok := strings.CutSuffix(entry.Name(), ".json") + if ok && fabricDevice.MatchString(dev) && l.Devices[dev].Revoked { + if err := os.Remove(filepath.Join(f.ns.root, "fabric", id, entry.Name())); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + } + } + } + return nil +} diff --git a/internal/relayserve/fabrictest/conformance.go b/internal/relayserve/fabrictest/conformance.go new file mode 100644 index 0000000000..b1172e6636 --- /dev/null +++ b/internal/relayserve/fabrictest/conformance.go @@ -0,0 +1,390 @@ +// Package fabrictest holds HTTP conformance cases shared by in-process and live relays. +package fabrictest + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "strconv" + "testing" + "time" + + "crypto/ed25519" + "github.com/Agent-Field/codeaf/internal/identity" + "github.com/Agent-Field/codeaf/internal/reqsign" +) + +func fabricCall(t *testing.T, r *Rig, d device, method, path string, body any, want int) []byte { + t.Helper() + var raw []byte + if body != nil { + var err error + raw, err = json.Marshal(body) + if err != nil { + t.Fatal(err) + } + } + req, err := http.NewRequest(method, r.URL()+path, bytes.NewReader(raw)) + if err != nil { + t.Fatal(err) + } + reqsign.Sign(req, raw, d, r.Now()) + resp, err := r.Client.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + out, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatal(err) + } + if resp.StatusCode != want { + t.Fatalf("%s %s: %d %s, want %d", method, path, resp.StatusCode, out, want) + } + return out +} + +func pairClient(t *testing.T, r *Rig) (device, device, string) { + t.Helper() + id, full := newIdentity(t) + client := newDevice(t, id) + path := "/v1/clients/" + client.dev.ID() + fabricCall(t, r, full, "PUT", path, map[string]any{"cert": client.Cert(), "methods": []string{"chat.read", "chat.send"}}, 204) + return full, client, path +} + +func testFabricClientScope(t *testing.T, r *Rig) { + full, client, path := pairClient(t, r) + raw := fabricCall(t, r, full, "GET", "/v1/dir/list", nil, 200) + var listing struct { + Devices map[string]struct { + Kind string + AddedBy string `json:"added_by"` + Methods []string + } + } + if err := json.Unmarshal(raw, &listing); err != nil { + t.Fatal(err) + } + rec := listing.Devices[client.dev.ID()] + if rec.Kind != "client" || len(rec.Methods) != 2 || rec.AddedBy != reqsign.IDOf(full.IdentityKey()) { + t.Fatalf("client record: %+v", rec) + } + for _, route := range []string{"/v1/dir/list", "/v1/store/stats", "/v1/identity/rotation"} { + fabricCall(t, r, client, "GET", route, nil, 401) + } + fabricCall(t, r, client, "PUT", "/v1/dir/devices/"+client.dev.ID(), map[string]any{"V": 1}, 401) + fabricCall(t, r, client, "POST", "/v1/dir/cells/chat", map[string]any{"head": "h"}, 401) + fabricCall(t, r, client, "POST", "/v1/dir/cells/chat/acquire", map[string]any{}, 401) + fabricCall(t, r, client, "POST", "/v1/dir/cells/chat/publish", map[string]any{"fence": 1, "old_head": "h", "head": "h2"}, 401) + fabricCall(t, r, client, "POST", "/v1/dir/cells/chat/release", map[string]any{"fence": 1}, 401) + fabricCall(t, r, client, "POST", "/v1/dir/vault", map[string]any{"old": "", "new": "v"}, 401) + fabricCall(t, r, client, "POST", "/v1/dir/devices/"+full.dev.ID()+"/revoke", nil, 401) + fabricCall(t, r, client, "POST", "/v1/store/has", map[string]any{"rids": []string{}}, 401) + fabricCall(t, r, full, "PUT", "/v1/dir/devices/"+full.dev.ID(), map[string]any{"V": 1, "kind": "client"}, 400) + fabricCall(t, r, full, "PUT", "/v1/dir/devices/"+full.dev.ID(), map[string]any{"V": 1, "methods": []string{"chat.read"}}, 400) + fabricCall(t, r, full, "PUT", path, map[string]any{"cert": client.Cert(), "methods": []string{}}, 409) + if r.Restart != nil { + r.Restart() + } + fabricCall(t, r, client, "GET", "/v1/dir/list", nil, 401) + fabricCall(t, r, full, "POST", "/v1/dir/devices/"+client.dev.ID()+"/revoke", nil, 204) + fabricCall(t, r, client, "PUT", path, nil, 401) +} + +func testFabricSlots(t *testing.T, r *Rig) { + full, client, path := pairClient(t, r) + slot := path + "/slots/home" + before := relayTime(t, r) + fabricCall(t, r, full, "PUT", slot, map[string]any{"sealed": []byte("first")}, 204) + raw := fabricCall(t, r, client, "GET", slot, nil, 200) + var got struct { + Sealed []byte + Writer string + At int64 + } + if err := json.Unmarshal(raw, &got); err != nil { + t.Fatal(err) + } + if string(got.Sealed) != "first" || got.Writer != full.dev.ID() || (got.At < before || got.At > relayTime(t, r)) { + t.Fatalf("slot: %+v", got) + } + fabricCall(t, r, full, "PUT", slot, map[string]any{"sealed": []byte("replacement")}, 204) + if r.Restart != nil { + r.Restart() + } + raw = fabricCall(t, r, client, "GET", slot, nil, 200) + if err := json.Unmarshal(raw, &got); err != nil || string(got.Sealed) != "replacement" { + t.Fatalf("after restart: %s %v", raw, err) + } + fabricCall(t, r, client, "PUT", slot, map[string]any{"sealed": []byte("forged")}, 401) + fabricCall(t, r, full, "GET", slot, nil, 401) + _, other := newIdentity(t) + fabricCall(t, r, other, "PUT", slot, map[string]any{"sealed": []byte("foreign")}, 404) + fabricCall(t, r, full, "PUT", slot, map[string]any{"sealed": make([]byte, 65537)}, 413) + if r.Advance == nil { + return + } + r.Advance(24 * time.Hour) + fabricCall(t, r, client, "GET", slot, nil, 404) + fabricCall(t, r, full, "PUT", slot, map[string]any{"sealed": []byte("fresh")}, 204) + fabricCall(t, r, full, "POST", "/v1/dir/devices/"+client.dev.ID()+"/revoke", nil, 204) + fabricCall(t, r, client, "GET", slot, nil, 401) +} + +func testFabricSlotCapacityAndClientIsolation(t *testing.T, r *Rig) { + full, client, path := pairClient(t, r) + other := newDevice(t, full.id) + otherPath := "/v1/clients/" + other.dev.ID() + fabricCall(t, r, full, "PUT", otherPath, map[string]any{"cert": other.Cert(), "methods": []string{}}, 204) + for n := 0; n < 64; n++ { + fabricCall(t, r, full, "PUT", path+"/slots/"+strconv.Itoa(n), map[string]any{"sealed": []byte("sealed")}, 204) + } + fabricCall(t, r, full, "PUT", path+"/slots/extra", map[string]any{"sealed": []byte("sealed")}, 429) + fabricCall(t, r, full, "PUT", path+"/slots/0", map[string]any{"sealed": []byte("replaced")}, 204) + fabricCall(t, r, other, "GET", path+"/slots/0", nil, 401) + fabricCall(t, r, client, "GET", otherPath+"/slots/0", nil, 401) + if r.Advance == nil { + return + } + r.Advance(24 * time.Hour) + fabricCall(t, r, full, "PUT", path+"/slots/extra", map[string]any{"sealed": []byte("sealed")}, 204) +} + +func testFabricQueueClaimsAndProof(t *testing.T, r *Rig) { + full, client, path := pairClient(t, r) + second := newDevice(t, full.id) + queue := path + "/queue" + body := map[string]any{"number": 1, "at": r.Now().UnixMilli(), "method": "chat.send", "target": "any", "sealed": []byte("sealed call")} + fabricCall(t, r, client, "POST", queue, body, 204) + raw := fabricCall(t, r, full, "GET", queue, nil, 200) + var got struct { + Entries []struct { + Number int64 + Sealed []byte + Body []byte + Signature string + Cert string + Identity string + RequestTime string `json:"request_time"` + RequestURI string `json:"request_uri"` + } + } + if err := json.Unmarshal(raw, &got); err != nil { + t.Fatal(err) + } + if len(got.Entries) != 1 || got.Entries[0].Number != 1 || string(got.Entries[0].Sealed) != "sealed call" { + t.Fatalf("queue: %s", raw) + } + e := got.Entries[0] + req, _ := http.NewRequest("POST", r.URL()+e.RequestURI, bytes.NewReader(e.Body)) + req.Header.Set(reqsign.HeaderIdentity, e.Identity) + req.Header.Set(reqsign.HeaderCert, e.Cert) + req.Header.Set(reqsign.HeaderTime, e.RequestTime) + req.Header.Set(reqsign.HeaderSig, e.Signature) + if _, dev, err := reqsign.Verify(req, e.Body, r.Now()); err != nil || dev != client.dev.ID() { + t.Fatalf("entry proof: %s %v", dev, err) + } + fabricCall(t, r, full, "DELETE", queue+"/1", nil, 409) + fabricCall(t, r, full, "POST", queue+"/1/claim", nil, 204) + if r.Restart != nil { + r.Restart() + } + fabricCall(t, r, full, "POST", queue+"/1/claim", nil, 204) + fabricCall(t, r, second, "POST", queue+"/1/claim", nil, 409) + fabricCall(t, r, second, "DELETE", queue+"/1", nil, 409) + fabricCall(t, r, full, "DELETE", queue+"/1", nil, 204) + raw = fabricCall(t, r, full, "GET", queue, nil, 200) + if string(raw) != "{\"entries\":[]}\n" { + t.Fatalf("acked queue: %s", raw) + } + fabricCall(t, r, client, "POST", queue, body, 409) + body["number"] = 3 + fabricCall(t, r, client, "POST", queue, body, 409) + body["number"] = 2 + fabricCall(t, r, client, "POST", queue, body, 204) +} + +func testFabricQueueLimitsRoutingAndExpiry(t *testing.T, r *Rig) { + full, client, path := pairClient(t, r) + queue := path + "/queue" + in := map[string]any{"number": 1, "at": r.Now().UnixMilli(), "method": "chat.send", "target": "any", "sealed": []byte("sealed")} + in["method"] = "shell.run" + fabricCall(t, r, client, "POST", queue, in, 403) + in["method"] = "chat.send" + in["at"] = r.Now().Add(-24 * time.Hour).UnixMilli() + fabricCall(t, r, client, "POST", queue, in, 400) + in["at"] = r.Now().Add(6 * time.Minute).UnixMilli() + fabricCall(t, r, client, "POST", queue, in, 400) + in["at"] = r.Now().UnixMilli() + in["sealed"] = make([]byte, 4097) + fabricCall(t, r, client, "POST", queue, in, 413) + in["sealed"] = []byte("sealed") + in["target"] = full.dev.ID() + fabricCall(t, r, client, "POST", queue, in, 400) + fabricCall(t, r, full, "PUT", "/v1/dir/devices/"+full.dev.ID(), map[string]any{"V": 1}, 204) + fabricCall(t, r, client, "POST", queue, in, 204) + other := newDevice(t, full.id) + raw := fabricCall(t, r, other, "GET", queue, nil, 200) + if string(raw) != "{\"entries\":[]}\n" { + t.Fatalf("target isolation: %s", raw) + } + fabricCall(t, r, other, "POST", queue+"/1/claim", nil, 401) + fabricCall(t, r, full, "POST", queue+"/1/claim", nil, 204) + fabricCall(t, r, full, "DELETE", queue+"/1", nil, 204) + in["target"] = "any" + for number := 2; number <= 129; number++ { + in["number"] = number + fabricCall(t, r, client, "POST", queue, in, 204) + } + in["number"] = 130 + fabricCall(t, r, client, "POST", queue, in, 429) + if r.Advance == nil { + return + } + r.Advance(24 * time.Hour) + fabricCall(t, r, full, "POST", queue+"/2/claim", nil, 404) + in["at"] = r.Now().UnixMilli() + fabricCall(t, r, client, "POST", queue, in, 204) + if r.Restart != nil { + r.Restart() + } + in["number"] = 129 + fabricCall(t, r, client, "POST", queue, in, 409) +} + +func testFabricQueueOneClaimer(t *testing.T, r *Rig) { + full, client, path := pairClient(t, r) + other := newDevice(t, full.id) + fabricCall(t, r, client, "POST", path+"/queue", map[string]any{"number": 1, "at": r.Now().UnixMilli(), "method": "chat.send", "target": "any", "sealed": []byte("sealed")}, 204) + statuses := make(chan int, 2) + for _, d := range []device{full, other} { + go func(d device) { + req, _ := http.NewRequest("POST", r.URL()+path+"/queue/1/claim", nil) + reqsign.Sign(req, nil, d, r.Now()) + resp, err := r.Client.Do(req) + if err != nil { + statuses <- 0 + return + } + resp.Body.Close() + statuses <- resp.StatusCode + }(d) + } + a, b := <-statuses, <-statuses + if !((a == 204 && b == 409) || (a == 409 && b == 204)) { + t.Fatalf("claim race: %d %d", a, b) + } +} + +// Rig drives the real signed HTTP seam. Live relays omit clock advancement and restart. +type Rig struct { + URL func() string + Client *http.Client + Now func() time.Time + Advance func(time.Duration) + Restart func() +} + +type device struct { + id identity.Identity + dev identity.Dev +} + +func (d device) IdentityKey() ed25519.PublicKey { return d.id.PublicKey() } +func (d device) Cert() identity.Cert { return d.dev.Cert } +func (d device) Sign(msg []byte) []byte { return d.dev.Sign(msg) } + +func newIdentity(t *testing.T) (identity.Identity, device) { + t.Helper() + id, err := identity.Ensure(t.TempDir()) + if err != nil { + t.Fatal(err) + } + return id, newDevice(t, id) +} +func newDevice(t *testing.T, id identity.Identity) device { + t.Helper() + seed, err := identity.NewSeed() + if err != nil { + t.Fatal(err) + } + dev, err := seed.Join(identity.IssueCert(id, seed.Public(), time.Now())) + if err != nil { + t.Fatal(err) + } + return device{id, dev} +} + +// Run holds every implementation to the same scoped-client, slot and queue contract. +func Run(t *testing.T, makeRig func(*testing.T) *Rig) { + t.Helper() + t.Run("AuthorizationAndBodies", func(t *testing.T) { testFabricAuthorizationAndBodies(t, makeRig(t)) }) + t.Run("Revocation", func(t *testing.T) { testFabricRevocation(t, makeRig(t)) }) + t.Run("ClientScope", func(t *testing.T) { testFabricClientScope(t, makeRig(t)) }) + t.Run("Slots", func(t *testing.T) { testFabricSlots(t, makeRig(t)) }) + t.Run("SlotCapacityAndClientIsolation", func(t *testing.T) { testFabricSlotCapacityAndClientIsolation(t, makeRig(t)) }) + t.Run("QueueClaimsAndProof", func(t *testing.T) { testFabricQueueClaimsAndProof(t, makeRig(t)) }) + t.Run("QueueLimitsRoutingAndExpiry", func(t *testing.T) { testFabricQueueLimitsRoutingAndExpiry(t, makeRig(t)) }) + t.Run("QueueOneClaimer", func(t *testing.T) { testFabricQueueOneClaimer(t, makeRig(t)) }) +} + +func testFabricAuthorizationAndBodies(t *testing.T, r *Rig) { + full, client, path := pairClient(t, r) + other := newDevice(t, full.id) + otherPath := "/v1/clients/" + other.dev.ID() + badCert := identity.IssueCert(full.id, ed25519.PublicKey{1}, r.Now()) + fabricCall(t, r, full, "PUT", "/v1/clients/"+badCert.DeviceID(), map[string]any{"cert": badCert, "methods": []string{}}, 400) + fabricCall(t, r, client, "PUT", otherPath, map[string]any{"cert": other.Cert(), "methods": []string{}}, 401) + fabricCall(t, r, full, "PUT", otherPath, map[string]any{"cert": client.Cert(), "methods": []string{}}, 400) + _, foreign := newIdentity(t) + fabricCall(t, r, full, "PUT", "/v1/clients/"+foreign.dev.ID(), map[string]any{"cert": foreign.Cert(), "methods": []string{}}, 400) + fabricCall(t, r, full, "PUT", otherPath, map[string]any{"cert": other.Cert(), "methods": []string{}, "grant": "forbidden"}, 400) + fabricCall(t, r, full, "PUT", otherPath, map[string]any{"cert": other.Cert(), "methods": []string{"chat.send", "chat.send"}}, 400) + fabricCall(t, r, full, "PUT", otherPath, map[string]any{"cert": other.Cert(), "methods": []string{}}, 204) + fabricCall(t, r, other, "POST", otherPath+"/queue", map[string]any{"number": 1, "at": r.Now().UnixMilli(), "method": "chat.send", "target": "any", "sealed": []byte("sealed")}, 403) + fabricCall(t, r, client, "GET", path+"/queue", nil, 401) + fabricCall(t, r, full, "POST", path+"/queue", map[string]any{"number": 1}, 401) + fabricCall(t, r, full, "PUT", path+"/slots/max", map[string]any{"sealed": make([]byte, 65536)}, 204) + fabricCall(t, r, client, "GET", path+"/slots/max", nil, 200) + fabricCall(t, r, full, "PUT", path+"/slots/empty", map[string]any{"sealed": []byte{}}, 400) + fabricCall(t, r, client, "GET", path+"/slots/max", map[string]any{}, 400) + fabricCall(t, r, full, "PUT", path+"/slots/huge", map[string]any{"sealed": make([]byte, 100000)}, 413) +} + +func testFabricRevocation(t *testing.T, r *Rig) { + full, client, path := pairClient(t, r) + fabricCall(t, r, full, "PUT", path+"/slots/home", map[string]any{"sealed": []byte("snapshot")}, 204) + fabricCall(t, r, client, "POST", path+"/queue", map[string]any{"number": 1, "at": r.Now().UnixMilli(), "method": "chat.send", "target": "any", "sealed": []byte("call")}, 204) + fabricCall(t, r, full, "POST", path+"/queue/1/claim", nil, 204) + fabricCall(t, r, full, "POST", "/v1/dir/devices/"+client.dev.ID()+"/revoke", nil, 204) + check := func() { + fabricCall(t, r, client, "GET", path+"/slots/home", nil, 401) + fabricCall(t, r, client, "POST", path+"/queue", map[string]any{"number": 2}, 401) + fabricCall(t, r, full, "PUT", path+"/slots/home", map[string]any{"sealed": []byte("snapshot")}, 404) + fabricCall(t, r, full, "GET", path+"/queue", nil, 404) + fabricCall(t, r, full, "POST", path+"/queue/1/claim", nil, 404) + fabricCall(t, r, full, "DELETE", path+"/queue/1", nil, 404) + } + check() + if r.Restart != nil { + r.Restart() + check() + } +} + +// relayTime brackets writes using the relay clock, including in live runs. +func relayTime(t *testing.T, r *Rig) int64 { + t.Helper() + resp, err := r.Client.Get(r.URL() + "/status") + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + at, err := strconv.ParseInt(resp.Header.Get("Codeaf-Now"), 10, 64) + if err != nil { + t.Fatal(err) + } + return at +} diff --git a/internal/relayserve/native.go b/internal/relayserve/native.go index 47e7e2202d..75786d196c 100644 --- a/internal/relayserve/native.go +++ b/internal/relayserve/native.go @@ -46,6 +46,7 @@ type nativeNamespace struct { now func() time.Time grace directory.GraceBounds identity clusterstate.IdentityRec + root string } func (d *nativeNamespace) Feed() *directory.Feed { return d.feed } @@ -187,7 +188,7 @@ func (n *namespaces) activate(ctx context.Context, id string, root []byte) (*nat if cfg.Grace != nil { grace = directory.GraceBounds{Min: time.Duration(cfg.Grace.Min) * time.Millisecond, Max: time.Duration(cfg.Grace.Max) * time.Millisecond, Default: time.Duration(cfg.Grace.Default) * time.Millisecond} } - d = &nativeNamespace{V1Directory: &raftdirectory.V1Directory{Host: host, Cluster: cfg.Cluster, Now: n.now, Grace: grace}, identity: clusterstate.IdentityRec{V: 1, Identity: id, SigningKey: cfg.SigningKey}, host: host, feed: feed, links: n.links.(*directory.Links), now: n.now, grace: grace} + d = &nativeNamespace{V1Directory: &raftdirectory.V1Directory{Host: host, Cluster: cfg.Cluster, Now: n.now, Grace: grace}, identity: clusterstate.IdentityRec{V: 1, Identity: id, SigningKey: cfg.SigningKey}, host: host, feed: feed, links: n.links.(*directory.Links), now: n.now, grace: grace, root: n.root} d.effects = raftdirectory.Effects{Host: host, Cluster: cfg.Cluster, Actor: clusterstate.DeviceIDOfKey(cfg.EffectsKey), Key: ed25519.PrivateKey(provision.Effects)} if err = host.InitializeMembership(ctx, ed25519.PrivateKey(provision.Anchors[0])); err != nil { d.Close() @@ -282,6 +283,7 @@ func (d *nativeNamespace) project(s *raftstore.SQLite, a raftstore.Application) return err } st.Stopped[string(id)] = true + _ = d.purgeFabric(string(id)) } } } @@ -289,7 +291,22 @@ func (d *nativeNamespace) project(s *raftstore.SQLite, a raftstore.Application) return nil } +func (d *nativeNamespace) purgeFabric(dev string) error { + if d.root == "" || !fabricDevice.MatchString(dev) { + return nil + } + err := os.Remove(filepath.Join(d.root, "fabric", d.identity.Identity, dev+".json")) + if errors.Is(err, os.ErrNotExist) { + return nil + } + return err +} + func (d *nativeNamespace) authority(ctx context.Context, actor string, write bool) error { + return d.authorityClient(ctx, actor, write, false) +} + +func (d *nativeNamespace) authorityClient(ctx context.Context, actor string, write, allowClient bool) error { err := d.host.Inspect(ctx, func(ctx context.Context, s *raftstore.SQLite, a raftstore.Application) error { keys := []raftstore.Key{{Namespace: raftstore.State, Key: clusterstate.IdentityKey().Key}, {Namespace: raftstore.State, Key: clusterstate.DeviceKey(actor).Key}} rows, _, err := s.Read(ctx, a.Applied, keys) @@ -308,6 +325,9 @@ func (d *nativeNamespace) authority(ctx context.Context, actor string, write boo if dev.Revoked { return wireauth.ErrRevoked } + if !allowClient && dev.Kind == "client" { + return wireauth.ErrUnauthorized + } } if write && identity.Rotation != nil { return wireauth.ErrRotated @@ -320,6 +340,63 @@ func (d *nativeNamespace) authority(ctx context.Context, actor string, write boo return err } +func (d *nativeNamespace) fabricListing(ctx context.Context) (directory.Listing, error) { + var out directory.Listing + out.Devices = map[string]directory.Device{} + err := d.host.Inspect(ctx, func(ctx context.Context, s *raftstore.SQLite, a raftstore.Application) error { + key := raftstore.Key{Namespace: raftstore.State, Key: clusterstate.IdentityKey().Key} + rows, _, err := s.Read(ctx, a.Applied, []raftstore.Key{key}) + if err != nil { + return err + } + if err = clusterstate.Decode(rows[key], &out.Identity); err != nil { + return err + } + after := "device/" + for { + page, _, err := s.ScanPage(ctx, a.Applied, raftstore.State, after) + if err != nil || len(page) == 0 { + return err + } + for _, r := range page { + after = r.Key.Key + if !strings.HasPrefix(after, "device/") { + return nil + } + var dev clusterstate.Device + if err = clusterstate.Decode(r.Value, &dev); err != nil { + return err + } + id, err := hex.DecodeString(strings.TrimPrefix(after, "device/")) + if err != nil { + return err + } + out.Devices[string(id)] = dev + } + } + }) + return out, err +} + +func (d *nativeNamespace) commitFabricV1(ctx context.Context, actor string, r *http.Request, body []byte) error { + p, err := reqsign.Capture(r, body) + if err != nil { + return directory.ErrUnauthorized + } + cmd, err := clusterstate.V1Command(d.Cluster, d.now().UnixMilli(), p) + if err != nil { + return directory.ErrBadRequest + } + if cmd.Actor != actor { + return directory.ErrUnauthorized + } + res, err := d.host.CommitV1(ctx, cmd) + if err != nil { + return err + } + return directory.ResultError(res.Code) +} + // SubscribeWatch retains the certificate from this exact authenticated socket, // rather than enrolling a row that would consume grandfathered capacity. func (d *nativeNamespace) SubscribeWatch(device string, r *http.Request, holds []directory.Hold, beat time.Duration) (*directory.Sub, error) { @@ -386,6 +463,14 @@ func (d *nativeNamespace) seen(device string, at int64) error { // Every signature is checked before allocating storage or looking up authority. // The actor then comes from that proof for both the initial and post-I/O check. func (n *namespaces) authenticate(r *http.Request, body []byte) (string, string, error) { + return n.authenticateClient(r, body, false) +} + +func (n *namespaces) authenticateFabric(r *http.Request, body []byte) (string, string, error) { + return n.authenticateClient(r, body, true) +} + +func (n *namespaces) authenticateClient(r *http.Request, body []byte, allowClient bool) (string, string, error) { id, actor, err := reqsign.Verify(r, body, n.now()) if err != nil { return "", "", err @@ -407,7 +492,7 @@ func (n *namespaces) authenticate(r *http.Request, body []byte) (string, string, n.unavailable("directory activation", err) return "", "", errors.Join(wireauth.ErrUnavailable, err) } - if err = d.authority(r.Context(), actor, false); err != nil { + if err = d.authorityClient(r.Context(), actor, false, allowClient); err != nil { // A caller that left before authentication finished has no authenticated // request to retry. A live caller still receives a real outage as 503. if r.Context().Err() != nil { diff --git a/internal/relayserve/retire.go b/internal/relayserve/retire.go index b6f4f9c6fd..3d4e3128b2 100644 --- a/internal/relayserve/retire.go +++ b/internal/relayserve/retire.go @@ -101,7 +101,7 @@ func (n *namespaces) erase(id string) error { failed = errors.Join(failed, err) } } - return errors.Join(failed, os.RemoveAll(filepath.Join(n.root, "blobs", id)), os.RemoveAll(filepath.Join(n.root, "raft", id))) + return errors.Join(failed, os.RemoveAll(filepath.Join(n.root, "blobs", id)), os.RemoveAll(filepath.Join(n.root, "fabric", id)), os.RemoveAll(filepath.Join(n.root, "raft", id))) } // forget closes the identity's open directory, if it has one. diff --git a/internal/relayserve/service.go b/internal/relayserve/service.go index 66d633ada7..046e4b8358 100644 --- a/internal/relayserve/service.go +++ b/internal/relayserve/service.go @@ -59,6 +59,7 @@ type Config struct { type Service struct { Handler http.Handler ns *namespaces + fabric *fabric stop chan struct{} swept sync.WaitGroup } @@ -80,7 +81,12 @@ func (s *Service) Sweep() error { if s.ns == nil { return nil } - return s.ns.sweep() + s.fabric.mu.Lock() + defer s.fabric.mu.Unlock() + if err := s.ns.sweep(); err != nil { + return err + } + return s.fabric.cleanRevoked() } // sweepLoop sweeps once at start, so a relay that was down past a deadline @@ -130,7 +136,11 @@ func New(cfg Config) *Service { } mux.Handle(clusterctl.Path, clusterctl.Handler(svc.ns.controlAuthenticate, svc.ns.controlBackend)) auth := noting(svc.ns.authenticate) - dirs := logged(cfg.Logf, directory.Handler(auth, svc.ns.directory)) + fabric := &fabric{ns: svc.ns, auth: noting(svc.ns.authenticateFabric)} + svc.fabric = fabric + dirs := logged(cfg.Logf, fabric.directoryHandler(directory.Handler(auth, svc.ns.directory))) + mux.Handle("/v1/clients/", logged(cfg.Logf, fabric.handler())) + mux.Handle("POST /v1/dir/devices/{device}/revoke", logged(cfg.Logf, fabric.serve(revokeClient))) mux.Handle("/v1/dir/", dirs) mux.Handle("/v1/identity/", dirs) mux.Handle(directory.LinkPath+"/", logged(cfg.Logf, directory.LinkHandler(links, peerOf(cfg)))) diff --git a/relay/hosted/src/directory.js b/relay/hosted/src/directory.js index 0d3aff4c3d..1e9eda37b7 100644 --- a/relay/hosted/src/directory.js +++ b/relay/hosted/src/directory.js @@ -138,18 +138,31 @@ export class Directory { putDevice(id, device) { // A device's outbox and took lists carry messages between its computers (Codeaf-Chat-Notes); both stay short. if ((device.outbox?.length ?? 0) > MAX_NOTES || (device.took?.length ?? 0) > MAX_TOOK) throw refuse('bad_request'); + if ((device.kind ?? '') !== '' || (device.methods?.length ?? 0) > 0) throw refuse('bad_request'); this.write('devices', id, this.#kept(id, { ...device, revoked: this.revoked(id) })); } + /** putClientDevice writes the record of a scoped client device. */ + putClientDevice(id, device) { + this.write('devices', id, { ...this.#kept(id, { ...device, revoked: false }), kind: 'client', methods: device.methods }); + } + /** approveDevice writes the record of a device a paired device approved, and answers what was stored. */ approveDevice(id, device) { + if ((device.kind ?? '') !== '' || (device.methods?.length ?? 0) > 0) throw refuse('bad_request'); return this.write('devices', id, this.#kept(id, { ...device, revoked: false })); } #kept(id, doc) { const before = this.read('devices', id); - const { created: _c, last_seen: _l, ...own } = doc; - return { ...own, created: before ? before.created : this.clock(), last_seen: before?.last_seen }; + const { created: _c, last_seen: _l, kind: _k, methods: _m, ...own } = doc; + return { + ...own, + ...(before?.kind && { kind: before.kind }), + ...(before?.methods?.length && { methods: before.methods }), + created: before ? before.created : this.clock(), + last_seen: before?.last_seen, + }; } /** seen stamps when the device last held a watch socket (hello or close). Only the relay writes it; it moves no version (contract 4). */ diff --git a/relay/hosted/src/fabric.js b/relay/hosted/src/fabric.js new file mode 100644 index 0000000000..c7b6797102 --- /dev/null +++ b/relay/hosted/src/fabric.js @@ -0,0 +1,323 @@ +import { base64, fromBase64Url, hex, sha256 } from './codec.js'; +import { SKEW_MS, Refusal } from './verify.js'; +import { Wire, badRequest, notFound } from './wire.js'; + +const DEVICE = /^dev_[0-9a-f]{32}$/; +const NAME = /^[A-Za-z0-9_.-]{1,128}$/; +const CERT_LABEL = 'codeaf/device-cert/v1\n'; +const decoder = new TextDecoder(); +const encoder = new TextEncoder(); + +export const MAX_FABRIC_BODY = 96 << 10; +const MAX_CLIENT_METHODS = 64; +const MAX_CLIENT_NAME = 256; +const MAX_SLOT_BYTES = 64 << 10; +const MAX_CLIENT_SLOTS = 64; +const MAX_QUEUE_BYTES = 4096; +const MAX_CLIENT_QUEUE = 128; +const FABRIC_MAX_AGE_MS = 24 * 60 * 60 * 1000; + +const claimed = () => new Wire('claimed', 409); +const stale = () => new Wire('stale', 400); +const full = () => new Wire('full', 429); +const exists = () => new Wire('exists', 409); +const sequence = () => new Wire('sequence', 409); +const methodDenied = () => new Wire('method_denied', 403); +const tooLarge = () => new Wire('too_large', 413); +const unauthorized = () => new Refusal('unauthorized', 'unauthorized'); + +const fromHex = (text) => { + if (typeof text !== 'string' || !/^([0-9a-f]{2})*$/.test(text)) return null; + return Uint8Array.from(text.match(/../g) ?? [], (h) => parseInt(h, 16)); +}; + +const fromBase64 = (text) => { + if (typeof text !== 'string' || text.length % 4 !== 0 || !/^[A-Za-z0-9+/]*={0,2}$/.test(text)) return null; + try { + return Uint8Array.from(atob(text), (c) => c.charCodeAt(0)); + } catch { + return null; + } +}; + +const certBody = (cert) => encoder.encode(CERT_LABEL + JSON.stringify({ V: cert.V, device: cert.device, created: cert.created, sig: '' })); + +function parseObject(body) { + let value; + try { + value = JSON.parse(decoder.decode(body)); + } catch { + throw badRequest(); + } + if (!value || typeof value !== 'object' || Array.isArray(value)) throw badRequest(); + return value; +} + +function expectKeys(body, required, optional = []) { + const obj = parseObject(body); + const allowed = new Set([...required, ...optional]); + for (const key of Object.keys(obj)) { + if (!allowed.has(key)) throw badRequest(); + } + for (const key of required) { + if (!(key in obj)) throw badRequest(); + } + return obj; +} + +const ensureDevice = (device) => { + if (!DEVICE.test(device)) throw badRequest(); +}; + +const ensureName = (name) => { + if (typeof name !== 'string' || !NAME.test(name)) throw badRequest(); +}; + +const asInt = (value) => { + if (typeof value !== 'number' || !Number.isInteger(value) || !Number.isSafeInteger(value)) throw badRequest(); + return value; +}; + +const parseEntryNumber = (text) => { + if (typeof text !== 'string' || !/^[1-9][0-9]*$/.test(text)) throw badRequest(); + const number = Number(text); + if (!Number.isSafeInteger(number)) throw badRequest(); + return number; +}; + +const verifyClientCert = async (cert, identityHeader, targetDevice) => { + const key = fromBase64Url(identityHeader ?? ''); + if (!key || key.length !== 32) throw badRequest(); + if (!cert || typeof cert !== 'object' || Array.isArray(cert)) throw badRequest(); + if (new Set(Object.keys(cert)).size !== Object.keys(cert).length) throw badRequest(); + for (const keyName of Object.keys(cert)) { + if (!new Set(['V', 'device', 'created', 'sig']).has(keyName)) throw badRequest(); + } + if (!('V' in cert) || !('device' in cert) || !('created' in cert) || !('sig' in cert)) throw badRequest(); + const devicePub = fromHex(cert.device); + const sig = fromHex(cert.sig); + if (!devicePub || !sig || devicePub.length !== 32) throw badRequest(); + const imported = await crypto.subtle.importKey('raw', key, { name: 'Ed25519' }, false, ['verify']); + const ok = await crypto.subtle.verify({ name: 'Ed25519' }, imported, sig, certBody(cert)); + if (!ok) throw badRequest(); + const id = 'dev_' + hex((await sha256(devicePub)).subarray(0, 16)); + if (id !== targetDevice) throw badRequest(); +}; + +export class Fabric { + constructor(sql, clock, dir) { + this.sql = sql; + this.clock = clock; + this.dir = dir; + sql.exec( + 'CREATE TABLE IF NOT EXISTS fabric_slots (device TEXT NOT NULL, name TEXT NOT NULL, sealed TEXT NOT NULL, writer TEXT NOT NULL, at INTEGER NOT NULL, PRIMARY KEY (device, name)) WITHOUT ROWID', + ); + sql.exec( + 'CREATE TABLE IF NOT EXISTS fabric_queue (device TEXT NOT NULL, number INTEGER NOT NULL, at INTEGER NOT NULL, method TEXT NOT NULL, target TEXT NOT NULL, sealed TEXT NOT NULL, claimed_by TEXT NOT NULL, signature TEXT NOT NULL, cert TEXT NOT NULL, identity TEXT NOT NULL, request_time TEXT NOT NULL, request_uri TEXT NOT NULL, body TEXT NOT NULL, PRIMARY KEY (device, number)) WITHOUT ROWID', + ); + sql.exec('CREATE TABLE IF NOT EXISTS fabric_seq (device TEXT PRIMARY KEY, last_number INTEGER NOT NULL) WITHOUT ROWID'); + } + + async withClockAndDir(now, dir, fn) { + const prevClock = this.clock; + const prevDir = this.dir; + this.clock = () => now; + this.dir = dir; + try { + return await fn(); + } finally { + this.clock = prevClock; + this.dir = prevDir; + } + } + + isClient(device) { + return this.dir.read('devices', device)?.kind === 'client'; + } + + revokeClient(device) { + this.sql.exec('DELETE FROM fabric_slots WHERE device = ?', device); + this.sql.exec('DELETE FROM fabric_queue WHERE device = ?', device); + this.sql.exec('DELETE FROM fabric_seq WHERE device = ?', device); + } + + async validateRegisterClient({ targetDevice, body, identityHeader }) { + ensureDevice(targetDevice); + const inBody = expectKeys(body, ['cert', 'methods'], ['name']); + if (!Array.isArray(inBody.methods)) throw badRequest(); + const name = inBody.name ?? ''; + if (typeof name !== 'string') throw badRequest(); + if (inBody.methods.length > MAX_CLIENT_METHODS || name.length > MAX_CLIENT_NAME) throw tooLarge(); + const seen = new Set(); + for (const method of inBody.methods) { + if (typeof method !== 'string' || !NAME.test(method) || seen.has(method)) throw badRequest(); + seen.add(method); + } + await verifyClientCert(inBody.cert, identityHeader, targetDevice); + return { inBody, name }; + } + + async registerClient({ caller, targetDevice, body, identityHeader, identity }) { + this.#requireFullCaller(caller); + const { inBody, name } = await this.validateRegisterClient({ targetDevice, body, identityHeader }); + if (this.dir.read('devices', targetDevice)) throw exists(); + this.dir.putClientDevice(targetDevice, { V: 1, kind: 'client', methods: inBody.methods, name, added_by: identity }); + } + + putSlot({ caller, targetDevice, name, body }) { + ensureDevice(targetDevice); + ensureName(name); + this.#requireFullCaller(caller); + this.#requireTargetClient(targetDevice); + const inBody = expectKeys(body, ['sealed']); + const sealed = fromBase64(inBody.sealed); + if (!sealed) throw badRequest(); + if (sealed.length > MAX_SLOT_BYTES) throw tooLarge(); + if (sealed.length === 0) throw badRequest(); + this.#prune(targetDevice); + const hasSlot = this.sql.exec('SELECT 1 FROM fabric_slots WHERE device = ? AND name = ?', targetDevice, name).toArray().length > 0; + if (!hasSlot && this.#slotCount(targetDevice) >= MAX_CLIENT_SLOTS) throw full(); + this.sql.exec( + 'INSERT OR REPLACE INTO fabric_slots VALUES (?,?,?,?,?)', + targetDevice, + name, + base64(sealed), + caller, + this.clock(), + ); + } + + getSlot({ caller, targetDevice, name, body }) { + ensureDevice(targetDevice); + ensureName(name); + this.#requireClientCaller(caller, targetDevice); + this.#requireTargetClient(targetDevice); + if (body.length !== 0) throw badRequest(); + this.#prune(targetDevice); + const row = this.sql.exec('SELECT sealed, writer, at FROM fabric_slots WHERE device = ? AND name = ?', targetDevice, name).toArray()[0]; + if (!row) throw notFound(); + return { sealed: row.sealed, writer: row.writer, at: row.at }; + } + + appendQueue({ caller, targetDevice, body, proof }) { + ensureDevice(targetDevice); + this.#requireClientCaller(caller, targetDevice); + this.#requireTargetClient(targetDevice); + const inBody = expectKeys(body, ['number', 'at', 'method', 'target', 'sealed']); + const number = asInt(inBody.number); + const at = asInt(inBody.at); + if (number <= 0) throw sequence(); + if (typeof inBody.method !== 'string' || !NAME.test(inBody.method)) throw badRequest(); + const callerRec = this.dir.read('devices', caller); + if (!Array.isArray(callerRec?.methods) || !callerRec.methods.includes(inBody.method)) throw methodDenied(); + const target = inBody.target; + if (typeof target !== 'string') throw badRequest(); + if (target !== 'any') { + if (!DEVICE.test(target)) throw badRequest(); + const rec = this.dir.read('devices', target); + if (!rec || rec.kind === 'client' || rec.revoked === true) throw badRequest(); + } + const sealed = fromBase64(inBody.sealed); + if (!sealed) throw badRequest(); + if (sealed.length > MAX_QUEUE_BYTES) throw tooLarge(); + if (sealed.length === 0) throw badRequest(); + const now = this.clock(); + if (at <= now - FABRIC_MAX_AGE_MS || at > now + SKEW_MS) throw stale(); + this.#prune(targetDevice); + const last = this.sql.exec('SELECT last_number FROM fabric_seq WHERE device = ?', targetDevice).toArray()[0]?.last_number ?? 0; + if (number !== last + 1) throw sequence(); + if (this.#queueCount(targetDevice) >= MAX_CLIENT_QUEUE) throw full(); + this.sql.exec( + 'INSERT INTO fabric_queue VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)', + targetDevice, + number, + at, + inBody.method, + target, + base64(sealed), + '', + proof.signature, + proof.cert, + proof.identity, + proof.requestTime, + proof.requestUri, + base64(proof.bodyBytes), + ); + this.sql.exec('INSERT INTO fabric_seq VALUES (?,?) ON CONFLICT(device) DO UPDATE SET last_number = excluded.last_number', targetDevice, number); + } + + listQueue({ caller, targetDevice, body }) { + ensureDevice(targetDevice); + this.#requireFullCaller(caller); + this.#requireTargetClient(targetDevice); + if (body.length !== 0) throw badRequest(); + this.#prune(targetDevice); + const rows = this.sql + .exec( + 'SELECT number, at, method, target, sealed, signature, cert, identity, request_time, request_uri, body FROM fabric_queue WHERE device = ? AND (target = ? OR target = ?) AND (claimed_by = ? OR claimed_by = ?) ORDER BY number', + targetDevice, + 'any', + caller, + '', + caller, + ) + .toArray(); + return rows.map((row) => ({ + number: row.number, + at: row.at, + method: row.method, + target: row.target, + sealed: row.sealed, + signature: row.signature, + cert: row.cert, + identity: row.identity, + request_time: row.request_time, + request_uri: row.request_uri, + body: row.body, + })); + } + + claimQueue({ caller, targetDevice, numberText, body, ack }) { + ensureDevice(targetDevice); + this.#requireFullCaller(caller); + this.#requireTargetClient(targetDevice); + if (body.length !== 0) throw badRequest(); + const number = parseEntryNumber(numberText); + this.#prune(targetDevice); + const row = this.sql + .exec('SELECT target, claimed_by FROM fabric_queue WHERE device = ? AND number = ?', targetDevice, number) + .toArray()[0]; + if (!row) throw notFound(); + if (row.target !== 'any' && row.target !== caller) throw unauthorized(); + if ((row.claimed_by !== '' && row.claimed_by !== caller) || (ack && row.claimed_by !== caller)) throw claimed(); + if (ack) this.sql.exec('DELETE FROM fabric_queue WHERE device = ? AND number = ?', targetDevice, number); + else this.sql.exec('UPDATE fabric_queue SET claimed_by = ? WHERE device = ? AND number = ?', caller, targetDevice, number); + } + + #requireFullCaller(caller) { + if (this.isClient(caller)) throw unauthorized(); + } + + #requireClientCaller(caller, targetDevice) { + if (!this.isClient(caller) || caller !== targetDevice) throw unauthorized(); + } + + #requireTargetClient(targetDevice) { + const target = this.dir.read('devices', targetDevice); + if (!target || target.kind !== 'client' || target.revoked === true) throw notFound(); + } + + #slotCount(device) { + return this.sql.exec('SELECT COUNT(*) AS n FROM fabric_slots WHERE device = ?', device).one().n; + } + + #queueCount(device) { + return this.sql.exec('SELECT COUNT(*) AS n FROM fabric_queue WHERE device = ?', device).one().n; + } + + #prune(device) { + const cutoff = this.clock() - FABRIC_MAX_AGE_MS; + this.sql.exec('DELETE FROM fabric_slots WHERE device = ? AND at <= ?', device, cutoff); + this.sql.exec('DELETE FROM fabric_queue WHERE device = ? AND at <= ?', device, cutoff); + } +} diff --git a/relay/hosted/src/link/decide.js b/relay/hosted/src/link/decide.js index df2ff90f14..d1a42cda5d 100644 --- a/relay/hosted/src/link/decide.js +++ b/relay/hosted/src/link/decide.js @@ -29,6 +29,7 @@ async function certDevice(c, cert) { /** checkApprovalDevice validates the certificate before a request can be reserved. */ export async function checkApprovalDevice(c, body, request) { if (!isObject(body) || !isObject(body.device)) throw badRequest(); + if ((body.device.kind ?? '') !== '' || (body.device.methods?.length ?? 0) > 0) throw badRequest(); if ((await certDevice(c, body.cert)) !== request.device) throw badRequest(); } diff --git a/relay/hosted/src/raft/hosted.js b/relay/hosted/src/raft/hosted.js index a91618f7d8..40b8c3ad31 100644 --- a/relay/hosted/src/raft/hosted.js +++ b/relay/hosted/src/raft/hosted.js @@ -348,7 +348,7 @@ export class HostedIdentity { await this.rememberRetirement(state.identity); if(this.notifiedVersion!==version){this.watchers.broadcast(String(version));this.notifiedVersion=version;} if(state.identity.rotation){this.watchers.closeAll(CLOSE_ROTATED,'rotated');for(const device of this.pendingIO.keys())this.stopIO(device,'rotated');} - for(const device of state.revoked){this.watchers.closeDevice(device,CLOSE_REVOKED,'revoked');this.stopIO(device,'revoked');} + for(const device of state.revoked){this.tenant.fabric.revokeClient(device);this.watchers.closeDevice(device,CLOSE_REVOKED,'revoked');this.stopIO(device,'revoked');} } if(this.retireAt)await this.armAlarm(this.retireAt); } @@ -451,10 +451,14 @@ export class HostedIdentity { async barrierOutbox(after,limit){const b=await this.barrier();return (await this.members.get(b.node).hostView(b.index,'outbox','',after)).rows.slice(0,limit);} projectJSON(kind,raw){return projectRecord(this.runtime,kind,raw);} project(kind,raw){return JSON.parse(this.projectJSON(kind,raw));} - async authorize(device,certificate,write=false,fresh=false){ + async authorize(device,certificate,write=false,fresh=false,allowClient=false){ await this.wakeMaintenance(); const b=await this.readAuthority(fresh),root=await this.readView(b,'state','identity'),actor=await this.readView(b,'state','device/'+segment(device)); - const r=response(this.runtime.call(frame(32).blob(root.raw).blob(actor.raw??new Uint8Array()).str(device).str(certificate).u32(write?1:0).finish()));requireOK(r.str());r.end();this.noteSeen(device);return {barrier:b,app:root.app,identity:this.project('identity',root.raw)}; + const actorRec=actor.raw?this.project('device',actor.raw):null; + if(!allowClient&&actorRec?.kind==='client')throw new Wire('unauthorized',401); + const r=response(this.runtime.call(frame(32).blob(root.raw).blob(actor.raw??new Uint8Array()).str(device).str(certificate).u32((write?1:0)|(allowClient?2:0)).finish()));requireOK(r.str());r.end(); + if(actorRec?.kind!=='client')this.noteSeen(device); + return {barrier:b,app:root.app,identity:this.project('identity',root.raw),actor:actorRec}; } async recheckAuthority(previous,device,certificate,write,fresh){ const current=await this.readAuthority(fresh); @@ -807,6 +811,7 @@ export class HostedIdentity { if(this.retireAt)await this.armAlarm(this.retireAt); if(q.command.change.revoke){ const device=q.command.change.revoke.device; + this.tenant.fabric.revokeClient(device); this.watchers.announce({t:'revoked',device},device);this.watchers.closeDevice(device,CLOSE_REVOKED,'revoked'); } } @@ -848,9 +853,34 @@ export class HostedIdentity { const storage=unwrap(await this.env.IDENTITY.get(this.env.IDENTITY.idFromName(successor)).inheritStorage(approval?.storage==='on'?approval:null)); return json({successor,storage}); } + async serveFabric(c,route,fresh=false){ + const {request,device,body}=c,cert=request.headers.get('codeaf-cert'),targetDevice=route.args[0]; + await this.ensure(fromBase64Url(request.headers.get('codeaf-identity'))); + await this.wakeMaintenance(); + const authority=await this.authorize(device,cert,route.write,fresh,true),b=authority.barrier; + const now=this.readNow(authority.app.now); + checkTime(request.headers.get('codeaf-time'),now); + const ids=new Set([device,targetDevice]); + if(request.method==='POST'&&body.length){ + try{const parsed=JSON.parse(dec.decode(body));if(typeof parsed?.target==='string'&&parsed.target&&parsed.target!=='any')ids.add(parsed.target);}catch{} + } + const devices={}; + if(authority.actor)devices[device]=authority.actor; + for(const id of ids){ + if(id===device||!id)continue; + const row=await this.readView(b,'state','device/'+segment(id)); + if(row.raw)devices[id]=this.project('device',row.raw); + } + if(devices[targetDevice]?.revoked)this.tenant.fabric.revokeClient(targetDevice); + const dir={read:(kind,id)=>(kind==='devices'?devices[id]:undefined)}; + const reply=await this.tenant.fabric.withClockAndDir(now,dir,()=>route.handler(c,route.args)); + return this.finishRead(b,reply); + } async dispatch(c,route){ - if(new URL(c.request.url).pathname==='/v1/store/handoff')return this.operation(()=>this.retryRead(fresh=>this.handoff(c,fresh))); - if(!new URL(c.request.url).pathname.startsWith('/v1/store/'))return this.operation(()=>c.request.method==='GET'?this.retryRead(fresh=>this.serve(c,route,fresh)):this.serve(c,route)); + const path=new URL(c.request.url).pathname; + if(path==='/v1/store/handoff')return this.operation(()=>this.retryRead(fresh=>this.handoff(c,fresh))); + if(route.fabric&&!(c.request.method==='PUT'&&/^\/v1\/clients\/[^/]+$/.test(path)))return this.operation(()=>this.retryRead(fresh=>this.serveFabric(c,route,fresh))); + if(!path.startsWith('/v1/store/'))return this.operation(()=>c.request.method==='GET'?this.retryRead(fresh=>this.serve(c,route,fresh)):this.serve(c,route)); const cert=c.request.headers.get('codeaf-cert'); const ticket=await this.operation(()=>this.retryRead(async fresh=>{ await this.ensure(fromBase64Url(c.request.headers.get('codeaf-identity'))); @@ -915,6 +945,10 @@ export class HostedIdentity { asGrant(approval.grant,this.tenant.limits.linkMaxGrant); } } + if(request.method==='PUT'&&/^\/v1\/clients\/[^/]+$/.test(path)){ + await this.authorize(device,cert,true,true); + await this.tenant.fabric.validateRegisterClient({targetDevice:decodeURIComponent(route.args[0]),body,identityHeader:request.headers.get('codeaf-identity')}); + } let raw;try{raw=v1Command(this.runtime,this.manifest.config.cluster,this.readNow(),this.proof(request,body));}catch{throw new Wire('bad_request',400);} if(join)await this.effects.admit(joinBox,joinCode,Date.now()); const result=await this.commitV1(raw); @@ -927,6 +961,7 @@ export class HostedIdentity { } if(path.endsWith('/revoke')){ const device=decodeURIComponent(route.args[0]); + this.tenant.fabric.revokeClient(device); this.watchers.announce({t:'revoked',device},device);this.watchers.closeDevice(device,CLOSE_REVOKED,'revoked'); } if(path.endsWith('/release')||path.endsWith('/archive'))return empty(); diff --git a/relay/hosted/src/routes.js b/relay/hosted/src/routes.js index a0e82698ef..8487eccad3 100644 --- a/relay/hosted/src/routes.js +++ b/relay/hosted/src/routes.js @@ -3,6 +3,8 @@ // is refused with, and the handler. Handlers get {tenant, device, body} and the path's captures, // and answer a Response or throw a refusal; the Durable Object does everything around them. import { MAX_FRAME } from './frame.js'; +import { signedUri } from './base.js'; +import { MAX_FABRIC_BODY } from './fabric.js'; import { approve, deny } from './link/decide.js'; import { parseBeat } from './presence.js'; import { parseHolds } from './vouch.js'; @@ -89,6 +91,8 @@ const DIR = { over: 'too_large' }; // `write` marks a route a replaced identity refuses: every verb that changes a record or a frame, and // the watch, which a replaced identity could only ever hear a thaw on (contract 21.2). const LINK = { over: 'too_big', limit: 16 << 10, write: true }; // an approve body: a device record, a cert and a grant of at most 4 KiB +const FABRIC = { over: 'too_large', limit: MAX_FABRIC_BODY, fabric: true }; +const FABRIC_WRITE = { ...FABRIC, write: true }; const STORE_WRITE = { ...STORE, write: true }; const DIR_WRITE = { ...DIR, write: true }; @@ -123,7 +127,51 @@ const ROUTES = [ c.tenant.dir.putDevice(id, object(c.body)); return empty(); }], - ['POST', /^\/v1\/dir\/devices\/([^/]+)\/revoke$/, DIR_WRITE, (c, [id]) => (c.tenant.dir.revoke(id, c.device), empty())], + ['POST', /^\/v1\/dir\/devices\/([^/]+)\/revoke$/, DIR_WRITE, (c, [id]) => (c.tenant.revoke(id, c.device), empty())], + ['PUT', /^\/v1\/clients\/([^/]+)$/, FABRIC_WRITE, async (c, [device]) => { + await c.tenant.fabric.registerClient({ + caller: c.device, + targetDevice: device, + body: c.body, + identityHeader: c.request.headers.get('codeaf-identity'), + identity: c.tenant.identity, + }); + return empty(); + }], + ['PUT', /^\/v1\/clients\/([^/]+)\/slots\/([^/]+)$/, FABRIC_WRITE, (c, [device, name]) => { + c.tenant.fabric.putSlot({ caller: c.device, targetDevice: device, name, body: c.body }); + return empty(); + }], + ['GET', /^\/v1\/clients\/([^/]+)\/slots\/([^/]+)$/, FABRIC, (c, [device, name]) => + json(c.tenant.fabric.getSlot({ caller: c.device, targetDevice: device, name, body: c.body })), + ], + ['POST', /^\/v1\/clients\/([^/]+)\/queue$/, FABRIC_WRITE, (c, [device]) => { + c.tenant.fabric.appendQueue({ + caller: c.device, + targetDevice: device, + body: c.body, + proof: { + signature: c.request.headers.get('codeaf-sig') ?? '', + cert: c.request.headers.get('codeaf-cert') ?? '', + identity: c.request.headers.get('codeaf-identity') ?? '', + requestTime: c.request.headers.get('codeaf-time') ?? '', + requestUri: signedUri(c.env, new URL(c.request.url)), + bodyBytes: c.body, + }, + }); + return empty(); + }], + ['GET', /^\/v1\/clients\/([^/]+)\/queue$/, FABRIC, (c, [device]) => + json({ entries: c.tenant.fabric.listQueue({ caller: c.device, targetDevice: device, body: c.body }) }), + ], + ['POST', /^\/v1\/clients\/([^/]+)\/queue\/([^/]+)\/claim$/, FABRIC_WRITE, (c, [device, number]) => { + c.tenant.fabric.claimQueue({ caller: c.device, targetDevice: device, numberText: number, body: c.body, ack: false }); + return empty(); + }], + ['DELETE', /^\/v1\/clients\/([^/]+)\/queue\/([^/]+)$/, FABRIC_WRITE, (c, [device, number]) => { + c.tenant.fabric.claimQueue({ caller: c.device, targetDevice: device, numberText: number, body: c.body, ack: true }); + return empty(); + }], ['GET', /^\/v1\/dir\/presence$/, DIR, presence], ['POST', /^\/v1\/dir\/requests\/([^/]+)\/approve$/, LINK, async (c, args) => (await approve(c, args), empty())], ['POST', /^\/v1\/dir\/requests\/([^/]+)\/deny$/, LINK, async (c, args) => (await deny(c, args), empty())], @@ -145,11 +193,11 @@ const ROUTES = [ ['POST', /^\/v1\/identity\/rotation$/, DIR, (c) => json(c.tenant.rotate(c.device, object(c.body)))], ]; -/** matchRoute answers {limit, over, frames, write, handler, args} for a request, or throws not_found. */ +/** matchRoute answers {limit, over, frames, write, fabric, handler, args} for a request, or throws not_found. */ export function matchRoute(method, path) { - for (const [m, pattern, { limit = MAX_SMALL, over, frames = false, write = false }, handler] of ROUTES) { + for (const [m, pattern, { limit = MAX_SMALL, over, frames = false, write = false, fabric = false }, handler] of ROUTES) { const hit = m === method && pattern.exec(path); - if (hit) return { limit, over, frames, write, handler, args: hit.slice(1) }; + if (hit) return { limit, over, frames, write, fabric, handler, args: hit.slice(1) }; } throw notFound(); } diff --git a/relay/hosted/src/tenant.js b/relay/hosted/src/tenant.js index d59c9e8f13..c5c578e51b 100644 --- a/relay/hosted/src/tenant.js +++ b/relay/hosted/src/tenant.js @@ -2,6 +2,7 @@ // what the identity asked of it. Route handlers talk to this and to nothing beneath it. import { LegacySource } from './raft/legacy.js'; import { Directory } from './directory.js'; +import { Fabric } from './fabric.js'; import { Quota } from './limits.js'; import { Meta } from './meta.js'; import { StorageAccess } from './storage.js'; @@ -34,6 +35,7 @@ export class Tenant { this.flight = flight; this.watchers = watchers; this.dir = new Directory(sql, identity, clock, policy, watchers); + this.fabric = new Fabric(sql, clock, this.dir); this.quota = new Quota(sql, limits); this.meta = new Meta(sql); this.storage = new StorageAccess(this.meta, limits); @@ -155,6 +157,18 @@ export class Tenant { return rotationView(this.dir.rotation(), this.clock(), this.limits); } + /** isClient answers whether the caller device is registered as a scoped client. */ + isClient(device) { + return this.fabric.isClient(device); + } + + /** revoke stops a device and wipes any client fabric state held for it. */ + revoke(device, caller) { + const wasClient = this.isClient(device); + this.dir.revoke(device, caller); + if (wasClient) this.fabric.revokeClient(device); + } + /** retireAt is the directory time the identity is deleted at, or undefined while it is not retired. */ retireAt() { return this.dir.rotation()?.retire_at; diff --git a/relay/hosted/src/wire.js b/relay/hosted/src/wire.js index a67360ca77..913715d22d 100644 --- a/relay/hosted/src/wire.js +++ b/relay/hosted/src/wire.js @@ -57,7 +57,7 @@ export function wireOf(e) { } export const json = (value, status = 200) => - new Response(JSON.stringify(value), { status, headers: { 'content-type': 'application/json' } }); + new Response(JSON.stringify(value) + '\n', { status, headers: { 'content-type': 'application/json' } }); export const empty = () => new Response(null, { status: 204 }); function refusal(w) { diff --git a/relay/hosted/test/directory.test.js b/relay/hosted/test/directory.test.js index f89a0d5cf3..90773c5012 100644 --- a/relay/hosted/test/directory.test.js +++ b/relay/hosted/test/directory.test.js @@ -1,6 +1,7 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; import { Directory } from '../src/directory.js'; +import { Fabric } from '../src/fabric.js'; import { AMENDED, STAGE1 } from '../src/rules.js'; import { memorySql } from './sql.js'; import { init } from './helpers.js'; @@ -226,3 +227,32 @@ test('stale fences and heads cannot change a folder', () => { assert.deepEqual(dir.cell('c1').cell, before); } }); + +test('putDevice and approveDevice refuse kind and methods, while Fabric slots and revocation respect Raft device state', async () => { + const sql = memorySql(); + const clock = { now: 1_000_000 }; + const dir = new Directory(sql, 'id_t', () => clock.now, AMENDED); + const fabric = new Fabric(sql, () => clock.now, dir); + const full = 'dev_' + 'a'.repeat(32); + const client = 'dev_' + 'b'.repeat(32); + + assert.equal(code(() => dir.putDevice(full, { V: 1, kind: 'client' })), 'bad_request'); + assert.equal(code(() => dir.putDevice(full, { V: 1, methods: ['status.get'] })), 'bad_request'); + assert.equal(code(() => dir.approveDevice(full, { V: 1, kind: 'client' })), 'bad_request'); + assert.equal(code(() => dir.approveDevice(full, { V: 1, methods: ['status.get'] })), 'bad_request'); + + dir.putDevice(full, { V: 1, name: 'workstation' }); + dir.putClientDevice(client, { V: 1, device: client, name: 'phone', methods: ['status.get'] }); + + const enc = new TextEncoder(); + const slotBody = enc.encode(JSON.stringify({ sealed: 'Y2lwaGVydGV4dA==' })); + assert.throws(() => fabric.putSlot({ caller: client, targetDevice: client, name: 'home', body: slotBody })); + fabric.putSlot({ caller: full, targetDevice: client, name: 'home', body: slotBody }); + const got = fabric.getSlot({ caller: client, targetDevice: client, name: 'home', body: new Uint8Array() }); + assert.equal(got.sealed, 'Y2lwaGVydGV4dA=='); + assert.equal(got.writer, full); + + dir.revoke(client, full); + fabric.revokeClient(client); + assert.equal(code(() => fabric.getSlot({ caller: client, targetDevice: client, name: 'home', body: new Uint8Array() })), 'not_found'); +}); diff --git a/scripts/relay-conformance.sh b/scripts/relay-conformance.sh index 78847e23bc..df08f39298 100755 --- a/scripts/relay-conformance.sh +++ b/scripts/relay-conformance.sh @@ -31,6 +31,7 @@ suites=( "bigtake:^TestBigTakeBack$" "watch:^TestWatchConformance$" "lease:^TestLeaseConformance$" + "fabric:^TestFabricConformance$" ) pids=() @@ -147,6 +148,7 @@ track_go() { gotest go-inproc rotation ./internal/relayserve 'Rotated|Frozen|Retire|Sweep' gotest go-inproc watch ./internal/relayserve '^TestWatch' gotest go-inproc lease ./internal/relayserve '^TestLease' + gotest go-inproc fabric ./internal/relayserve '^TestFabricConformance$' fi local relay_bin=${CONFORMANCE_CODEAF:-$root/bin/codeaf} # An explicit candidate avoids rebuilding a binary the caller has already prepared with make build.