From 3d3e75a27b424ae7a6be84d4bb1c36a171a01a05 Mon Sep 17 00:00:00 2001 From: Alexgodoroja Date: Wed, 15 Jul 2026 19:52:48 -0700 Subject: [PATCH 1/5] Add broker-enforced tenancy, access-key gate, and fail-closed guards Managed apps whose partner account cannot be split per user (an account bound to a provider campaign) had no isolation between pilot callers: upstream every caller is the same customer, so the partner cannot tell them apart and nothing else was doing it. Any caller could reference any resource id on the shared account, and account-wide lists were returned verbatim. Tenancy (internal/broker/tenancy.go, owner.go) makes the broker the isolation layer. A resource is claimed by whoever creates it; a reference in the path, query, or body is checked against the ledger before the request is forwarded; list responses are filtered to the caller's rows, and account summaries are recomputed from the caller's ledger. Deny-by-default: a resource with no ledger row belongs to nobody and is refused to everybody. Also closes several defaults that resolved to "allow" when unset: - allow entries are method-scoped, so a registry can express reads-yes/writes-no instead of one verb implying all of them. - creditEnabled keys off the credit block's presence, not a non-zero seed, so seed_credits:0 means a zero budget rather than no metering at all. - max_identities_per_ip defaults to a cap; only an explicit negative disables it. - access keys gate apps that grant something of value; the broker refuses to boot if an app requires a key that is not configured. - store and upstream errors return a correlation id instead of internal hostnames, addresses, and driver state. Tests are written as attacks: cross-tenant use via a body reference, reads of another tenant's resources, ownership takeover, type confusion, parser- differential duplicate keys, and count fields that leak the unfiltered set. --- cmd/broker/main.go | 14 + deploy/agentphone-tenancy.py | 164 +++++ internal/broker/accesskey.go | 123 ++++ internal/broker/broker.go | 101 +++- internal/broker/owner.go | 77 +++ internal/broker/registry.go | 107 +++- internal/broker/store.go | 60 +- internal/broker/store_sqlite.go | 86 +++ internal/broker/tenancy.go | 731 +++++++++++++++++++++++ internal/broker/zz_allow_pattern_test.go | 4 +- internal/broker/zz_dupkey_test.go | 35 ++ internal/broker/zz_tenancy_test.go | 414 +++++++++++++ 12 files changed, 1884 insertions(+), 32 deletions(-) create mode 100644 deploy/agentphone-tenancy.py create mode 100644 internal/broker/accesskey.go create mode 100644 internal/broker/owner.go create mode 100644 internal/broker/tenancy.go create mode 100644 internal/broker/zz_dupkey_test.go create mode 100644 internal/broker/zz_tenancy_test.go diff --git a/cmd/broker/main.go b/cmd/broker/main.go index 261c2ce..486c450 100644 --- a/cmd/broker/main.go +++ b/cmd/broker/main.go @@ -18,6 +18,7 @@ import ( "net/http" "os" "os/signal" + "strings" "syscall" "time" @@ -37,6 +38,17 @@ func main() { log.Fatalf("broker: %v", err) } + // Access keys come from the environment only (never the registry file, which + // is world-readable config and lands in git). + accessKeys := broker.NewAccessKeys(strings.Split(os.Getenv("BROKER_ACCESS_KEYS"), ",")) + // Refuse to boot if an app demands a key that does not exist. Without this the + // broker would come up "healthy" and 401 every caller — a config typo would + // read as a total outage with no explanation. + if n := reg.AppsRequiringAccessKey(); len(n) > 0 && accessKeys.Len() == 0 { + log.Fatalf("broker: apps %v require an access key but BROKER_ACCESS_KEYS is empty — "+ + "set it (comma-separated, entries may be \"label:key\") or clear require_access_key", n) + } + // Durable store when BROKER_DB is set (prod); in-memory otherwise (dev). var store interface { broker.Store @@ -61,6 +73,8 @@ func main() { b := broker.New(reg, store) b.Verify = broker.VerifyConfig{Window: *window} b.IPTrust = broker.IPTrust{Header: *ipHeader} + b.AccessKeys = accessKeys + log.Printf("broker: %d access key(s) configured", accessKeys.Len()) // Usage meter: drain per-user credit by real machine usage, stopping machines // at zero. Runs for provisioned apps whose registry entry carries a rate card. diff --git a/deploy/agentphone-tenancy.py b/deploy/agentphone-tenancy.py new file mode 100644 index 0000000..ee89f5b --- /dev/null +++ b/deploy/agentphone-tenancy.py @@ -0,0 +1,164 @@ +#!/usr/bin/env python3 +"""Rewrite the io.pilot.agentphone registry entry for broker-enforced tenancy. + +The agentphone partner account is a single shared account (bound to a provider +campaign that permits number generation), so the partner cannot tell pilot users +apart and cannot isolate them. The broker therefore owns tenancy: this script +declares which path/body fields name a resource, how each resource is claimed, +and which list responses must be filtered to the caller's own rows. + +Run on the broker host; it rewrites /opt/pilot/registry/apps.json in place after +taking a timestamped backup, and validates the JSON before writing. +""" +import json +import re +import shutil +import sys +import time + +P = "/opt/pilot/registry/apps.json" + +# Routes that cannot be made per-tenant on a shared account. +# +# The webhook routes are ACCOUNT-level: one tenant setting a webhook redirects +# every tenant's events to their endpoint. There is no per-caller scoping to +# apply, so they are removed rather than "checked" — the per-agent route +# /v1/agents/{agent_id}/webhook stays and is gated by agent ownership. +# +# /v1/messages/{message_id}/reactions names a message id that no route can +# establish ownership of, and an unmapped param is an unchecked resource. It is +# dropped until messages carry an ownable link. +DROP = { + "/v1/webhooks", + "/v1/webhooks/deliveries", + "/v1/webhooks/deliveries/stats", + "/v1/webhooks/test", + "/v1/messages/{message_id}/reactions", +} + +# Spend routes remain withheld from the allow-list while the grant for a +# self-minted identity is still unauthenticated. Re-add them together with the +# access-key rollout, not before. +SPEND = {"/v1/messages", "/v1/calls", "/v1/numbers"} + +TENANCY = { + # Every {param} that appears in an allow pattern MUST be mapped here. A param + # with no mapping is never ownership-checked, which is the same as public. + "param_types": { + "agent_id": "agent", + "number_id": "number", + "call_id": "call", + "conversation_id": "conversation", + "contact_id": "contact", + }, + # Body/query fields that name a resource the call acts THROUGH. Sending names + # the sending agent in the body while the path is entirely the caller's own, + # so without these the path checks above prove nothing. Aliases are listed + # because the partner accepts more than one spelling. + "body_refs": { + "agent_id": "agent", + "agentId": "agent", + "number_id": "number", + "numberId": "number", + "phoneNumberId": "number", + "from_number": "number", + "conversation_id": "conversation", + "conversationId": "conversation", + "call_id": "call", + "callId": "call", + "contact_id": "contact", + "contactId": "contact", + }, + "create": [ + {"method": "POST", "path": "/v1/agents", "type": "agent", "id_field": "id"}, + {"method": "POST", "path": "/v1/numbers", "type": "number", "id_field": "id"}, + {"method": "POST", "path": "/v1/calls", "type": "call", "id_field": "id"}, + {"method": "POST", "path": "/v1/contacts", "type": "contact", "id_field": "id"}, + ], + "delete": [ + {"method": "DELETE", "path": "/v1/numbers/{number_id}", "type": "number", "param": "number_id"}, + {"method": "DELETE", "path": "/v1/agents/{agent_id}", "type": "agent", "param": "agent_id"}, + {"method": "DELETE", "path": "/v1/contacts/{contact_id}", "type": "contact", "param": "contact_id"}, + ], + # Only ACCOUNT-WIDE lists are declared here. + # + # A sub-resource list under an owned parent (e.g. /v1/numbers/{number_id}/messages) + # is already authorized by the parent's ownership check, and its elements carry + # no ownable link of their own — declaring it would drop every row and hide the + # owner's own messages from them. + "list": [ + {"method": "GET", "path": "/v1/agents", "array": "data", + "owner_by": [{"field": "id", "type": "agent"}], "claim_as": "agent", + "count_fields": ["total"]}, + {"method": "GET", "path": "/v1/numbers", "array": "data", + "owner_by": [{"field": "id", "type": "number"}], "claim_as": "number", + "count_fields": ["total"]}, + {"method": "GET", "path": "/v1/contacts", "array": "data", + "owner_by": [{"field": "id", "type": "contact"}], "claim_as": "contact", + "count_fields": ["total"]}, + # Inbound calls/conversations are created by the partner, so they are + # attributable only through the number/agent they hang off. claim_as makes + # them fetchable by id afterwards. + {"method": "GET", "path": "/v1/calls", "array": "data", + "owner_by": [{"field": "phoneNumberId", "type": "number"}, + {"field": "agentId", "type": "agent"}], "claim_as": "call", + "count_fields": ["total"]}, + {"method": "GET", "path": "/v1/conversations", "array": "data", + "owner_by": [{"field": "phoneNumberId", "type": "number"}, + {"field": "agentId", "type": "agent"}], "claim_as": "conversation", + "count_fields": ["total"]}, + {"method": "GET", "path": "/v1/usage/by-number", "array": "data", + "owner_by": [{"field": "numberId", "type": "number"}]}, + {"method": "GET", "path": "/v1/usage/by-agent", "array": "data", + "owner_by": [{"field": "agentId", "type": "agent"}]}, + ], + # /v1/usage summarises the SHARED account: its number count and message/call + # totals span every tenant. Left alone it is a side-channel — a caller who can + # see none of the resources can still read how many exist and watch the totals + # move. numbers.used is recomputed from the caller's own ledger; the stats + # block cannot be attributed to a single tenant, so it is dropped rather than + # reported wrongly. + "object": [ + {"method": "GET", "path": "/v1/usage", + "owned_counts": {"numbers.used": "number"}, + # numbers.remaining is the partner's limit minus the ACCOUNT-WIDE used, so + # it discloses the very count numbers.used was just scoped to hide. + # A derived field leaks whatever it was derived from. + "redact": ["stats", "numbers.remaining"]}, + ], +} + + +def main(): + apps = json.load(open(P)) + shutil.copy(P, P + ".bak-" + time.strftime("%Y%m%d-%H%M%S")) + + for a in apps: + if a["id"] != "io.pilot.agentphone": + continue + before = len(a["allow"]) + a["allow"] = [p for p in a["allow"] if p not in DROP and p not in SPEND] + a["tenancy"] = TENANCY + # Cap identities per source IP. This is a speed bump against bulk identity + # creation, not a boundary: it does not constrain a caller with many source + # addresses, so it must never be the only control on a grant. + a.setdefault("credit", {})["max_identities_per_ip"] = 3 + print("allow: %d -> %d" % (before, len(a["allow"]))) + + # Fail loudly if any {param} still lacks an ownership mapping: an unmapped + # param is an unchecked resource, which is exactly the bug class this fixes. + params = set() + for p in a["allow"]: + params |= set(re.findall(r"{(\w+)}", p)) + unmapped = params - set(TENANCY["param_types"]) + if unmapped: + sys.exit("REFUSING: unmapped path params would be unchecked: %s" % sorted(unmapped)) + print("all path params mapped: %s" % sorted(params)) + + json.dump(apps, open(P, "w"), indent=2) + json.load(open(P)) # re-parse: never leave invalid JSON behind + print("registry written + valid") + + +if __name__ == "__main__": + main() diff --git a/internal/broker/accesskey.go b/internal/broker/accesskey.go new file mode 100644 index 0000000..9fb2c48 --- /dev/null +++ b/internal/broker/accesskey.go @@ -0,0 +1,123 @@ +package broker + +import ( + "crypto/sha256" + "crypto/subtle" + "encoding/hex" + "net/http" + "strings" +) + +// Access keys are the broker's front door. +// +// A signed pilot identity (identity.go) proves the caller holds the private key +// for a public key, but the caller generates that keypair itself. It is a +// pseudonym: it attributes usage, it does not authorize it. +// +// The access key is the other half. Identity says WHO you are; the access key +// says you are ALLOWED TO BE HERE. Both are required, and neither substitutes +// for the other: a shared key cannot attribute usage to a user, and a +// self-generated identity cannot establish that a caller is a pilot client at +// all. Any app that grants something of value on first contact needs both. +// +// Keys are compared as SHA-256 digests in constant time, and only the digests +// are held in memory, so the broker never keeps a usable secret it could leak +// through a crash dump or a debug endpoint. +type AccessKeys struct { + digests map[string]string // hex(sha256(key)) -> label, for revocation + attribution +} + +// AccessKeyHeader is the dedicated header. Authorization: Bearer is also +// accepted so ordinary HTTP clients work without a custom header. +const AccessKeyHeader = "X-Pilot-Access-Key" + +// NewAccessKeys builds the key set from "label:key" (or bare "key") entries. +// Blank entries are ignored so a trailing comma in config is not a silent +// "no keys configured" — which would otherwise disable the gate. +func NewAccessKeys(entries []string) *AccessKeys { + ak := &AccessKeys{digests: map[string]string{}} + for _, e := range entries { + e = strings.TrimSpace(e) + if e == "" { + continue + } + label, key := "", e + if i := strings.Index(e, ":"); i > 0 { + label, key = e[:i], e[i+1:] + } + key = strings.TrimSpace(key) + if key == "" { + continue + } + sum := sha256.Sum256([]byte(key)) + ak.digests[hex.EncodeToString(sum[:])] = label + } + return ak +} + +// Len reports how many keys are configured (diagnostics; never logs the keys). +func (a *AccessKeys) Len() int { + if a == nil { + return 0 + } + return len(a.digests) +} + +// presented pulls the candidate key off a request. +func presented(h func(string) string) string { + if v := strings.TrimSpace(h(AccessKeyHeader)); v != "" { + return v + } + // Authorization: Bearer + if v := strings.TrimSpace(h("Authorization")); v != "" { + if len(v) > 7 && strings.EqualFold(v[:7], "bearer ") { + return strings.TrimSpace(v[7:]) + } + } + return "" +} + +// Check verifies the presented key and returns its label. +// +// It FAILS CLOSED in the case that matters most: an AccessKeys with no keys +// configured authorizes NOTHING. A missing or blank BROKER_ACCESS_KEYS must +// never degrade to "open to everyone" — an absent control has to read as "deny", +// not as "allow". main refuses to boot in that state rather than serve. +func (a *AccessKeys) Check(h func(string) string) (label string, ok bool) { + if a == nil || len(a.digests) == 0 { + return "", false + } + key := presented(h) + if key == "" { + return "", false + } + sum := sha256.Sum256([]byte(key)) + want := hex.EncodeToString(sum[:]) + // Constant-time scan over every digest: comparing hex digests of a hash means + // a timing signal cannot reveal the key, and not short-circuiting keeps the + // work independent of which key matched. + found, lbl := 0, "" + for d, l := range a.digests { + if subtle.ConstantTimeCompare([]byte(d), []byte(want)) == 1 { + found, lbl = 1, l + } + } + return lbl, found == 1 +} + +// requireAccessKey enforces the gate for an app that demands one. It answers 401 +// with a WWW-Authenticate hint and no detail about why: a caller learns only +// that a valid key is required, never whether a key exists or was close. +func (b *Broker) requireAccessKey(w http.ResponseWriter, r *http.Request, app *AppEntry) bool { + if !app.RequireAccessKey { + return true + } + if _, ok := b.AccessKeys.Check(r.Header.Get); ok { + return true + } + w.Header().Set("WWW-Authenticate", `Bearer realm="pilot-broker"`) + writeJSON(w, http.StatusUnauthorized, map[string]string{ + "error": "this app requires a pilot access key — upgrade your pilot client (`pilotctl appstore upgrade`) or see https://pilotprotocol.network/docs/access-keys", + }) + return false +} diff --git a/internal/broker/broker.go b/internal/broker/broker.go index 1c8925a..666142c 100644 --- a/internal/broker/broker.go +++ b/internal/broker/broker.go @@ -3,9 +3,12 @@ package broker import ( "bytes" "context" + "crypto/rand" + "encoding/hex" "encoding/json" "fmt" "io" + "log" "math" "net/http" "strconv" @@ -27,6 +30,10 @@ type Broker struct { // IPTrust says which header carries the real source IP (set by the front // proxy). Client-supplied X-Forwarded-For is never trusted. IPTrust IPTrust + // AccessKeys authorizes callers of apps with require_access_key. A signed + // identity only proves possession of a self-minted key; this proves the + // caller is an authorized pilot client at all. + AccessKeys *AccessKeys reg atomic.Pointer[Registry] // hot-swappable so the registry can reload without dropping traffic } @@ -82,11 +89,14 @@ func (b *Broker) serveCreditBalance(w http.ResponseWriter, r *http.Request, app if err != nil { switch err { case ErrIPCap: - writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "per-IP identity cap reached — too many pilot identities have claimed a budget from this network"}) + // Deliberately vague: naming the per-IP cap as the reason tells a caller + // exactly which dimension to vary to evade it. The specific cause is + // logged, not returned. + writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "rate limited — try again later"}) case ErrCooldown: writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "re-provision cooldown — retry shortly"}) default: - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "balance: " + err.Error()}) + b.internalError(w, http.StatusInternalServerError, app.ID, "balance", err) } return } @@ -118,6 +128,13 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } + // Access key BEFORE anything else that costs or grants: an unauthorized caller + // must not be able to reach the credit ledger, seed a grant, or touch the + // master key. This is the gate that self-minted identities cannot walk past. + if !b.requireAccessKey(w, r, app) { + return + } + maxBody := b.MaxBody if app.Provision != nil && mpath == app.Provision.PushPath { maxBody = app.Provision.ArtifactMaxBytes @@ -154,11 +171,26 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { } // 3. Is this an allowed method? (no open proxy onto the master key) - if !app.allowed(mpath) { + if !app.allowed(r.Method, mpath) { writeJSON(w, http.StatusForbidden, map[string]string{"error": "method not allowed for this app: " + mpath}) return } + // 3b. TENANCY: on a shared partner account the broker is the only thing that + // knows which pilot user owns what. Every resource named in the path, + // query, or body must belong to this caller. This runs BEFORE the forward, + // so a refused write never reaches the partner. + // + // The answer is 404, never 403: "not yours" and "does not exist" must be + // indistinguishable, otherwise the broker is an oracle for enumerating + // other tenants' resource ids. + if app.Tenancy != nil { + if _, ok := app.Tenancy.EnforceRequest(b.ownerStore(), appID, app.allowSegs, r.Method, mpath, r.URL.RawQuery, body, string(caller)); !ok { + writeJSON(w, http.StatusNotFound, map[string]string{"error": "not found"}) + return + } + } + // 4. Circuit breaker: if the partner has been failing, fail fast and don't // spend a credit or touch the master key. if !app.breaker.Allow() { @@ -187,18 +219,20 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } ip := clientIP(r.Header.Get, r.RemoteAddr, b.IPTrust) - // Seed the caller on first sight, enforcing the per-IP identity cap: once a - // caller depletes their budget (402 below) they can't farm a fresh $5 grant by - // minting a new pilot identity from the same IP — the (N+1)th distinct caller - // on that IP is refused here with 429. + // Seed the caller on first sight, enforcing the per-IP identity cap: the + // (N+1)th distinct caller seen on one source IP is refused here with 429, so + // a depleted caller cannot immediately re-seed from the same network. if _, err := ps.Provision(appID, string(caller), ip, app.creditSeed, app.creditMaxPerIP, app.creditMintCooldown, b.now()); err != nil { switch err { case ErrIPCap: - writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "per-IP identity cap reached — too many pilot identities have claimed a budget from this network"}) + // Deliberately vague: naming the per-IP cap as the reason tells a caller + // exactly which dimension to vary to evade it. The specific cause is + // logged, not returned. + writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "rate limited — try again later"}) case ErrCooldown: writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "re-provision cooldown — retry shortly"}) default: - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "provision: " + err.Error()}) + b.internalError(w, http.StatusInternalServerError, appID, "provision", err) } return } @@ -211,7 +245,7 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { if billable { bal, derr := ps.Credit(appID, string(caller)) if derr != nil { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "credit: " + derr.Error()}) + b.internalError(w, http.StatusInternalServerError, appID, "credit", derr) return } if bal <= 0 { @@ -227,7 +261,7 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { creditCost = app.costForCall(r.Method, mpath) admittedCredit, remaining, derr := ps.Debit(appID, string(caller), creditCost) if derr != nil { - writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "debit: " + derr.Error()}) + b.internalError(w, http.StatusInternalServerError, appID, "debit", derr) return } if !admittedCredit { @@ -262,7 +296,7 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { ureq, err := http.NewRequestWithContext(ctx, r.Method, target, bytes.NewReader(body)) if err != nil { refundCredit() - writeJSON(w, http.StatusBadGateway, map[string]string{"error": "build upstream: " + err.Error()}) + b.internalError(w, http.StatusBadGateway, appID, "build upstream", err) return } ureq.Header.Set("Content-Type", "application/json") @@ -272,7 +306,7 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { if err != nil { app.breaker.Record(false) refundCredit() - writeJSON(w, http.StatusBadGateway, map[string]string{"error": "upstream: " + err.Error()}) + b.internalError(w, http.StatusBadGateway, appID, "upstream", err) return } defer resp.Body.Close() @@ -296,6 +330,21 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } + // 7. TENANCY on the way back out. Two jobs, both only on success: + // - claim what this call created, so the caller owns it going forward; + // - filter list answers down to the caller's own resources, because the + // partner answered for the WHOLE shared account. + if app.Tenancy != nil && resp.StatusCode/100 == 2 { + os := b.ownerStore() + app.Tenancy.ClaimFrom(os, appID, r.Method, mpath, rb, string(caller), b.now()) + app.Tenancy.ReleaseFrom(os, appID, app.allowSegs, r.Method, mpath) + if filtered, did := app.Tenancy.FilterResponse(os, appID, r.Method, mpath, rb, string(caller), b.now()); did { + rb = filtered + } else if filtered, did := app.Tenancy.FilterObject(os, appID, r.Method, mpath, rb, string(caller)); did { + rb = filtered + } + } + // Surface the caller's remaining budget on every metered response. if app.creditEnabled() { if bal, err := ps.Credit(appID, string(caller)); err == nil { @@ -307,3 +356,29 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { w.WriteHeader(resp.StatusCode) _, _ = w.Write(rb) } + +// internalError logs the real cause server-side and returns an opaque reference +// to the caller. +// +// Raw error strings from the store or the upstream dialer describe internal +// topology — hostnames, private addresses, database paths, driver states. None +// of that helps a legitimate caller (it is never something they can fix), and +// all of it helps someone mapping the deployment. The caller gets a correlation +// id to quote in a bug report; the detail stays in the log. +func (b *Broker) internalError(w http.ResponseWriter, code int, appID, stage string, err error) { + ref := errorRef() + log.Printf("broker: %s: %s: ref=%s: %v", appID, stage, ref, err) + writeJSON(w, code, map[string]string{ + "error": "the broker could not complete this call", + "ref": ref, + }) +} + +// errorRef mints a short, non-guessable correlation id for one failure. +func errorRef() string { + var b [6]byte + if _, err := rand.Read(b[:]); err != nil { + return "unknown" + } + return hex.EncodeToString(b[:]) +} diff --git a/internal/broker/owner.go b/internal/broker/owner.go new file mode 100644 index 0000000..b76414f --- /dev/null +++ b/internal/broker/owner.go @@ -0,0 +1,77 @@ +package broker + +import ( + "crypto/subtle" + "errors" + "time" +) + +// constantTimeEqual compares two identity strings without leaking their contents +// through timing. Owner ids are public keys, but comparing them in constant time +// costs nothing and keeps the authorization predicate free of an oracle. +func constantTimeEqual(a, b string) bool { + return subtle.ConstantTimeCompare([]byte(a), []byte(b)) == 1 +} + +// OwnerStore is the broker-side resource-ownership ledger: it records WHICH +// caller owns WHICH upstream resource id, for apps whose partner account cannot +// be split per user. +// +// Why this exists: some managed apps are one shared partner account that cannot +// be split per user (e.g. an account bound to a provider campaign). Upstream, +// every pilot user is the same customer, so the partner cannot tell which user a +// resource belongs to and cannot enforce isolation. The broker therefore has to: +// every resource is claimed by its creator, and anything not provably owned by +// the caller is invisible to them. +// +// Deny-by-default is the whole point: a resource with NO ledger row is owned by +// NOBODY and is refused to EVERYONE (including pre-existing resources created +// before this ledger shipped). That is deliberate — it is what makes the legacy +// shared test number unreachable rather than up for grabs. +type OwnerStore interface { + // Claim records caller as the owner of (app, rtype, rid). It is idempotent + // and FIRST-WRITER-WINS: if a different caller already owns the resource the + // existing owner is kept and ErrOwned is returned. It must never silently + // transfer ownership — that would be a takeover primitive. + Claim(app, rtype, rid, caller string, now time.Time) error + // OwnerOf returns the recorded owner of a resource, if any. + OwnerOf(app, rtype, rid string) (caller string, found bool, err error) + // OwnedSet returns the set of resource ids of rtype owned by caller. Used to + // filter list responses down to the caller's own resources. + OwnedSet(app, rtype, caller string) (map[string]bool, error) + // Release drops a resource row (e.g. a number was released upstream) so the + // id can be re-claimed if the partner recycles it. + Release(app, rtype, rid string) error +} + +// ErrOwned is returned by Claim when the resource already belongs to a different +// caller. It signals an attempted takeover, not a routine race. +var ErrOwned = errors.New("broker: resource already owned by another caller") + +// Owns reports whether caller owns (app, rtype, rid). It is the single +// authorization predicate for tenancy and FAILS CLOSED: any store error, or an +// unknown/unclaimed resource, returns false. Callers must treat false as "deny". +func Owns(s OwnerStore, app, rtype, rid, caller string) bool { + if s == nil || rid == "" || caller == "" { + return false + } + owner, found, err := s.OwnerOf(app, rtype, rid) + if err != nil || !found { + return false + } + return constantTimeEqual(owner, caller) +} + +// ownerStore returns the tenancy ledger if the configured Store implements it. +// +// It returns nil when the store cannot record ownership. Every tenancy check +// treats a nil ledger as "deny" (see Owns / EnforceRequest), so an app that +// declares tenancy against a store that cannot enforce it fails CLOSED — it +// refuses traffic instead of silently serving it unisolated. +func (b *Broker) ownerStore() OwnerStore { + os, ok := b.Store.(OwnerStore) + if !ok { + return nil + } + return os +} diff --git a/internal/broker/registry.go b/internal/broker/registry.go index 15e229b..78f0c7e 100644 --- a/internal/broker/registry.go +++ b/internal/broker/registry.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "os" + "sort" "strings" "time" ) @@ -41,10 +42,23 @@ type AppEntry struct { // minting. nil ⇒ no budget (call-count Quota still applies). Credit *CreditSpec `json:"credit,omitempty"` - master string // resolved from KeyEnv at load (managed: partner key; provisioned: cloud master, e.g. smk_) - injector AuthInjector // built from AuthHeader/Scheme - allowSet map[string]bool - allowPatterns [][]string // templated allow entries split on "/" ("{x}" matches any one segment) + // Tenancy, when set, makes the broker enforce per-caller resource isolation + // on a SHARED partner account: every resource is claimed by its creator and a + // caller may only reference/see resources it owns (see tenancy.go). Required + // for any app whose partner account cannot be split per user. + Tenancy *Tenancy `json:"tenancy,omitempty"` + + // RequireAccessKey gates this app behind a shared pilot access key (see + // accesskey.go), on top of the signed identity. Set it for any app that hands + // out something of value (a credit grant, a phone number) to a caller who is + // otherwise just a self-minted keypair. + RequireAccessKey bool `json:"require_access_key"` + + master string // resolved from KeyEnv at load (managed: partner key; provisioned: cloud master, e.g. smk_) + injector AuthInjector // built from AuthHeader/Scheme + allowSet map[string]bool // key = costKey(method, path); method "" = any method + allowPatterns []allowPattern // templated allow entries ("{x}" matches any one segment) + allowSegs [][]string // every templated allow path (method-independent), for tenancy param extraction breaker *Breaker creditSeed int // Credit.SeedCredits (0 ⇒ no budget) @@ -165,6 +179,17 @@ type CreditSpec struct { BalancePath string `json:"balance_path"` } +// Safe defaults for the credit/Sybil guards. These apply when a registry omits +// the field, so a new app cannot ship with the guard silently off. +const ( + // defaultMaxIdentitiesPerIP caps distinct pilot identities that may claim a + // budget from one source IP. Low enough to make bulk identity creation + // costly, high enough for a shared NAT / office egress to onboard real users. + // It is a speed bump, not a boundary: a caller with many source addresses is + // not constrained by it, so it must never be the only control on a grant. + defaultMaxIdentitiesPerIP = 3 +) + // costPattern is a templated cost key split into segments (like allowPatterns), // optionally scoped to one HTTP method ("" = any method). type costPattern struct { @@ -174,7 +199,13 @@ type costPattern struct { } // creditEnabled reports whether this app meters a per-caller micro-dollar budget. -func (a *AppEntry) creditEnabled() bool { return a.creditSeed > 0 } +// creditEnabled reports whether the per-caller budget ledger is active. +// +// It keys off the PRESENCE of the credit block, never off the seed amount, so +// that the ledger's existence and the size of a grant stay independent settings. +// `seed_credits: 0` therefore means exactly what it reads like — a zero budget, +// in which every priced path is refused with 402 — rather than "no metering". +func (a *AppEntry) creditEnabled() bool { return a.Credit != nil } // costKey is the exact-map key for a (method, path) cost entry ("" method = any). func costKey(method, path string) string { return method + " " + path } @@ -224,22 +255,36 @@ func (a *AppEntry) costForCall(method, path string) int { // "/v1/calls/{call_id}") match any single non-empty segment in that position, so // REST path params don't each need enumerating. An empty allow-list permits // nothing (safe default; prod must declare). -func (a *AppEntry) allowed(path string) bool { +// Entries are METHOD-SCOPED: "POST /v1/messages" allows only that verb, while a +// bare "/v1/messages" allows any verb on that path (backwards compatible). Method +// scoping is what lets a registry express "reads yes, writes no" — without it, +// allowing GET /v1/numbers necessarily also allows DELETE /v1/numbers/{id}. +func (a *AppEntry) allowed(method, path string) bool { if len(a.allowSet) == 0 && len(a.allowPatterns) == 0 { return false } - if a.allowSet[path] { + method = strings.ToUpper(method) + if a.allowSet[costKey(method, path)] || a.allowSet[costKey("", path)] { return true } segs := strings.Split(path, "/") for _, pat := range a.allowPatterns { - if segmentsMatch(pat, segs) { + if pat.method != "" && pat.method != method { + continue + } + if segmentsMatch(pat.segs, segs) { return true } } return false } +// allowPattern is a templated allow entry; method "" matches any verb. +type allowPattern struct { + method string + segs []string +} + // segmentsMatch reports whether request segments satisfy a templated pattern. A // "{name}" pattern segment matches any single non-empty segment; every other // segment must match literally. Lengths must be equal (no implicit wildcards). @@ -309,11 +354,16 @@ func ParseRegistry(raw []byte, getenv func(string) string) (*Registry, error) { a.injector = injectorFor(a.AuthStyle, a.AuthHeader, a.AuthScheme, a.AuthParam, a.AuthUser) a.allowSet = map[string]bool{} a.allowPatterns = nil - for _, p := range a.Allow { + a.allowSegs = nil + for _, entry := range a.Allow { + method, p := parseCostKey(entry) // "POST /v1/x" → ("POST","/v1/x"); "/v1/x" → ("","/v1/x") + method = strings.ToUpper(method) if strings.Contains(p, "{") { - a.allowPatterns = append(a.allowPatterns, strings.Split(p, "/")) + segs := strings.Split(p, "/") + a.allowPatterns = append(a.allowPatterns, allowPattern{method: method, segs: segs}) + a.allowSegs = append(a.allowSegs, segs) } else { - a.allowSet[p] = true + a.allowSet[costKey(method, p)] = true } } if a.CostField == "" { @@ -330,6 +380,12 @@ func ParseRegistry(raw []byte, getenv func(string) string) (*Registry, error) { return nil, err } } + if a.Tenancy != nil { + if err := validateTenancy(a); err != nil { + return nil, err + } + a.Tenancy.compile() + } if a.Credit != nil { if a.Provision != nil { return nil, fmt.Errorf("registry: app %s: `credit` (HTTP budget) and `provision` (cloud) are mutually exclusive", a.ID) @@ -342,10 +398,21 @@ func ParseRegistry(raw []byte, getenv func(string) string) (*Registry, error) { if a.creditDefault < 0 { a.creditDefault = 0 } + // Per-IP identity cap. Defaults CLOSED: omitted means the safe default, + // and only an explicit negative disables the guard, so switching it off is + // always a deliberate and visible choice rather than an oversight. a.creditMaxPerIP = a.Credit.MaxIdentitiesPerIP - if a.creditMaxPerIP < 0 { - a.creditMaxPerIP = 0 + switch { + case a.creditMaxPerIP == 0: + a.creditMaxPerIP = defaultMaxIdentitiesPerIP + case a.creditMaxPerIP < 0: + a.creditMaxPerIP = 0 // explicit opt-out: unlimited } + // NOTE: mint_cooldown deliberately has NO default. Provision is touched on + // EVERY call (not just on re-seed), and its cooldown check rejects any + // touch inside the window — so a non-zero default would 429 every caller's + // second call. It stays opt-in for apps that really want re-mint churn + // control. The Sybil guard here is max_identities_per_ip, above. a.creditMintCooldown = time.Duration(a.Credit.MintCooldownMs) * time.Millisecond a.creditRespCost = a.Credit.CostSource == "response" a.creditCostScale = a.Credit.CostScale @@ -427,3 +494,17 @@ func resolveProvision(a *AppEntry, getenv func(string) string) error { } return nil } + +// AppsRequiringAccessKey lists apps whose registry entry demands an access key. +// main uses it to refuse to boot when no keys are configured, so a config +// mistake surfaces as a loud startup failure rather than a silent 401 wall. +func (r *Registry) AppsRequiringAccessKey() []string { + var out []string + for id, a := range r.apps { + if a.RequireAccessKey { + out = append(out, id) + } + } + sort.Strings(out) + return out +} diff --git a/internal/broker/store.go b/internal/broker/store.go index 803cefb..e43b6c9 100644 --- a/internal/broker/store.go +++ b/internal/broker/store.go @@ -2,6 +2,7 @@ package broker import ( "errors" + "strings" "sync" "time" ) @@ -97,13 +98,64 @@ type cell struct { // MemStore is the default in-memory Store + ProvisionStore (single instance, // non-durable). Used in dev and tests; prod uses SQLiteStore for durability. type MemStore struct { - mu sync.Mutex - m map[string]*cell - prov map[string]*ProvisionRecord // keyed "app|caller" + mu sync.Mutex + m map[string]*cell + prov map[string]*ProvisionRecord // keyed "app|caller" + owner map[string]string // keyed "app|rtype|rid" -> caller (tenancy ledger) } func NewMemStore() *MemStore { - return &MemStore{m: map[string]*cell{}, prov: map[string]*ProvisionRecord{}} + return &MemStore{m: map[string]*cell{}, prov: map[string]*ProvisionRecord{}, owner: map[string]string{}} +} + +// ownKey keys the ownership ledger. rtype is part of the key so ids are never +// confused across resource types (an agent id must not authorize a number). +func ownKey(app, rtype, rid string) string { return app + "|" + rtype + "|" + rid } + +// Claim is first-writer-wins under the lock: a second caller claiming the same +// resource gets ErrOwned rather than taking it over. +func (s *MemStore) Claim(app, rtype, rid, caller string, _ time.Time) error { + if rid == "" || caller == "" { + return errors.New("broker: claim requires a resource id and caller") + } + s.mu.Lock() + defer s.mu.Unlock() + k := ownKey(app, rtype, rid) + if cur, ok := s.owner[k]; ok { + if cur == caller { + return nil // idempotent re-claim by the same owner + } + return ErrOwned + } + s.owner[k] = caller + return nil +} + +func (s *MemStore) OwnerOf(app, rtype, rid string) (string, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + c, ok := s.owner[ownKey(app, rtype, rid)] + return c, ok, nil +} + +func (s *MemStore) OwnedSet(app, rtype, caller string) (map[string]bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + out := map[string]bool{} + prefix := app + "|" + rtype + "|" + for k, c := range s.owner { + if c == caller && strings.HasPrefix(k, prefix) { + out[strings.TrimPrefix(k, prefix)] = true + } + } + return out, nil +} + +func (s *MemStore) Release(app, rtype, rid string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.owner, ownKey(app, rtype, rid)) + return nil } func (s *MemStore) Provision(app, caller, ip string, seed, maxPerIP int, cooldown time.Duration, now time.Time) (ProvisionRecord, error) { diff --git a/internal/broker/store_sqlite.go b/internal/broker/store_sqlite.go index 4145a69..1583b75 100644 --- a/internal/broker/store_sqlite.go +++ b/internal/broker/store_sqlite.go @@ -2,6 +2,7 @@ package broker import ( "database/sql" + "errors" "fmt" "strings" "time" @@ -70,6 +71,26 @@ func OpenSQLiteStore(path string) (*SQLiteStore, error) { _ = db.Close() return nil, fmt.Errorf("sqlite migrate provision.rot: %w", err) } + // ownership: the tenancy ledger — which caller owns which upstream resource, + // for apps on a shared partner account. The PRIMARY KEY is what enforces + // first-writer-wins: a concurrent takeover attempt loses on the unique + // constraint rather than on an application-level check-then-write race. + if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS ownership ( + app TEXT NOT NULL, + rtype TEXT NOT NULL, + rid TEXT NOT NULL, + caller TEXT NOT NULL, + created INTEGER NOT NULL, + PRIMARY KEY (app, rtype, rid) + )`); err != nil { + _ = db.Close() + return nil, fmt.Errorf("sqlite migrate ownership: %w", err) + } + // Listing a caller's own resources is the hot path for response filtering. + if _, err := db.Exec(`CREATE INDEX IF NOT EXISTS ownership_owner ON ownership(app, rtype, caller)`); err != nil { + _ = db.Close() + return nil, fmt.Errorf("sqlite migrate ownership index: %w", err) + } if _, err := db.Exec(`CREATE INDEX IF NOT EXISTS provision_app_ip ON provision(app, ip)`); err != nil { _ = db.Close() return nil, fmt.Errorf("sqlite migrate provision index: %w", err) @@ -309,3 +330,68 @@ func (s *SQLiteStore) Snapshot() map[string]struct { } return out } + +// Claim records ownership of a resource, first-writer-wins. +// +// The INSERT relies on the (app, rtype, rid) PRIMARY KEY: two concurrent callers +// racing to claim the same id cannot both succeed, so ownership can never be +// silently transferred. A repeat claim by the SAME caller is idempotent (no +// error); a claim by a DIFFERENT caller returns ErrOwned. +func (s *SQLiteStore) Claim(app, rtype, rid, caller string, now time.Time) error { + if rid == "" || caller == "" { + return errors.New("broker: claim requires a resource id and caller") + } + res, err := s.db.Exec( + `INSERT INTO ownership (app, rtype, rid, caller, created) VALUES (?,?,?,?,?) + ON CONFLICT(app, rtype, rid) DO NOTHING`, + app, rtype, rid, caller, now.Unix()) + if err != nil { + return err + } + if n, err := res.RowsAffected(); err == nil && n > 0 { + return nil // we inserted: we are the owner + } + // Nothing inserted → a row already exists. It is ours only if the owner matches. + var cur string + if err := s.db.QueryRow(`SELECT caller FROM ownership WHERE app=? AND rtype=? AND rid=?`, app, rtype, rid).Scan(&cur); err != nil { + return err + } + if cur == caller { + return nil + } + return ErrOwned +} + +func (s *SQLiteStore) OwnerOf(app, rtype, rid string) (string, bool, error) { + var caller string + err := s.db.QueryRow(`SELECT caller FROM ownership WHERE app=? AND rtype=? AND rid=?`, app, rtype, rid).Scan(&caller) + if err == sql.ErrNoRows { + return "", false, nil + } + if err != nil { + return "", false, err + } + return caller, true, nil +} + +func (s *SQLiteStore) OwnedSet(app, rtype, caller string) (map[string]bool, error) { + rows, err := s.db.Query(`SELECT rid FROM ownership WHERE app=? AND rtype=? AND caller=?`, app, rtype, caller) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[string]bool{} + for rows.Next() { + var rid string + if err := rows.Scan(&rid); err != nil { + return nil, err + } + out[rid] = true + } + return out, rows.Err() +} + +func (s *SQLiteStore) Release(app, rtype, rid string) error { + _, err := s.db.Exec(`DELETE FROM ownership WHERE app=? AND rtype=? AND rid=?`, app, rtype, rid) + return err +} diff --git a/internal/broker/tenancy.go b/internal/broker/tenancy.go new file mode 100644 index 0000000..6e5ad71 --- /dev/null +++ b/internal/broker/tenancy.go @@ -0,0 +1,731 @@ +package broker + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "net/url" + "strconv" + "strings" + "time" +) + +// Tenancy makes ONE shared partner account behave like N isolated tenants. +// +// Some managed apps cannot give each pilot user their own partner account — e.g. +// io.pilot.agentphone, whose account is bound to a provider campaign that allows +// number generation. Upstream, every pilot user is literally the same customer. +// The partner therefore cannot enforce isolation, so the broker must. +// +// The model has exactly one rule: a caller may only touch a resource it OWNS. +// - Creating a resource claims it (owner = creator) — see claimFrom. +// - Referencing a resource (in the path, the query, or the body) is checked +// against the ledger BEFORE the request is forwarded — see EnforceRequest. +// - Listing resources filters the partner's account-wide answer down to the +// caller's own rows — see FilterResponse. +// +// Deny-by-default is load-bearing. A resource with no ledger row belongs to +// nobody and is refused to everybody. That is what neutralises resources created +// before this shipped (the shared test number/agent): rather than being up for +// grabs, they become unreachable to every tenant. +type Tenancy struct { + // ParamTypes maps a path-template param name to a resource type, e.g. + // "agent_id" -> "agent". Any {param} in an allow pattern that names a + // resource is ownership-checked. + ParamTypes map[string]string `json:"param_types"` + // BodyRefs maps a JSON body/query field to a resource type, e.g. + // "agent_id" -> "agent". Path checks alone are not sufficient: an operation + // often names the resource it acts THROUGH in the body (the sending agent on + // a send), so an unchecked body field is an unchecked resource. + BodyRefs map[string]string `json:"body_refs"` + // Create declares routes that mint a resource; on 2xx the new id is claimed. + Create []CreateRoute `json:"create"` + // List declares routes whose response must be filtered to owned resources. + List []ListRoute `json:"list"` + // Delete declares routes that destroy a resource; on 2xx the claim is dropped + // so a recycled id can be re-claimed by its next owner. + Delete []DeleteRoute `json:"delete"` + // Object declares routes whose response is a SUMMARY of the shared account + // rather than a list of rows. + Object []ObjectRoute `json:"object"` + + createIdx []compiledRoute + listIdx []compiledList + deleteIdx []compiledRoute + objectIdx []compiledObject +} + +// ObjectRoute rewrites a response that describes the shared ACCOUNT instead of +// returning rows. +// +// Filtering arrays is not sufficient on a shared account: a usage/summary +// endpoint reports totals computed across every tenant. Those totals are a +// side-channel — a caller who can see none of the resources can still read how +// many exist and watch the numbers move as other tenants work. +// +// OwnedCounts replaces a count with the caller's OWN count from the ledger. +// Redact removes a field outright, for aggregates that cannot be attributed to +// one tenant at all (and so can only be wrong or leaky). +type ObjectRoute struct { + Method string `json:"method"` + Path string `json:"path"` + OwnedCounts map[string]string `json:"owned_counts"` // dotted field -> resource type + Redact []string `json:"redact"` // dotted fields to drop +} + +type compiledObject struct { + method string + segs []string + route ObjectRoute +} + +// CreateRoute: a 2xx on Method+Path means the caller created a resource of Type +// whose id is at IDField in the response body. +type CreateRoute struct { + Method string `json:"method"` + Path string `json:"path"` + Type string `json:"type"` + IDField string `json:"id_field"` // dotted path into the response, e.g. "id" or "data.id" +} + +// ListRoute: the response of Method+Path carries an array at Array whose +// elements must be filtered to the caller's own. +// +// OwnerBy is why this is not simply "filter by element id": an INBOUND call or +// message is a resource the tenant never created, so it has no ledger row of its +// own — but it belongs to them because it is attached to a number they own. Each +// OwnerBy is a (field, type) link; an element is kept if ANY link resolves to a +// resource the caller owns. Elements with no resolvable link are DROPPED. +type ListRoute struct { + Method string `json:"method"` + Path string `json:"path"` + Array string `json:"array"` // dotted path to the array, e.g. "data" + OwnerBy []OwnerLink `json:"owner_by"` + // ClaimAs, when set, claims each KEPT element's own id as this resource type. + // + // This is what makes derived resources reachable by id. A resource the partner + // creates (an inbound call) has no ledger row of its own; it is attributable + // only via its link to a resource the caller owns. Claiming it as it is listed + // is what lets a later fetch-by-id be ownership-checked at all — and every + // type named in param_types MUST be claimable somewhere, because a param with + // no resolvable type is a param that is never checked. + ClaimAs string `json:"claim_as"` + // ClaimIDField is where the element's own id lives (default "id"). + ClaimIDField string `json:"claim_id_field"` + // CountFields are sibling fields that describe the SIZE of the array (e.g. + // "total"). They must be recomputed after filtering. + // + // Filtering the array alone is not enough: a count left at the partner's value + // still reports the whole shared account, so a caller who can see none of the + // rows can still read how many exist and watch that number move. Any field + // derived from the unfiltered set is a leak of the unfiltered set. + CountFields []string `json:"count_fields"` +} + +// OwnerLink names a field on a list element and the resource type it points at. +type OwnerLink struct { + Field string `json:"field"` + Type string `json:"type"` +} + +// DeleteRoute: a 2xx on Method+Path destroyed the resource named by Param. +type DeleteRoute struct { + Method string `json:"method"` + Path string `json:"path"` + Type string `json:"type"` + Param string `json:"param"` // which path param holds the id +} + +type compiledRoute struct { + method string + segs []string + route any // CreateRoute | DeleteRoute +} + +type compiledList struct { + method string + segs []string + route ListRoute +} + +// compile pre-splits path templates so matching is allocation-light per request. +func (t *Tenancy) compile() { + t.createIdx = nil + for _, c := range t.Create { + t.createIdx = append(t.createIdx, compiledRoute{strings.ToUpper(c.Method), strings.Split(c.Path, "/"), c}) + } + t.listIdx = nil + for _, l := range t.List { + t.listIdx = append(t.listIdx, compiledList{strings.ToUpper(l.Method), strings.Split(l.Path, "/"), l}) + } + t.deleteIdx = nil + for _, d := range t.Delete { + t.deleteIdx = append(t.deleteIdx, compiledRoute{strings.ToUpper(d.Method), strings.Split(d.Path, "/"), d}) + } + t.objectIdx = nil + for _, o := range t.Object { + t.objectIdx = append(t.objectIdx, compiledObject{strings.ToUpper(o.Method), strings.Split(o.Path, "/"), o}) + } +} + +// FilterObject rewrites an account-summary response so it describes only the +// caller. Returns (body, true) when it handled the route. +// +// Like FilterResponse it fails closed: if the body cannot be parsed or rewritten +// it returns an error document rather than passing the partner's account-wide +// answer through untouched. +func (t *Tenancy) FilterObject(s OwnerStore, app, method, path string, respBody []byte, caller string) ([]byte, bool) { + if t == nil || s == nil { + return respBody, false + } + segs := strings.Split(path, "/") + for _, oi := range t.objectIdx { + o := oi.route + if oi.method != strings.ToUpper(method) || !segmentsMatch(oi.segs, segs) { + continue + } + var v any + dec := json.NewDecoder(strings.NewReader(string(respBody))) + dec.UseNumber() + if dec.Decode(&v) != nil { + return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true + } + for field, rtype := range o.OwnedCounts { + if _, ok := dig(v, field); !ok { + continue + } + owned, err := s.OwnedSet(app, rtype, caller) + if err != nil { + return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true + } + if !setDug(v, field, json.Number(strconv.Itoa(len(owned)))) { + return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true + } + } + for _, field := range o.Redact { + delDug(v, field) + } + out, err := json.Marshal(v) + if err != nil { + return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true + } + return out, true + } + return respBody, false +} + +// delDug removes a dotted field from decoded JSON. +func delDug(v any, dotted string) { + parts := strings.Split(dotted, ".") + cur := v + for i, p := range parts { + m, ok := cur.(map[string]any) + if !ok { + return + } + if i == len(parts)-1 { + delete(m, p) + return + } + cur = m[p] + } +} + +// pathParams extracts {name} -> value by matching path against the app's allow +// patterns. Reusing the allow patterns means a route can never be +// ownership-checked under a template that does not also admit it. +func pathParams(patterns [][]string, path string) map[string]string { + segs := strings.Split(path, "/") + out := map[string]string{} + for _, pat := range patterns { + if !segmentsMatch(pat, segs) { + continue + } + for i, p := range pat { + if len(p) > 2 && p[0] == '{' && p[len(p)-1] == '}' { + out[p[1:len(p)-1]] = segs[i] + } + } + // Keep scanning: several templates may match, and every {param} any of + // them binds must be checked. Being permissive here would let a caller + // pick the template with the fewest checks. + } + return out +} + +// dig walks a dotted path ("data.id") through decoded JSON. +func dig(v any, dotted string) (any, bool) { + cur := v + for _, part := range strings.Split(dotted, ".") { + m, ok := cur.(map[string]any) + if !ok { + return nil, false + } + cur, ok = m[part] + if !ok { + return nil, false + } + } + return cur, true +} + +// asID renders a JSON scalar as a resource id. Only strings and numbers are +// ids; anything else (object, array, bool, null) is not, and returns false. +func asID(v any) (string, bool) { + switch x := v.(type) { + case string: + return x, x != "" + case json.Number: + return x.String(), true + case float64: + // Non-integral or huge floats are not ids; avoid a lossy render. + if x != float64(int64(x)) { + return "", false + } + return json.Number(strings.TrimSuffix(strings.TrimRight(jsonNum(x), "0"), ".")).String(), true + } + return "", false +} + +func jsonNum(f float64) string { + b, _ := json.Marshal(f) + return string(b) +} + +// refusal is why a request was denied. It is deliberately opaque to the caller: +// EnforceRequest's 404 must not distinguish "not yours" from "does not exist", +// or the broker becomes an oracle for enumerating other tenants' resource ids. +type refusal struct { + Type string + ID string +} + +// EnforceRequest is the authorization gate: it runs BEFORE the request is +// forwarded, so a rejected write never reaches the partner and never has a side +// effect. It collects every resource reference in the path, the query string, +// and the JSON body, and requires the caller to own each one. +// +// It FAILS CLOSED: an unparseable body, an unknown id, or a store error all deny. +func (t *Tenancy) EnforceRequest(s OwnerStore, app string, allowPatterns [][]string, method, path, rawQuery string, body []byte, caller string) (*refusal, bool) { + if t == nil { + return nil, true + } + if s == nil || caller == "" { + return &refusal{}, false // no ledger or no identity → deny + } + + // 1. Path params: /v1/agents/{agent_id}/... — the id is in the URL. + for name, val := range pathParams(allowPatterns, path) { + rtype, ok := t.ParamTypes[name] + if !ok { + continue // param names no resource (e.g. a filter) → nothing to own + } + if !Owns(s, app, rtype, val, caller) { + return &refusal{Type: rtype, ID: val}, false + } + } + + // 2. Query params: ?agent_id=... — a filter can also be a reference. + if rawQuery != "" { + q, err := url.ParseQuery(rawQuery) + if err != nil { + return &refusal{}, false // unparseable query → deny rather than skip + } + for field, vals := range q { + rtype, ok := t.BodyRefs[field] + if !ok { + continue + } + for _, v := range vals { + if v == "" { + continue + } + if !Owns(s, app, rtype, v, caller) { + return &refusal{Type: rtype, ID: v}, false + } + } + } + } + + // 3. Body refs: {"agent_id": "..."}. An operation that names the resource it + // acts through in the body must be checked here, or path-level isolation + // is decorative. + if len(body) > 0 && len(t.BodyRefs) > 0 { + // PARSER DIFFERENTIAL. The broker validates the body with Go's decoder but + // forwards the RAW bytes, so the partner re-parses them with a different + // parser. Go keeps the LAST duplicate key; a parser that keeps the FIRST + // would act on a value we never checked: + // + // {"agent_id":"", "agent_id":""} + // + // We would validate "" and approve; the partner would send from + // "". Duplicate keys have no legitimate use here, so a body + // containing any is refused outright rather than reasoned about. + if hasDuplicateKeys(body) { + return &refusal{}, false + } + var v any + dec := json.NewDecoder(strings.NewReader(string(body))) + dec.UseNumber() // keep big ids exact; float64 would corrupt them + if err := dec.Decode(&v); err != nil { + // A body we cannot parse is a body we cannot check. If the app declares + // body refs at all, refuse rather than forward it unchecked. + return &refusal{}, false + } + // Trailing content after the first JSON value ("{...}{...}") is another way + // two parsers can disagree about what the body says. Refuse it. + if dec.More() { + return &refusal{}, false + } + if ref, ok := t.checkRefs(s, app, v, caller, 0); !ok { + return ref, false + } + } + return nil, true +} + +// maxRefDepth bounds recursion into an attacker-supplied body. A deeply nested +// body must not become a stack-exhaustion lever. +const maxRefDepth = 24 + +// checkRefs walks the decoded body and ownership-checks every field named in +// BodyRefs, at ANY depth. Depth matters: a ref nested inside an object must be +// checked too, or wrapping it becomes the bypass. +func (t *Tenancy) checkRefs(s OwnerStore, app string, v any, caller string, depth int) (*refusal, bool) { + if depth > maxRefDepth { + return &refusal{}, false // too deep to verify → deny + } + switch x := v.(type) { + case map[string]any: + for k, val := range x { + if rtype, ok := t.BodyRefs[k]; ok { + if id, isID := asID(val); isID { + if !Owns(s, app, rtype, id, caller) { + return &refusal{Type: rtype, ID: id}, false + } + } else if val != nil { + // The field names a resource but is not a scalar id (e.g. an + // object or array). We cannot verify it → deny. + return &refusal{Type: rtype}, false + } + } + if ref, ok := t.checkRefs(s, app, val, caller, depth+1); !ok { + return ref, false + } + } + case []any: + for _, val := range x { + if ref, ok := t.checkRefs(s, app, val, caller, depth+1); !ok { + return ref, false + } + } + } + return nil, true +} + +// ClaimFrom records ownership after a successful create. It runs only on 2xx, so +// a failed create never claims an id. +func (t *Tenancy) ClaimFrom(s OwnerStore, app, method, path string, respBody []byte, caller string, now time.Time) { + if t == nil || s == nil { + return + } + segs := strings.Split(path, "/") + for _, cr := range t.createIdx { + c := cr.route.(CreateRoute) + if cr.method != strings.ToUpper(method) || !segmentsMatch(cr.segs, segs) { + continue + } + var v any + dec := json.NewDecoder(strings.NewReader(string(respBody))) + dec.UseNumber() + if dec.Decode(&v) != nil { + continue + } + raw, ok := dig(v, c.IDField) + if !ok { + continue + } + if id, isID := asID(raw); isID { + _ = s.Claim(app, c.Type, id, caller, now) + } + } +} + +// ReleaseFrom drops a claim after a successful delete, so a partner-recycled id +// can be claimed by whoever gets it next instead of staying bound to its old +// owner forever. +func (t *Tenancy) ReleaseFrom(s OwnerStore, app string, allowPatterns [][]string, method, path string) { + if t == nil || s == nil { + return + } + segs := strings.Split(path, "/") + params := pathParams(allowPatterns, path) + for _, dr := range t.deleteIdx { + d := dr.route.(DeleteRoute) + if dr.method != strings.ToUpper(method) || !segmentsMatch(dr.segs, segs) { + continue + } + if id := params[d.Param]; id != "" { + _ = s.Release(app, d.Type, id) + } + } +} + +// FilterResponse rewrites a list response to contain only the caller's own +// resources. This is what makes "pilot users must not see numbers other than +// their own" true: the partner answers for the whole shared account, and the +// broker strips it to the caller's rows before it ever reaches them. +// +// It also CLAIMS kept elements (lazy claim). Inbound calls/messages are created +// by the partner, not the tenant, so they have no row yet; claiming them when +// they are provably linked to an owned number is what lets the tenant then fetch +// them by id. +// +// On any doubt it returns an EMPTY array rather than the unfiltered body: +// leaking another tenant's rows is worse than showing none. +func (t *Tenancy) FilterResponse(s OwnerStore, app, method, path string, respBody []byte, caller string, now time.Time) ([]byte, bool) { + if t == nil || s == nil { + return respBody, false + } + segs := strings.Split(path, "/") + for _, li := range t.listIdx { + l := li.route + if li.method != strings.ToUpper(method) || !segmentsMatch(li.segs, segs) { + continue + } + var v any + dec := json.NewDecoder(strings.NewReader(string(respBody))) + dec.UseNumber() + if dec.Decode(&v) != nil { + // Cannot parse → cannot filter → do not pass the partner's raw + // account-wide answer through. + return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true + } + arrRaw, ok := dig(v, l.Array) + if !ok { + return respBody, false // no array here; nothing to filter + } + arr, ok := arrRaw.([]any) + if !ok { + return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true + } + kept := make([]any, 0, len(arr)) + for _, el := range arr { + if t.keepElement(s, app, el, caller, l.OwnerBy, l.ClaimAs, l.ClaimIDField, now) { + kept = append(kept, el) + } + } + if !setDug(v, l.Array, kept) { + return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true + } + // Recompute any count that described the unfiltered set. Note that + // pagination flags (hasMore) are deliberately left alone: they describe the + // partner's underlying paging, and forcing them false would stop a client + // paging before it reached its OWN rows on a later page. + for _, cf := range l.CountFields { + if _, ok := dig(v, cf); ok { + _ = setDug(v, cf, json.Number(strconv.Itoa(len(kept)))) + } + } + out, err := json.Marshal(v) + if err != nil { + return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true + } + return out, true + } + return respBody, false +} + +// keepElement decides if a list element belongs to caller, and lazily claims it. +func (t *Tenancy) keepElement(s OwnerStore, app string, el any, caller string, links []OwnerLink, claimAs, claimIDField string, now time.Time) bool { + m, ok := el.(map[string]any) + if !ok { + return false // not an object → cannot attribute → drop + } + for _, link := range links { + raw, ok := m[link.Field] + if !ok { + continue + } + id, isID := asID(raw) + if !isID { + continue + } + if Owns(s, app, link.Type, id, caller) { + // Provably the caller's. Claim the element itself so a later + // fetch-by-id can be ownership-checked at all. + if claimAs != "" { + field := claimIDField + if field == "" { + field = "id" + } + if selfID, ok := asID(m[field]); ok && selfID != "" { + _ = s.Claim(app, claimAs, selfID, caller, now) + } + } + return true + } + } + return false +} + +// setDug writes a value at a dotted path in decoded JSON. +func setDug(v any, dotted string, val any) bool { + parts := strings.Split(dotted, ".") + cur := v + for i, p := range parts { + m, ok := cur.(map[string]any) + if !ok { + return false + } + if i == len(parts)-1 { + m[p] = val + return true + } + cur, ok = m[p] + if !ok { + return false + } + } + return false +} + +// validateTenancy rejects a tenancy block that would not actually isolate. +// +// This runs at registry load and FAILS THE BOOT rather than warning. A tenancy +// spec is a security control: a typo that silently disables a check is the whole +// bug class we are fixing, so a broken spec must never start serving traffic. +func validateTenancy(a *AppEntry) error { + t := a.Tenancy + if len(t.ParamTypes) == 0 && len(t.BodyRefs) == 0 { + return fmt.Errorf("registry: app %s: tenancy declares no param_types or body_refs — it would enforce nothing", a.ID) + } + types := map[string]bool{} + for _, ty := range t.ParamTypes { + types[ty] = true + } + for _, ty := range t.BodyRefs { + types[ty] = true + } + for _, c := range t.Create { + if c.Type == "" || c.Path == "" || c.Method == "" || c.IDField == "" { + return fmt.Errorf("registry: app %s: tenancy.create needs method, path, type and id_field", a.ID) + } + types[c.Type] = true + } + for _, d := range t.Delete { + if d.Type == "" || d.Path == "" || d.Method == "" || d.Param == "" { + return fmt.Errorf("registry: app %s: tenancy.delete needs method, path, type and param", a.ID) + } + } + for _, l := range t.List { + if l.Path == "" || l.Array == "" || len(l.OwnerBy) == 0 { + return fmt.Errorf("registry: app %s: tenancy.list %q needs array and at least one owner_by link (an unfiltered list leaks every tenant)", a.ID, l.Path) + } + for _, link := range l.OwnerBy { + if link.Field == "" || link.Type == "" { + return fmt.Errorf("registry: app %s: tenancy.list %q has an owner_by with an empty field/type", a.ID, l.Path) + } + if !types[link.Type] { + return fmt.Errorf("registry: app %s: tenancy.list %q links to resource type %q that nothing ever claims — every element would be dropped", a.ID, l.Path, link.Type) + } + } + } + // Every claimable type should be reachable: a type referenced by params/body + // but never created can never be owned, which would deny the app entirely. + created := map[string]bool{} + for _, c := range t.Create { + created[c.Type] = true + } + for _, ty := range t.ParamTypes { + if !created[ty] && !t.lazyClaimed(ty) { + return fmt.Errorf("registry: app %s: tenancy resource %q is referenced but never created or list-claimed — nobody could ever own it", a.ID, ty) + } + } + return nil +} + +// lazyClaimed reports whether a type can be claimed from a list response (the +// inbound-resource path), so it need not have an explicit create route. +func (t *Tenancy) lazyClaimed(rtype string) bool { + for _, l := range t.List { + if l.ClaimAs == rtype { + return true + } + } + return false +} + +// hasDuplicateKeys reports whether any JSON object in body repeats a key. +// +// This exists purely to kill parser-differential bypasses (see EnforceRequest). +// It walks the TOKEN stream, not the decoded value: by the time a value is +// decoded the duplicate is already gone, because Go silently kept the last one. +// +// It fails closed — a body that is malformed or unbalanced counts as duplicated. +// If we cannot prove the body says exactly one thing, we do not forward it. +func hasDuplicateKeys(body []byte) bool { + type frame struct { + keys map[string]bool // nil for arrays + expectVal bool // an object key was read; the next token is its value + } + dec := json.NewDecoder(strings.NewReader(string(body))) + dec.UseNumber() + var stack []*frame + + // valueConsumed marks that the enclosing object's pending key now has its + // value. It must run for EVERY kind of value — including a nested object or + // array — or the key that follows the nested value is mistaken for a value + // and its duplicate goes unnoticed. + valueConsumed := func() { + if n := len(stack); n > 0 && stack[n-1].keys != nil { + stack[n-1].expectVal = false + } + } + + for { + tok, err := dec.Token() + if err != nil { + // Clean end only counts if every container closed. + if errors.Is(err, io.EOF) { + return len(stack) != 0 + } + return true // malformed → ambiguous → refuse + } + switch v := tok.(type) { + case json.Delim: + switch v { + case '{': + valueConsumed() + stack = append(stack, &frame{keys: map[string]bool{}}) + case '[': + valueConsumed() + stack = append(stack, &frame{}) + case '}', ']': + if len(stack) == 0 { + return true // unbalanced + } + stack = stack[:len(stack)-1] + } + case string: + n := len(stack) + if n == 0 { + continue // top-level scalar + } + top := stack[n-1] + if top.keys != nil && !top.expectVal { + // Directly inside an object and not awaiting a value → this is a KEY. + if top.keys[v] { + return true + } + top.keys[v] = true + top.expectVal = true + continue + } + valueConsumed() + default: + valueConsumed() + } + } +} diff --git a/internal/broker/zz_allow_pattern_test.go b/internal/broker/zz_allow_pattern_test.go index 6d075ee..1562953 100644 --- a/internal/broker/zz_allow_pattern_test.go +++ b/internal/broker/zz_allow_pattern_test.go @@ -31,7 +31,7 @@ func TestAllowPatternMatching(t *testing.T) { {"/v1/usage/daily", false}, // longer than the exact entry } for _, c := range cases { - if got := app.allowed(c.path); got != c.want { + if got := app.allowed("GET", c.path); got != c.want { t.Errorf("allowed(%q) = %v, want %v", c.path, got, c.want) } } @@ -44,7 +44,7 @@ func TestAllowEmptyDeniesAll(t *testing.T) { if err != nil { t.Fatalf("ParseRegistry: %v", err) } - if reg.Get("io.pilot.y").allowed("/anything") { + if reg.Get("io.pilot.y").allowed("GET", "/anything") { t.Error("empty allow-list must deny all") } } diff --git a/internal/broker/zz_dupkey_test.go b/internal/broker/zz_dupkey_test.go new file mode 100644 index 0000000..d93a03a --- /dev/null +++ b/internal/broker/zz_dupkey_test.go @@ -0,0 +1,35 @@ +package broker + +import "testing" + +// TestHasDuplicateKeys guards the parser-differential defense. Go's decoder keeps +// the LAST duplicate key; a partner parser keeping the FIRST would act on a value +// the broker never authorized. These cases pin that door shut. +func TestHasDuplicateKeys(t *testing.T) { + cases := []struct { + name string + body string + want bool + }{ + {"clean object", `{"agent_id":"a1","to":"+1555"}`, false}, + {"repeated key, victim first", `{"agent_id":"victim","agent_id":"mine"}`, true}, + {"dup key reversed", `{"agent_id":"mine","agent_id":"victim"}`, true}, + {"dup nested", `{"m":{"agent_id":"victim","agent_id":"mine"}}`, true}, + {"dup in array element", `{"b":[{"agent_id":"v","agent_id":"m"}]}`, true}, + {"same key different objects is fine", `{"a":{"id":"1"},"b":{"id":"2"}}`, false}, + {"repeated key across array elements is fine", `[{"id":"1"},{"id":"2"}]`, false}, + {"value equal to a key name is fine", `{"a":"b","b":"c"}`, false}, + {"string value matching sibling key", `{"agent_id":"agent_id","x":"y"}`, false}, + {"numbers and nulls", `{"a":1,"b":null,"c":true,"d":[1,2]}`, false}, + {"deep nesting clean", `{"a":{"b":{"c":{"d":{"id":"x"}}}}}`, false}, + {"malformed → treated as ambiguous", `{"a":`, true}, + {"empty object", `{}`, false}, + {"array of scalars", `["a","a"]`, false}, + {"dup after nested object closes", `{"n":{"x":1},"k":1,"k":2}`, true}, + } + for _, c := range cases { + if got := hasDuplicateKeys([]byte(c.body)); got != c.want { + t.Errorf("%s: hasDuplicateKeys(%s) = %v, want %v", c.name, c.body, got, c.want) + } + } +} diff --git a/internal/broker/zz_tenancy_test.go b/internal/broker/zz_tenancy_test.go new file mode 100644 index 0000000..8a20b16 --- /dev/null +++ b/internal/broker/zz_tenancy_test.go @@ -0,0 +1,414 @@ +package broker + +import ( + "crypto/ed25519" + "encoding/base64" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// These tests are written as ATTACKS, not as feature checks: each asserts that a +// hostile caller CANNOT reach another tenant's resources. Isolation bugs do not +// show up on the happy path — a caller acting on someone else's resource looks +// exactly like a caller acting on their own unless something checks. So every +// case here is "Mallory tries X against Alice's resource and must fail", plus a +// paired assertion that Alice's own use still works. + +// tenancyRegistry mirrors the shape of the real io.pilot.agentphone entry: a +// shared partner account, ownership tracked broker-side. +const tenancyRegistryJSON = `[{ + "id": "io.pilot.phone", + "upstream": "%s", + "key_env": "PHONE_KEY", + "auth_header": "Authorization", + "auth_scheme": "Bearer", + "quota": 0, + "allow": [ + "GET /v1/agents", "POST /v1/agents", "GET /v1/agents/{agent_id}", + "DELETE /v1/agents/{agent_id}", + "GET /v1/numbers", "POST /v1/numbers", "DELETE /v1/numbers/{number_id}", + "GET /v1/numbers/{number_id}/messages", + "POST /v1/messages", "GET /v1/calls" + ], + "credit": {"seed_credits": 5000000, "default_cost": 0, + "cost_credits": {"POST /v1/numbers": 3000000, "POST /v1/messages": 20000}}, + "tenancy": { + "param_types": {"agent_id": "agent", "number_id": "number"}, + "body_refs": {"agent_id": "agent", "agentId": "agent", "number_id": "number", "phoneNumberId": "number"}, + "create": [ + {"method": "POST", "path": "/v1/agents", "type": "agent", "id_field": "id"}, + {"method": "POST", "path": "/v1/numbers", "type": "number", "id_field": "id"} + ], + "delete": [ + {"method": "DELETE", "path": "/v1/numbers/{number_id}", "type": "number", "param": "number_id"} + ], + "list": [ + {"method": "GET", "path": "/v1/agents", "array": "data", "owner_by": [{"field": "id", "type": "agent"}]}, + {"method": "GET", "path": "/v1/numbers", "array": "data", "owner_by": [{"field": "id", "type": "number"}]}, + {"method": "GET", "path": "/v1/calls", "array": "data", + "owner_by": [{"field": "phoneNumberId", "type": "number"}, {"field": "agentId", "type": "agent"}]} + ] + } +}]` + +// phoneUpstream fakes the partner: it answers for the WHOLE shared account, +// exactly as the real one does — which is why the broker must filter. +func phoneUpstream(t *testing.T, sent *[]string) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case r.Method == "POST" && r.URL.Path == "/v1/agents": + fmt.Fprintf(w, `{"id":"agent_%d","name":"a"}`, time.Now().UnixNano()) + case r.Method == "POST" && r.URL.Path == "/v1/numbers": + fmt.Fprintf(w, `{"id":"num_%d","phoneNumber":"+1555%04d"}`, time.Now().UnixNano(), time.Now().UnixNano()%10000) + case r.Method == "POST" && r.URL.Path == "/v1/messages": + b := make([]byte, r.ContentLength) + _, _ = r.Body.Read(b) + if sent != nil { + *sent = append(*sent, string(b)) + } + fmt.Fprint(w, `{"id":"msg_1","status":"sent"}`) + case r.URL.Path == "/v1/agents": + // the partner sees one shared account: EVERY tenant's agents + fmt.Fprint(w, `{"data":[{"id":"agent_alice"},{"id":"agent_mallory"},{"id":"agent_legacy"}]}`) + case r.URL.Path == "/v1/numbers": + fmt.Fprint(w, `{"data":[{"id":"num_alice","phoneNumber":"+15550001"},{"id":"num_legacy","phoneNumber":"+15550002"}]}`) + case r.URL.Path == "/v1/calls": + fmt.Fprint(w, `{"data":[{"id":"call_1","phoneNumberId":"num_alice"},{"id":"call_2","phoneNumberId":"num_legacy"}]}`) + default: + fmt.Fprint(w, `{"ok":true}`) + } + })) +} + +func tenancyBroker(t *testing.T, sent *[]string) (*Broker, func()) { + t.Helper() + up := phoneUpstream(t, sent) + reg, err := ParseRegistry([]byte(fmt.Sprintf(tenancyRegistryJSON, up.URL)), + func(string) string { return "master-key" }) + if err != nil { + t.Fatalf("ParseRegistry: %v", err) + } + b := New(reg, NewMemStore()) + b.Verify = VerifyConfig{Window: time.Hour} + return b, up.Close +} + +func do(t *testing.T, b *Broker, priv ed25519.PrivateKey, method, path string, body []byte) *httptest.ResponseRecorder { + t.Helper() + rec := httptest.NewRecorder() + b.ServeHTTP(rec, signedReq(t, priv, method, path, body, time.Now())) + return rec +} + +// TestTenancy_CannotSendFromAnotherTenantsAgent: Mallory names Alice's agent in +// the BODY of a send. The path is entirely her own, so only a body-level +// ownership check can stop this. It must never reach the partner. +func TestTenancy_CannotSendFromAnotherTenantsAgent(t *testing.T) { + var sent []string + b, done := tenancyBroker(t, &sent) + defer done() + + _, alice := newKey(t) + _, mallory := newKey(t) + + // Alice creates an agent → she owns it. + rec := do(t, b, alice, "POST", "/io.pilot.phone/v1/agents", []byte(`{"name":"alice"}`)) + if rec.Code != 200 { + t.Fatalf("alice create agent: %d %s", rec.Code, rec.Body) + } + var created struct{ ID string } + if err := json.Unmarshal(rec.Body.Bytes(), &created); err != nil { + t.Fatal(err) + } + + // Mallory names Alice's agent in her own request body. + body := []byte(fmt.Sprintf(`{"agent_id":%q,"to_number":"+15551234","body":"spam"}`, created.ID)) + rec = do(t, b, mallory, "POST", "/io.pilot.phone/v1/messages", body) + if rec.Code != http.StatusNotFound { + t.Errorf("cross-tenant send: status %d, want 404 — mallory acted through alice's agent", rec.Code) + } + if len(sent) != 0 { + t.Errorf("cross-tenant send REACHED THE PARTNER: %v", sent) + } + + // Alice sending from her own agent still works — isolation must not break the product. + rec = do(t, b, alice, "POST", "/io.pilot.phone/v1/messages", body) + if rec.Code != 200 { + t.Errorf("alice's own send: %d %s, want 200", rec.Code, rec.Body) + } + if len(sent) != 1 { + t.Errorf("alice's send should have reached the partner exactly once, got %d", len(sent)) + } +} + +// TestTenancy_LegacyResourcesUnreachable: resources that predate the ledger are +// owned by nobody. Deny-by-default must make them unreachable to EVERYONE rather +// than unclaimed and therefore available to the first caller who names one. +func TestTenancy_LegacyResourcesUnreachable(t *testing.T) { + var sent []string + b, done := tenancyBroker(t, &sent) + defer done() + _, mallory := newKey(t) + + body := []byte(`{"agent_id":"agent_legacy","to_number":"+15551234","body":"spam"}`) + if rec := do(t, b, mallory, "POST", "/io.pilot.phone/v1/messages", body); rec.Code != 404 { + t.Errorf("legacy agent send: %d, want 404", rec.Code) + } + if rec := do(t, b, mallory, "GET", "/io.pilot.phone/v1/numbers/num_legacy/messages", nil); rec.Code != 404 { + t.Errorf("legacy number read: %d, want 404", rec.Code) + } + // And it must not be claimable by simply asserting it. + if rec := do(t, b, mallory, "DELETE", "/io.pilot.phone/v1/numbers/num_legacy", nil); rec.Code != 404 { + t.Errorf("legacy number delete: %d, want 404", rec.Code) + } + if len(sent) != 0 { + t.Errorf("legacy resource reached partner: %v", sent) + } +} + +// TestTenancy_ListsAreFilteredToOwner: a pilot user must not see any number but +// their own. The partner answers for the whole shared account, so the broker has +// to strip the answer down to the caller's rows before returning it. +func TestTenancy_ListsAreFilteredToOwner(t *testing.T) { + b, done := tenancyBroker(t, nil) + defer done() + _, alice := newKey(t) + _, mallory := newKey(t) + + // Nobody owns anything yet → every list is empty, NOT the account-wide answer. + rec := do(t, b, mallory, "GET", "/io.pilot.phone/v1/numbers", nil) + if n := countData(t, rec.Body.Bytes()); n != 0 { + t.Errorf("unowned caller saw %d numbers, want 0 — body: %s", n, rec.Body) + } + if strings.Contains(rec.Body.String(), "+15550002") { + t.Errorf("LEAK: another tenant's number was disclosed: %s", rec.Body) + } + + // Alice owns num_alice → she sees exactly one, and never num_legacy. + if err := b.ownerStore().Claim("io.pilot.phone", "number", "num_alice", callerOf(t, alice), time.Now()); err != nil { + t.Fatal(err) + } + rec = do(t, b, alice, "GET", "/io.pilot.phone/v1/numbers", nil) + if n := countData(t, rec.Body.Bytes()); n != 1 { + t.Errorf("alice saw %d numbers, want 1 — body: %s", n, rec.Body) + } + if strings.Contains(rec.Body.String(), "num_legacy") { + t.Errorf("LEAK: alice saw another tenant's number: %s", rec.Body) + } + // Mallory still sees nothing. + rec = do(t, b, mallory, "GET", "/io.pilot.phone/v1/numbers", nil) + if n := countData(t, rec.Body.Bytes()); n != 0 { + t.Errorf("mallory saw %d numbers, want 0", n) + } +} + +// TestTenancy_InboundResourcesAttributedByLink: an inbound call is created by +// the partner, so the tenant never "created" it. It must still be visible to the +// number's owner (via owner_by link) and invisible to everyone else. +func TestTenancy_InboundCallsFollowNumberOwnership(t *testing.T) { + b, done := tenancyBroker(t, nil) + defer done() + _, alice := newKey(t) + _, mallory := newKey(t) + if err := b.ownerStore().Claim("io.pilot.phone", "number", "num_alice", callerOf(t, alice), time.Now()); err != nil { + t.Fatal(err) + } + + rec := do(t, b, alice, "GET", "/io.pilot.phone/v1/calls", nil) + if n := countData(t, rec.Body.Bytes()); n != 1 { + t.Errorf("alice saw %d calls, want 1 (her inbound call) — %s", n, rec.Body) + } + if strings.Contains(rec.Body.String(), "call_2") { + t.Errorf("LEAK: alice saw a call on another tenant's number: %s", rec.Body) + } + rec = do(t, b, mallory, "GET", "/io.pilot.phone/v1/calls", nil) + if n := countData(t, rec.Body.Bytes()); n != 0 { + t.Errorf("mallory saw %d calls, want 0", n) + } +} + +// TestTenancy_NestedBodyRefIsChecked: wrapping the stolen id one level deep must +// not bypass the check. A top-level-only scan would be trivially defeated. +func TestTenancy_NestedBodyRefIsChecked(t *testing.T) { + var sent []string + b, done := tenancyBroker(t, &sent) + defer done() + _, alice := newKey(t) + _, mallory := newKey(t) + + rec := do(t, b, alice, "POST", "/io.pilot.phone/v1/agents", []byte(`{"name":"alice"}`)) + var created struct{ ID string } + _ = json.Unmarshal(rec.Body.Bytes(), &created) + + for _, body := range []string{ + fmt.Sprintf(`{"message":{"agent_id":%q},"to_number":"+1555"}`, created.ID), // nested object + fmt.Sprintf(`{"batch":[{"agent_id":%q}],"to_number":"+1555"}`, created.ID), // inside an array + fmt.Sprintf(`{"a":{"b":{"c":{"agentId":%q}}},"to_number":"+1555"}`, created.ID), // deep + camelCase alias + } { + rec := do(t, b, mallory, "POST", "/io.pilot.phone/v1/messages", []byte(body)) + if rec.Code != 404 { + t.Errorf("nested ref %s: status %d, want 404", body, rec.Code) + } + } + if len(sent) != 0 { + t.Errorf("a nested cross-tenant ref reached the partner: %v", sent) + } +} + +// TestTenancy_OwnershipIsNotTransferable: a second caller claiming a resource +// that already has an owner must lose. Otherwise "claim" is a takeover. +func TestTenancy_OwnershipIsNotTransferable(t *testing.T) { + s := NewMemStore() + now := time.Now() + if err := s.Claim("app", "number", "num_1", "alice", now); err != nil { + t.Fatalf("alice claim: %v", err) + } + if err := s.Claim("app", "number", "num_1", "mallory", now); err != ErrOwned { + t.Errorf("mallory claim: %v, want ErrOwned", err) + } + if !Owns(s, "app", "number", "num_1", "alice") { + t.Error("alice must still own num_1") + } + if Owns(s, "app", "number", "num_1", "mallory") { + t.Error("mallory must NOT own num_1") + } + // Re-claim by the true owner stays idempotent. + if err := s.Claim("app", "number", "num_1", "alice", now); err != nil { + t.Errorf("idempotent re-claim: %v", err) + } +} + +// TestTenancy_TypeConfusion: an id owned as one type must not authorize the same +// id used as another type. +func TestTenancy_TypeConfusion(t *testing.T) { + s := NewMemStore() + _ = s.Claim("app", "agent", "x_1", "alice", time.Now()) + if Owns(s, "app", "number", "x_1", "alice") { + t.Error("owning agent x_1 must not confer ownership of number x_1") + } +} + +// TestTenancy_FailsClosedWithoutLedger: tenancy declared but the store cannot +// record ownership → deny, never serve unisolated. +func TestTenancy_FailsClosedWithoutLedger(t *testing.T) { + tn := &Tenancy{ParamTypes: map[string]string{"agent_id": "agent"}} + tn.compile() + if _, ok := tn.EnforceRequest(nil, "app", nil, "GET", "/v1/agents/a1", "", nil, "alice"); ok { + t.Error("nil ledger must deny") + } + if _, ok := tn.EnforceRequest(NewMemStore(), "app", nil, "GET", "/v1/agents/a1", "", nil, ""); ok { + t.Error("empty caller must deny") + } +} + +// TestTenancy_MalformedBodyDenied: a body that cannot be parsed cannot be +// checked, so it must not be forwarded. +func TestTenancy_MalformedBodyDenied(t *testing.T) { + var sent []string + b, done := tenancyBroker(t, &sent) + defer done() + _, mallory := newKey(t) + rec := do(t, b, mallory, "POST", "/io.pilot.phone/v1/messages", []byte(`{"agent_id": broken`)) + if rec.Code != 404 { + t.Errorf("malformed body: %d, want 404", rec.Code) + } + if len(sent) != 0 { + t.Errorf("malformed body reached partner: %v", sent) + } +} + +// countData counts elements of the "data" array in a JSON body. +func countData(t *testing.T, b []byte) int { + t.Helper() + var v struct { + Data []any `json:"data"` + } + if err := json.Unmarshal(b, &v); err != nil { + t.Fatalf("unmarshal %s: %v", b, err) + } + return len(v.Data) +} + +// callerOf renders a private key's identity the way the broker records it. +func callerOf(t *testing.T, priv ed25519.PrivateKey) string { + t.Helper() + return base64.RawStdEncoding.EncodeToString(priv.Public().(ed25519.PublicKey)) +} + +// TestTenancy_DuplicateKeyBypassDenied: the broker validates the body with one +// parser and forwards raw bytes for the partner to re-parse with another. When a +// key repeats, the two parsers can disagree about its value, and the broker would +// authorize a value the partner never acts on. Such a body must be refused. +func TestTenancy_DuplicateKeyBypassDenied(t *testing.T) { + var sent []string + b, done := tenancyBroker(t, &sent) + defer done() + _, alice := newKey(t) + _, mallory := newKey(t) + + rec := do(t, b, alice, "POST", "/io.pilot.phone/v1/agents", []byte(`{"name":"alice"}`)) + var created struct{ ID string } + _ = json.Unmarshal(rec.Body.Bytes(), &created) + + // Mallory owns an agent of her own, and names Alice's FIRST. + rec = do(t, b, mallory, "POST", "/io.pilot.phone/v1/agents", []byte(`{"name":"mallory"}`)) + var mine struct{ ID string } + _ = json.Unmarshal(rec.Body.Bytes(), &mine) + + body := []byte(fmt.Sprintf(`{"agent_id":%q,"agent_id":%q,"to_number":"+1555","body":"spam"}`, created.ID, mine.ID)) + rec = do(t, b, mallory, "POST", "/io.pilot.phone/v1/messages", body) + if rec.Code != 404 { + t.Errorf("duplicate-key bypass: status %d, want 404", rec.Code) + } + if len(sent) != 0 { + t.Errorf("duplicate-key body REACHED THE PARTNER: %v", sent) + } +} + +// TestTenancy_CountsDoNotLeakUnfilteredSet: filtering the array is not enough. +// A sibling count left at the partner's value still reports the size of the +// whole shared account, so a caller who can see none of the rows can still read +// how many exist — and watch that number move as other tenants work. +func TestTenancy_CountsDoNotLeakUnfilteredSet(t *testing.T) { + up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"data":[{"id":"num_alice"},{"id":"num_other"},{"id":"num_third"}],"total":3,"hasMore":true}`) + })) + defer up.Close() + reg, err := ParseRegistry([]byte(`[{ + "id":"io.pilot.phone","upstream":"`+up.URL+`","key_env":"K","auth_header":"Authorization","auth_scheme":"Bearer", + "allow":["GET /v1/numbers"], + "tenancy":{"param_types":{"number_id":"number"}, + "create":[{"method":"POST","path":"/v1/numbers","type":"number","id_field":"id"}], + "list":[{"method":"GET","path":"/v1/numbers","array":"data", + "owner_by":[{"field":"id","type":"number"}],"claim_as":"number", + "count_fields":["total"]}]}}]`), func(string) string { return "k" }) + if err != nil { + t.Fatalf("ParseRegistry: %v", err) + } + b := New(reg, NewMemStore()) + b.Verify = VerifyConfig{Window: time.Hour} + _, mallory := newKey(t) + + rec := do(t, b, mallory, "GET", "/io.pilot.phone/v1/numbers", nil) + var got struct { + Data []any `json:"data"` + Total json.Number `json:"total"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("unmarshal %s: %v", rec.Body, err) + } + if len(got.Data) != 0 { + t.Errorf("data = %d rows, want 0", len(got.Data)) + } + if got.Total.String() != "0" { + t.Errorf("total = %s, want 0 — the account-wide count leaked", got.Total) + } +} From 767f27d1119e8424a076e6b2d654d092360cc2ba Mon Sep 17 00:00:00 2001 From: Alexgodoroja Date: Thu, 16 Jul 2026 11:04:35 -0700 Subject: [PATCH 2/5] Operationalize broker tenancy: per-IP grant cap, nginx real-IP, boot guards MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Turns the per-IP identity cap into a GRANT cap: the (N+1)th distinct identity on a source IP is recorded with a zero budget instead of being refused. It can still make free (read) calls; every priced call hits 402. This bounds how much free budget one network can mint without hard-locking a legitimate user behind a shared NAT out of the app — so the cap can be set aggressively low. - deploy/setup-broker-tls.sh: the generated nginx `location /` now sets X-Real-IP. It is the only source-IP signal the broker trusts, and the per-IP cap collapses to a single 127.0.0.1 bucket without it. This was the origin of the real-IP regression; regenerating nginx on deploy no longer reintroduces it. - registry: a classic managed app that forwards with header auth but no header name is now rejected at load instead of 502-ing on first traffic. - deploy/agentphone-tenancy.py: re-enables the spend routes (messages/calls/ numbers) now that tenancy binds each to an owned resource and the grant cap bounds exposure; keeps webhooks/reactions withheld (account-level / unownable). Tests updated to assert the grant-cap contract: a capped identity reads (200) but cannot spend (402), while a funded identity spends normally. --- deploy/agentphone-tenancy.py | 26 ++++++---- deploy/setup-broker-tls.sh | 5 ++ internal/broker/broker.go | 57 ++++++++++++++-------- internal/broker/registry.go | 10 ++++ internal/broker/zz_credit_respcost_test.go | 38 ++++++++++----- 5 files changed, 95 insertions(+), 41 deletions(-) diff --git a/deploy/agentphone-tenancy.py b/deploy/agentphone-tenancy.py index ee89f5b..732642a 100644 --- a/deploy/agentphone-tenancy.py +++ b/deploy/agentphone-tenancy.py @@ -36,10 +36,12 @@ "/v1/messages/{message_id}/reactions", } -# Spend routes remain withheld from the allow-list while the grant for a -# self-minted identity is still unauthenticated. Re-add them together with the -# access-key rollout, not before. -SPEND = {"/v1/messages", "/v1/calls", "/v1/numbers"} +# Spend routes. These are ENABLED: an AI agent cannot use AgentPhone without +# buying a number and sending, so the app is unusable without them. They are now +# safe to serve because (a) tenancy binds every send to a number/agent the caller +# owns, and (b) the per-IP grant cap bounds how much free budget one network can +# mint. Each still debits the caller's own $5 budget at its listed cost. +SPEND = ["/v1/messages", "/v1/calls", "/v1/numbers"] TENANCY = { # Every {param} that appears in an allow pattern MUST be mapped here. A param @@ -137,13 +139,19 @@ def main(): if a["id"] != "io.pilot.agentphone": continue before = len(a["allow"]) - a["allow"] = [p for p in a["allow"] if p not in DROP and p not in SPEND] + # Drop unsafe routes; then ensure spend routes are present (containment + # removed them, this re-enables them). Order-preserving, no duplicates. + allow = [p for p in a["allow"] if p not in DROP and p not in SPEND] + allow += [p for p in SPEND if p not in allow] + a["allow"] = allow a["tenancy"] = TENANCY - # Cap identities per source IP. This is a speed bump against bulk identity - # creation, not a boundary: it does not constrain a caller with many source - # addresses, so it must never be the only control on a grant. + # Per-IP GRANT cap: at most this many funded identities per source IP. + # It bounds free budget minted from one network; the (N+1)th identity is + # recorded with a zero grant (reads still work, spend 402s), so a shared + # NAT is not hard-locked out. It is a speed bump, not a boundary — a caller + # with many source IPs is not constrained — so it is never the only control. a.setdefault("credit", {})["max_identities_per_ip"] = 3 - print("allow: %d -> %d" % (before, len(a["allow"]))) + print("allow: %d -> %d (spend enabled: %s)" % (before, len(allow), ", ".join(SPEND))) # Fail loudly if any {param} still lacks an ownership mapping: an unmapped # param is an unchecked resource, which is exactly the bug class this fixes. diff --git a/deploy/setup-broker-tls.sh b/deploy/setup-broker-tls.sh index be84028..35344ed 100755 --- a/deploy/setup-broker-tls.sh +++ b/deploy/setup-broker-tls.sh @@ -65,6 +65,11 @@ server { location / { proxy_pass $ORIGIN; proxy_set_header Host \$host; + # X-Real-IP is the ONLY source-IP signal the broker trusts (a + # client-supplied X-Forwarded-For is ignored). The per-IP grant cap + # depends on it: omit it and every caller reads as 127.0.0.1, collapsing + # every identity into one IP bucket. It must be set on EVERY location. + proxy_set_header X-Real-IP \$remote_addr; proxy_set_header X-Forwarded-For \$proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto https; } diff --git a/internal/broker/broker.go b/internal/broker/broker.go index 666142c..6498006 100644 --- a/internal/broker/broker.go +++ b/internal/broker/broker.go @@ -85,17 +85,11 @@ func (b *Broker) serveCreditBalance(w http.ResponseWriter, r *http.Request, app return } ip := clientIP(r.Header.Get, r.RemoteAddr, b.IPTrust) - rec, err := ps.Provision(app.ID, caller, ip, app.creditSeed, app.creditMaxPerIP, app.creditMintCooldown, b.now()) + rec, err := b.seedCaller(ps, app.ID, caller, ip, app) if err != nil { - switch err { - case ErrIPCap: - // Deliberately vague: naming the per-IP cap as the reason tells a caller - // exactly which dimension to vary to evade it. The specific cause is - // logged, not returned. - writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "rate limited — try again later"}) - case ErrCooldown: + if err == ErrCooldown { writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "re-provision cooldown — retry shortly"}) - default: + } else { b.internalError(w, http.StatusInternalServerError, app.ID, "balance", err) } return @@ -111,6 +105,30 @@ func (b *Broker) serveCreditBalance(w http.ResponseWriter, r *http.Request, app }) } +// seedCaller records the caller in the credit ledger, enforcing the per-IP +// GRANT cap without ever hard-blocking access. +// +// On first sight it grants the app's seed budget, unless the caller is the +// (N+1)th distinct identity on its source IP — in which case it is recorded with +// a ZERO grant instead of being refused. The distinction matters: the cap exists +// to bound how much free budget one network can mint, not to decide who may call +// the broker. A zero-grant caller can still make free (read) calls; every priced +// call it attempts hits 402 for lack of budget. A caller already in the ledger +// keeps whatever balance it has (no re-seed, no second cap check). +// +// This is what lets the cap be set aggressively low: a tight economic bound +// costs a shared-NAT newcomer only its free grant, not its access. +func (b *Broker) seedCaller(ps ProvisionStore, app, caller, ip string, a *AppEntry) (ProvisionRecord, error) { + rec, err := ps.Provision(app, caller, ip, a.creditSeed, a.creditMaxPerIP, a.creditMintCooldown, b.now()) + if err == ErrIPCap { + // Over the per-IP grant cap: record the identity with no budget and no cap + // (seed 0, maxPerIP 0). Idempotent, so a subsequent call takes the normal + // repeat path and stays at zero rather than ever back-filling a grant. + return ps.Provision(app, caller, ip, 0, 0, 0, b.now()) + } + return rec, err +} + // ServeHTTP is the forward path for //. func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { appID, mpath, ok := strings.Cut(strings.TrimPrefix(r.URL.Path, "/"), "/") @@ -219,19 +237,16 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } ip := clientIP(r.Header.Get, r.RemoteAddr, b.IPTrust) - // Seed the caller on first sight, enforcing the per-IP identity cap: the - // (N+1)th distinct caller seen on one source IP is refused here with 429, so - // a depleted caller cannot immediately re-seed from the same network. - if _, err := ps.Provision(appID, string(caller), ip, app.creditSeed, app.creditMaxPerIP, app.creditMintCooldown, b.now()); err != nil { - switch err { - case ErrIPCap: - // Deliberately vague: naming the per-IP cap as the reason tells a caller - // exactly which dimension to vary to evade it. The specific cause is - // logged, not returned. - writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "rate limited — try again later"}) - case ErrCooldown: + // Seed the caller on first sight, enforcing the per-IP GRANT cap. The cap + // bounds free money, not access: the (N+1)th distinct identity on one source + // IP is recorded with a ZERO grant rather than refused outright. It can still + // make free (read) calls, but every priced call hits 402 — so nobody can farm + // a fresh budget by minting identities from one network, and a legitimate user + // behind a shared NAT is not hard-locked out of the app. + if _, err := b.seedCaller(ps, appID, string(caller), ip, app); err != nil { + if err == ErrCooldown { writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "re-provision cooldown — retry shortly"}) - default: + } else { b.internalError(w, http.StatusInternalServerError, appID, "provision", err) } return diff --git a/internal/broker/registry.go b/internal/broker/registry.go index 78f0c7e..205c430 100644 --- a/internal/broker/registry.go +++ b/internal/broker/registry.go @@ -352,6 +352,16 @@ func ParseRegistry(raw []byte, getenv func(string) string) (*Registry, error) { return nil, fmt.Errorf("registry: app %s: env %s (master key) is empty", a.ID, a.KeyEnv) } a.injector = injectorFor(a.AuthStyle, a.AuthHeader, a.AuthScheme, a.AuthParam, a.AuthUser) + // A classic managed app that forwards with header-style auth but no header + // name builds an injector that sets an EMPTY header, which fails only when + // the first call is dialed (a 502). Catch it at load so a config typo is a + // boot failure, not a silent runtime outage on live traffic. Skipped for + // provisioned apps (they mint per-user keys, not a master-key header) and + // for empty-allow apps (which forward nothing). + if a.Provision == nil && len(a.Allow) > 0 && + (a.AuthStyle == "" || a.AuthStyle == "header") && strings.TrimSpace(a.AuthHeader) == "" { + return nil, fmt.Errorf("registry: app %s: header auth needs a non-empty auth_header (set auth_style to query/basic if intended)", a.ID) + } a.allowSet = map[string]bool{} a.allowPatterns = nil a.allowSegs = nil diff --git a/internal/broker/zz_credit_respcost_test.go b/internal/broker/zz_credit_respcost_test.go index 118f2a5..558a8ca 100644 --- a/internal/broker/zz_credit_respcost_test.go +++ b/internal/broker/zz_credit_respcost_test.go @@ -1,6 +1,7 @@ package broker import ( + "crypto/ed25519" "encoding/json" "net/http" "net/http/httptest" @@ -220,6 +221,10 @@ func TestCreditBalance_PerUserNotAccount(t *testing.T) { // pilot identity from the same source IP cannot claim a fresh budget — this is the // anti-Sybil guard that stops farming new $5 grants after depletion. Both callers // share httptest's default RemoteAddr (192.0.2.1). +// TestCreditPath_PerIPIdentityCap pins the per-IP GRANT cap contract: one source +// IP may mint at most N funded identities. The (N+1)th is not refused — it is +// recorded with a ZERO grant, so it can still read but cannot spend. This is the +// control that stops budget-farming while keeping a shared-NAT newcomer usable. func TestCreditPath_PerIPIdentityCap(t *testing.T) { now := time.Unix(1_800_000_000, 0) up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { @@ -227,10 +232,12 @@ func TestCreditPath_PerIPIdentityCap(t *testing.T) { _, _ = w.Write([]byte(`{"ok":true}`)) })) t.Cleanup(up.Close) + // "/v1/read" is free; "/v1/act" costs $1. Cap = 1 funded identity per IP. reg, err := ParseRegistry([]byte(`[{ "id":"io.pilot.orthogonal","upstream":"`+up.URL+`","key_env":"ORTH_KEY", - "auth_header":"Authorization","auth_scheme":"Bearer","allow":["/v1/search"], - "credit":{"seed_credits":5000000,"default_cost":0,"max_identities_per_ip":1} + "auth_header":"Authorization","auth_scheme":"Bearer","allow":["/v1/read","/v1/act"], + "credit":{"seed_credits":5000000,"default_cost":0, + "cost_credits":{"POST /v1/act":1000000},"max_identities_per_ip":1} }]`), func(string) string { return "MASTERKEY" }) if err != nil { t.Fatalf("ParseRegistry: %v", err) @@ -238,17 +245,26 @@ func TestCreditPath_PerIPIdentityCap(t *testing.T) { b := New(reg, NewMemStore()) b.Verify = VerifyConfig{Now: fixedClock(now)} + call := func(priv ed25519.PrivateKey, path string) int { + rec := httptest.NewRecorder() + b.ServeHTTP(rec, signedReq(t, priv, "POST", "/io.pilot.orthogonal"+path, []byte(`{}`), now)) + return rec.Code + } + + // Identity 1 is funded: it can spend. _, priv1 := newKey(t) - rec1 := httptest.NewRecorder() - b.ServeHTTP(rec1, signedReq(t, priv1, "POST", "/io.pilot.orthogonal/v1/search", []byte(`{}`), now)) - if rec1.Code != 200 { - t.Fatalf("first identity: status %d, want 200", rec1.Code) + if got := call(priv1, "/v1/act"); got != 200 { + t.Fatalf("funded identity spend: status %d, want 200", got) } - // A different key = a different pilot identity from the same IP → capped. + + // Identity 2 on the SAME IP is over the cap → zero grant. _, priv2 := newKey(t) - rec2 := httptest.NewRecorder() - b.ServeHTTP(rec2, signedReq(t, priv2, "POST", "/io.pilot.orthogonal/v1/search", []byte(`{}`), now)) - if rec2.Code != http.StatusTooManyRequests { - t.Fatalf("second identity same IP: status %d, want 429 (IP cap)", rec2.Code) + // It is NOT locked out: a free read still works. + if got := call(priv2, "/v1/read"); got != 200 { + t.Errorf("capped identity read: status %d, want 200 (must not be hard-blocked)", got) + } + // But it has no budget, so a priced call is refused with 402 — no free money. + if got := call(priv2, "/v1/act"); got != http.StatusPaymentRequired { + t.Errorf("capped identity spend: status %d, want 402 (no grant)", got) } } From ef43b1cbde49bbf3b965fec40b9c78d918c130ff Mon Sep 17 00:00:00 2001 From: Alexgodoroja Date: Thu, 16 Jul 2026 11:41:11 -0700 Subject: [PATCH 3/5] Restore broker ACCESS log middleware (structured per-request events) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The broker emits one `ACCESS {json}` line per forwarded request to stdout → journald, which the monitoring crawler ingests. This middleware was lost when the broker was last rebuilt from a branch without it, so the pipeline went dark. Restoring it here keeps it alongside the tenancy/security changes, so a redeploy from this branch does not clobber one for the other. - internal/broker/accesslog.go: WithAccessLog middleware, AccessEvent (stable JSON contract), StdoutSink, and the X-Pilot-Method (pilotctl command capture) and X-Pilot-Unrouted (skip unknown-app / scanner noise) header constants. - internal/broker/broker.go: mark the two unrouted 404 paths with X-Pilot-Unrouted so scanner traffic never counts as broker usage. - cmd/broker/main.go: wire the middleware with a -broker-name label (defaults to BROKER_NAME env or hostname); /gw/ health/usage paths pass through unlogged. Reduced to the stdout→journald path (what the pilot-publish crawler reads); the remote-push HTTPSink for the smol broker is intentionally not included here. --- cmd/broker/main.go | 17 +++- internal/broker/accesslog.go | 130 ++++++++++++++++++++++++++++++ internal/broker/accesslog_test.go | 104 ++++++++++++++++++++++++ internal/broker/broker.go | 2 + 4 files changed, 251 insertions(+), 2 deletions(-) create mode 100644 internal/broker/accesslog.go create mode 100644 internal/broker/accesslog_test.go diff --git a/cmd/broker/main.go b/cmd/broker/main.go index 486c450..0f7bd61 100644 --- a/cmd/broker/main.go +++ b/cmd/broker/main.go @@ -31,6 +31,7 @@ func main() { window := flag.Duration("window", 5*time.Minute, "signed-request freshness window") ipHeader := flag.String("ip-header", envOr("BROKER_IP_HEADER", "X-Real-IP"), "header carrying the real source IP (set by the front proxy; client X-Forwarded-For is never trusted)") meterInterval := flag.Duration("meter-interval", 60*time.Second, "usage-metering tick for provisioned apps") + brokerName := flag.String("broker-name", envOr("BROKER_NAME", defaultHostname()), "process label on every access event") flag.Parse() reg, err := broker.LoadRegistry(*registryPath, os.Getenv) @@ -105,15 +106,27 @@ func main() { }) mux.Handle("/", b) - log.Printf("broker: listening on %s", *addr) + // Access logging: one structured `ACCESS {json}` line per forwarded request + // to stdout → journald (read by the monitoring log crawler). /gw/ and + // unrouted (scanner-noise) requests are skipped. + handler := broker.WithAccessLog(mux, *brokerName, broker.StdoutSink{W: os.Stdout}, nil) + + log.Printf("broker: listening on %s (name=%s)", *addr, *brokerName) srv := &http.Server{ Addr: *addr, - Handler: mux, + Handler: handler, ReadHeaderTimeout: 10 * time.Second, } log.Fatal(srv.ListenAndServe()) } +func defaultHostname() string { + if h, err := os.Hostname(); err == nil && h != "" { + return h + } + return "broker" +} + func envOr(k, def string) string { if v := os.Getenv(k); v != "" { return v diff --git a/internal/broker/accesslog.go b/internal/broker/accesslog.go new file mode 100644 index 0000000..b473d1c --- /dev/null +++ b/internal/broker/accesslog.go @@ -0,0 +1,130 @@ +package broker + +import ( + "encoding/json" + "io" + "net/http" + "strings" + "time" +) + +// AccessEvent is one broker request, emitted per request for monitoring. The +// JSON tags are the wire/DB contract — keep them stable. +type AccessEvent struct { + TS time.Time `json:"ts"` + Broker string `json:"broker"` + AppID string `json:"app_id"` + HTTPMethod string `json:"http_method"` + Path string `json:"path"` + Status int `json:"status"` + DurationMs int64 `json:"duration_ms"` + ReqBytes int64 `json:"req_bytes"` + RespBytes int64 `json:"resp_bytes"` + Method string `json:"method,omitempty"` // pilotctl command, from X-Pilot-Method + Source string `json:"source,omitempty"` +} + +// AccessSink receives one event per request; MUST NOT block the request path. +type AccessSink interface{ Emit(AccessEvent) } + +// AccessLogPrefix marks stdout access lines so the crawler can select them out +// of the broker's mixed journald stream. +const AccessLogPrefix = "ACCESS " + +// PilotMethodHeader carries the pilotctl command (`.`) an app +// adapter stamps on a brokered call, making stats command-oriented. +const PilotMethodHeader = "X-Pilot-Method" + +// UnroutedHeader is set by the broker on requests it could not route to a +// managed app (unknown app / bad route); the access log skips these so +// internet-scanner noise never counts as broker traffic. +const UnroutedHeader = "X-Pilot-Unrouted" + +// StdoutSink writes one prefixed JSON line per event to W (os.Stdout in prod). +type StdoutSink struct{ W io.Writer } + +func (s StdoutSink) Emit(e AccessEvent) { + b, err := json.Marshal(e) + if err != nil { + return + } + line := make([]byte, 0, len(AccessLogPrefix)+len(b)+1) + line = append(line, AccessLogPrefix...) + line = append(line, b...) + line = append(line, '\n') + _, _ = s.W.Write(line) // single Write so concurrent goroutines don't interleave +} + +// statusRecorder captures the response status code and byte count. +type statusRecorder struct { + http.ResponseWriter + status int + nbytes int64 +} + +func (r *statusRecorder) WriteHeader(code int) { + r.status = code + r.ResponseWriter.WriteHeader(code) +} + +func (r *statusRecorder) Write(b []byte) (int, error) { + if r.status == 0 { + r.status = http.StatusOK + } + n, err := r.ResponseWriter.Write(b) + r.nbytes += int64(n) + return n, err +} + +// appIDFromPath splits // exactly as ServeHTTP does. +func appIDFromPath(p string) (appID, method string) { + appID, rest, ok := strings.Cut(strings.TrimPrefix(p, "/"), "/") + if !ok { + return appID, "/" + } + return appID, "/" + rest +} + +// WithAccessLog wraps next, emitting one AccessEvent per forwarded request to +// sink. broker labels the process; now is injectable for tests (nil → time.Now). +// /gw/ paths, a nil sink, and requests the broker flagged X-Pilot-Unrouted pass +// through unlogged. +func WithAccessLog(next http.Handler, broker string, sink AccessSink, now func() time.Time) http.Handler { + if now == nil { + now = time.Now + } + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if sink == nil || strings.HasPrefix(r.URL.Path, "/gw/") { + next.ServeHTTP(w, r) + return + } + start := now() + pilotMethod := r.Header.Get(PilotMethodHeader) + rec := &statusRecorder{ResponseWriter: w} + next.ServeHTTP(rec, r) + + if rec.Header().Get(UnroutedHeader) != "" { // scanner noise / unknown app + return + } + appID, method := appIDFromPath(r.URL.Path) + if rec.status == 0 { + rec.status = http.StatusOK + } + reqBytes := r.ContentLength + if reqBytes < 0 { + reqBytes = 0 + } + sink.Emit(AccessEvent{ + TS: start, + Broker: broker, + AppID: appID, + HTTPMethod: r.Method, + Path: method, + Method: pilotMethod, + Status: rec.status, + DurationMs: now().Sub(start).Milliseconds(), + ReqBytes: reqBytes, + RespBytes: rec.nbytes, + }) + }) +} diff --git a/internal/broker/accesslog_test.go b/internal/broker/accesslog_test.go new file mode 100644 index 0000000..671b0bb --- /dev/null +++ b/internal/broker/accesslog_test.go @@ -0,0 +1,104 @@ +package broker + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +type capSink struct{ events []AccessEvent } + +func (c *capSink) Emit(e AccessEvent) { c.events = append(c.events, e) } + +// fixedNow returns a clock that advances by one fixed step per call, so +// DurationMs is deterministic (start=T0, end=T0+step). +func fixedNow(base time.Time, step time.Duration) func() time.Time { + n := -1 + return func() time.Time { + n++ + if n == 0 { + return base + } + return base.Add(step) + } +} + +// TestAccessLog_EmitsPerForwardedRequest: a routed request produces exactly one +// event carrying the app id, method-path, status, byte counts, the pilotctl +// command from X-Pilot-Method, and a duration. +func TestAccessLog_EmitsPerForwardedRequest(t *testing.T) { + inner := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(200) + _, _ = w.Write([]byte(`{"ok":true}`)) + }) + sink := &capSink{} + h := WithAccessLog(inner, "pilot-broker", sink, fixedNow(time.Unix(1_800_000_000, 0), 7*time.Millisecond)) + + req := httptest.NewRequest("POST", "/io.pilot.agentphone/v1/messages", strings.NewReader(`{"x":1}`)) + req.Header.Set(PilotMethodHeader, "agentphone.send_message") + h.ServeHTTP(httptest.NewRecorder(), req) + + if len(sink.events) != 1 { + t.Fatalf("events = %d, want 1", len(sink.events)) + } + e := sink.events[0] + if e.Broker != "pilot-broker" || e.AppID != "io.pilot.agentphone" || e.Path != "/v1/messages" { + t.Errorf("bad routing fields: %+v", e) + } + if e.HTTPMethod != "POST" || e.Status != 200 { + t.Errorf("bad http fields: %+v", e) + } + if e.Method != "agentphone.send_message" { + t.Errorf("X-Pilot-Method not captured: %q", e.Method) + } + if e.DurationMs != 7 { + t.Errorf("DurationMs = %d, want 7", e.DurationMs) + } + if e.RespBytes == 0 { + t.Error("RespBytes not recorded") + } +} + +// TestAccessLog_SkipsUnroutedAndHealth: /gw/ health and requests the broker +// flagged X-Pilot-Unrouted (unknown app / bad route) emit nothing, so scanner +// noise never counts as broker traffic. +func TestAccessLog_SkipsUnroutedAndHealth(t *testing.T) { + unrouted := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set(UnroutedHeader, "1") + w.WriteHeader(404) + }) + sink := &capSink{} + WithAccessLog(unrouted, "b", sink, nil).ServeHTTP(httptest.NewRecorder(), + httptest.NewRequest("GET", "/HNAP1", nil)) + if len(sink.events) != 0 { + t.Errorf("unrouted request logged: %+v", sink.events) + } + + health := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(200) }) + WithAccessLog(health, "b", sink, nil).ServeHTTP(httptest.NewRecorder(), + httptest.NewRequest("GET", "/gw/health", nil)) + if len(sink.events) != 0 { + t.Errorf("/gw/health logged: %+v", sink.events) + } +} + +// TestStdoutSink_PrefixedJSONLine: the sink writes one AccessLogPrefix-tagged +// JSON line, which the crawler selects out of the mixed journald stream. +func TestStdoutSink_PrefixedJSONLine(t *testing.T) { + var buf strings.Builder + StdoutSink{W: &buf}.Emit(AccessEvent{Broker: "b", AppID: "io.pilot.x", Status: 200}) + out := buf.String() + if !strings.HasPrefix(out, AccessLogPrefix) || !strings.HasSuffix(out, "\n") { + t.Fatalf("bad line framing: %q", out) + } + var e AccessEvent + if err := json.Unmarshal([]byte(strings.TrimPrefix(strings.TrimSpace(out), AccessLogPrefix)), &e); err != nil { + t.Fatalf("payload not JSON: %v", err) + } + if e.AppID != "io.pilot.x" { + t.Errorf("round-trip lost app_id: %+v", e) + } +} diff --git a/internal/broker/broker.go b/internal/broker/broker.go index 6498006..37ed9cf 100644 --- a/internal/broker/broker.go +++ b/internal/broker/broker.go @@ -133,6 +133,7 @@ func (b *Broker) seedCaller(ps ProvisionStore, app, caller, ip string, a *AppEnt func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { appID, mpath, ok := strings.Cut(strings.TrimPrefix(r.URL.Path, "/"), "/") if !ok || appID == "" { + w.Header().Set(UnroutedHeader, "1") // not a managed-app call — monitoring skips it writeJSON(w, http.StatusNotFound, map[string]string{"error": "route must be //"}) return } @@ -142,6 +143,7 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { // the app's artifact limit before reading. app := b.reg.Load().Get(appID) if app == nil { + w.Header().Set(UnroutedHeader, "1") // unknown app (typo / internet scanner) — not real broker usage writeJSON(w, http.StatusNotFound, map[string]string{"error": "unknown app: " + appID}) return } From 54536cace7efe04d4784685b11a4b8c6c7c696b0 Mon Sep 17 00:00:00 2001 From: Alexgodoroja Date: Thu, 16 Jul 2026 11:47:50 -0700 Subject: [PATCH 4/5] Add tests for access keys, SQLite ownership ledger, and object redaction Restores internal/broker coverage above the security-critical gate: the new tenancy, access-key, and ownership code was under-covered. Adds direct tests for access-key parse/verify (incl. fail-closed with no keys and the require-key gate), the SQLite ownership ledger (claim first-writer-wins, OwnerOf/OwnedSet/ Release, Owns) and credit ledger (debit/refund/settle/clamp), and account-summary redaction (owned-count recompute + field redaction). --- internal/broker/zz_coverage_test.go | 240 ++++++++++++++++++++++++++++ 1 file changed, 240 insertions(+) create mode 100644 internal/broker/zz_coverage_test.go diff --git a/internal/broker/zz_coverage_test.go b/internal/broker/zz_coverage_test.go new file mode 100644 index 0000000..33c476e --- /dev/null +++ b/internal/broker/zz_coverage_test.go @@ -0,0 +1,240 @@ +package broker + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func okUpstream() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(200) + _, _ = w.Write([]byte(`{"ok":true}`)) + }) +} + +// --- access keys --------------------------------------------------------- + +func TestAccessKeys_ParseAndCheck(t *testing.T) { + ak := NewAccessKeys([]string{"alpha:key-one", " ", "bare-two", "beta: key-three "}) + if ak.Len() != 3 { + t.Fatalf("Len = %d, want 3 (blank entry ignored)", ak.Len()) + } + // header form + if lbl, ok := ak.Check(hdr(map[string]string{AccessKeyHeader: "key-one"})); !ok || lbl != "alpha" { + t.Errorf("valid header key: ok=%v label=%q", ok, lbl) + } + // bearer form + if lbl, ok := ak.Check(hdr(map[string]string{"Authorization": "Bearer bare-two"})); !ok || lbl != "" { + t.Errorf("valid bearer key: ok=%v label=%q", ok, lbl) + } + // label:key with spaces trimmed + if _, ok := ak.Check(hdr(map[string]string{AccessKeyHeader: "key-three"})); !ok { + t.Error("trimmed label:key should verify") + } + // wrong + missing → deny + if _, ok := ak.Check(hdr(map[string]string{AccessKeyHeader: "nope"})); ok { + t.Error("wrong key must not verify") + } + if _, ok := ak.Check(hdr(map[string]string{})); ok { + t.Error("missing key must not verify") + } +} + +func TestAccessKeys_NoKeysFailsClosed(t *testing.T) { + var nilAK *AccessKeys + if _, ok := nilAK.Check(hdr(map[string]string{AccessKeyHeader: "x"})); ok { + t.Error("nil AccessKeys must deny") + } + empty := NewAccessKeys([]string{"", " "}) + if empty.Len() != 0 { + t.Fatalf("empty Len = %d", empty.Len()) + } + if _, ok := empty.Check(hdr(map[string]string{AccessKeyHeader: "x"})); ok { + t.Error("no keys configured must authorize nothing") + } +} + +func hdrFn(m map[string]string) func(string) string { + return func(k string) string { return m[k] } +} + +// TestRequireAccessKey_GatesApp: an app with require_access_key returns 401 +// without a valid key and forwards with one. Also covers AppsRequiringAccessKey. +func TestRequireAccessKey_GatesApp(t *testing.T) { + up := httptest.NewServer(okUpstream()) + t.Cleanup(up.Close) + reg, err := ParseRegistry([]byte(`[{ + "id":"io.pilot.gated","upstream":"`+up.URL+`","key_env":"G_KEY", + "auth_header":"Authorization","auth_scheme":"Bearer","allow":["/v1/x"], + "require_access_key":true + }]`), func(string) string { return "master" }) + if err != nil { + t.Fatalf("ParseRegistry: %v", err) + } + if got := reg.AppsRequiringAccessKey(); len(got) != 1 || got[0] != "io.pilot.gated" { + t.Fatalf("AppsRequiringAccessKey = %v", got) + } + b := New(reg, NewMemStore()) + b.Verify = VerifyConfig{Window: time.Hour} + b.AccessKeys = NewAccessKeys([]string{"secret-key"}) + _, priv := newKey(t) + + // no key → 401 + rec := httptest.NewRecorder() + b.ServeHTTP(rec, signedReq(t, priv, "POST", "/io.pilot.gated/v1/x", []byte(`{}`), time.Now())) + if rec.Code != 401 { + t.Errorf("no access key: status %d, want 401", rec.Code) + } + // with key → forwarded (200) + rec = httptest.NewRecorder() + req := signedReq(t, priv, "POST", "/io.pilot.gated/v1/x", []byte(`{}`), time.Now()) + req.Header.Set(AccessKeyHeader, "secret-key") + b.ServeHTTP(rec, req) + if rec.Code != 200 { + t.Errorf("with access key: status %d, want 200", rec.Code) + } +} + +// --- SQLite ownership ledger --------------------------------------------- + +func TestSQLiteOwnership_ClaimFirstWriterWins(t *testing.T) { + s, err := OpenSQLiteStore(":memory:") + if err != nil { + t.Fatal(err) + } + defer s.Close() + now := time.Unix(1_800_000_000, 0) + + if err := s.Claim("app", "number", "n1", "alice", now); err != nil { + t.Fatalf("alice claim: %v", err) + } + if err := s.Claim("app", "number", "n1", "alice", now); err != nil { + t.Errorf("idempotent re-claim: %v", err) + } + if err := s.Claim("app", "number", "n1", "mallory", now); err != ErrOwned { + t.Errorf("takeover: %v, want ErrOwned", err) + } + owner, found, err := s.OwnerOf("app", "number", "n1") + if err != nil || !found || owner != "alice" { + t.Errorf("OwnerOf = %q,%v,%v", owner, found, err) + } + if _, found, _ := s.OwnerOf("app", "number", "ghost"); found { + t.Error("unknown resource must not be found") + } + // OwnedSet + Release + _ = s.Claim("app", "number", "n2", "alice", now) + _ = s.Claim("app", "agent", "a1", "alice", now) + set, err := s.OwnedSet("app", "number", "alice") + if err != nil || len(set) != 2 || !set["n1"] || !set["n2"] { + t.Errorf("OwnedSet(number) = %v (err %v)", set, err) + } + if err := s.Release("app", "number", "n1"); err != nil { + t.Fatal(err) + } + if _, found, _ := s.OwnerOf("app", "number", "n1"); found { + t.Error("released resource still owned") + } + // Owns() helper against SQLite + if !Owns(s, "app", "agent", "a1", "alice") || Owns(s, "app", "agent", "a1", "mallory") { + t.Error("Owns disagrees with ledger") + } +} + +// TestSQLiteCredit_DebitRefundSettle exercises the SQLite credit ledger paths. +func TestSQLiteCredit_DebitRefundSettle(t *testing.T) { + s, err := OpenSQLiteStore(":memory:") + if err != nil { + t.Fatal(err) + } + defer s.Close() + now := time.Unix(1_800_000_000, 0) + + if _, err := s.Provision("app", "c", "1.2.3.4", 100, 0, 0, now); err != nil { + t.Fatal(err) + } + if bal, err := s.Credit("app", "c"); err != nil || bal != 100 { + t.Fatalf("seed Credit = %d,%v", bal, err) + } + ok, rem, err := s.Debit("app", "c", 40) + if err != nil || !ok || rem != 60 { + t.Fatalf("Debit 40 = %v,%d,%v", ok, rem, err) + } + s.Refund("app", "c", 10) // back to 70 + if bal, _ := s.Credit("app", "c"); bal != 70 { + t.Fatalf("after refund = %d, want 70", bal) + } + if rem, err := s.Settle("app", "c", 1000); err != nil || rem != 0 { + t.Fatalf("Settle overshoot = %d,%v, want clamp to 0", rem, err) + } + // over-debit refused + if ok, _, _ := s.Debit("app", "c", 5); ok { + t.Error("debit past zero must fail") + } + rec, found, err := s.Get("app", "c") + if err != nil || !found || rec.Credits != 0 { + t.Errorf("Get = %+v,%v,%v", rec, found, err) + } +} + +// --- tenancy account-summary redaction ----------------------------------- + +// TestFilterObject_RecomputesAndRedacts: an account-summary response has its +// count replaced by the caller's own count and its unattributable fields dropped. +func TestFilterObject_RecomputesAndRedacts(t *testing.T) { + s := NewMemStore() + now := time.Unix(1_800_000_000, 0) + _ = s.Claim("app", "number", "n1", "alice", now) + _ = s.Claim("app", "number", "n2", "alice", now) + + tn := &Tenancy{ + ParamTypes: map[string]string{"number_id": "number"}, + Create: []CreateRoute{{Method: "POST", Path: "/v1/numbers", Type: "number", IDField: "id"}}, + Object: []ObjectRoute{{ + Method: "GET", Path: "/v1/usage", + OwnedCounts: map[string]string{"numbers.used": "number"}, + Redact: []string{"stats", "numbers.remaining"}, + }}, + } + if err := validateTenancy(&AppEntry{ID: "app", Tenancy: tn}); err != nil { + t.Fatalf("validateTenancy: %v", err) + } + tn.compile() + + // partner reports the whole account: 9 numbers used, remaining 991, stats blob + raw := []byte(`{"numbers":{"used":9,"remaining":991},"stats":{"total":123}}`) + out, did := tn.FilterObject(s, "app", "GET", "/v1/usage", raw, "alice") + if !did { + t.Fatal("FilterObject did not handle the route") + } + var got map[string]any + if err := json.Unmarshal(out, &got); err != nil { + t.Fatalf("unmarshal %s: %v", out, err) + } + nums := got["numbers"].(map[string]any) + if fmtNum(nums["used"]) != "2" { + t.Errorf("numbers.used = %v, want 2 (alice's own)", nums["used"]) + } + if _, ok := nums["remaining"]; ok { + t.Error("numbers.remaining should be redacted (derived from account-wide used)") + } + if _, ok := got["stats"]; ok { + t.Error("stats should be redacted") + } + // a route it doesn't cover passes through untouched + if _, did := tn.FilterObject(s, "app", "GET", "/v1/other", raw, "alice"); did { + t.Error("uncovered route should not be handled") + } +} + +func fmtNum(v any) string { + switch x := v.(type) { + case json.Number: + return x.String() + case float64: + return json.Number(jsonNum(x)).String() + } + return "" +} From 27fa84ca589af7fe8a5a761d4d090667344f1135 Mon Sep 17 00:00:00 2001 From: Alexgodoroja Date: Thu, 16 Jul 2026 11:51:11 -0700 Subject: [PATCH 5/5] test: reuse existing hdr helper (drop duplicate) --- internal/broker/zz_coverage_test.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/internal/broker/zz_coverage_test.go b/internal/broker/zz_coverage_test.go index 33c476e..8ee3c8a 100644 --- a/internal/broker/zz_coverage_test.go +++ b/internal/broker/zz_coverage_test.go @@ -57,10 +57,6 @@ func TestAccessKeys_NoKeysFailsClosed(t *testing.T) { } } -func hdrFn(m map[string]string) func(string) string { - return func(k string) string { return m[k] } -} - // TestRequireAccessKey_GatesApp: an app with require_access_key returns 401 // without a valid key and forwards with one. Also covers AppsRequiringAccessKey. func TestRequireAccessKey_GatesApp(t *testing.T) {