Broker tenancy + per-IP grant cap + fail-closed guards

cmd/broker/main.go

@@ -18,6 +18,7 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"strings"
 	"syscall"
 	"time"
 
@@ -30,13 +31,25 @@ 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)
 	if err != nil {
 		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 +74,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.
@@ -91,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

deploy/agentphone-tenancy.py

@@ -0,0 +1,172 @@
+#!/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. 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
+    # 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"])
+        # 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
+        # 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  (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.
+        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()

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;
     }

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 <key> 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 <key>
+	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
+}