Skip to content

fix(errors): ingest snapshots under the authenticated caller identity #106

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
woodsonl wants to merge 1 commit into from
Closed

fix(errors): ingest snapshots under the authenticated caller identity #106

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 24 additions & 11 deletions services/nvpair-errors/httpserver.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"crypto/tls"
"encoding/json"
"fmt"
"io"
"log/slog"
"net"
"net/http"
Expand All @@ -32,6 +33,10 @@ import (
"nvpair-shared/errors"
)

// maxIngestBodyBytes bounds an inbound error-sync snapshot, matching the
// 1 MiB cap the other inter-node endpoints apply to request bodies.
const maxIngestBodyBytes = 1 << 20

// newErrorsMux builds the HTTP routes backed by the manager. Split out
// from the listener so tests can exercise the handlers via
// httptest.NewServer without binding a real port or touching mDNS.
Expand All @@ -51,13 +56,14 @@ func newErrorsMux(mgr *Manager, mesh *clustertrust.Mesh) *http.ServeMux {
// unclustered node serve this data in the clear, so there is no branch here
// for a future caller to widen.
mesh.Refresh()
if _, ok := mesh.VerifyClientPin(r); !ok {
callerUUID, ok := mesh.VerifyClientPin(r)
if !ok {
http.Error(w, "forbidden: not a pinned cluster peer", http.StatusForbidden)
return
}
switch r.Method {
case http.MethodPost:
handleIngest(w, r, mgr)
handleIngest(w, r, mgr, callerUUID)
case http.MethodGet:
handleServeLocal(w, mgr)
default:
Expand All @@ -68,15 +74,16 @@ func newErrorsMux(mgr *Manager, mesh *clustertrust.Mesh) *http.ServeMux {
return mux
}

// handleIngest reconciles a peer's pushed SyncEnvelope. A malformed
// body or an envelope missing its nodeId is a 400 — we never let a
// peer reconcile our own origin (the manager rejects nodeId ==
// localNodeID as a no-op, but we reject empty here so the client gets
// a clear signal rather than a silent accept).
func handleIngest(w http.ResponseWriter, r *http.Request, mgr *Manager) {
// handleIngest reconciles a peer's pushed SyncEnvelope. The envelope's nodeId
// is NOT trusted: the origin identity is the mTLS-authenticated caller UUID,
// which the pin store keys identically to the nodeId the push side uses. A
// mismatched body value is a 400 (caught by tests and honest peers), a
// malformed body is a 400, and the manager still rejects reconciling our own
// origin as a no-op.
func handleIngest(w http.ResponseWriter, r *http.Request, mgr *Manager, callerUUID string) {
defer r.Body.Close()
var env errors.SyncEnvelope
if err := json.NewDecoder(r.Body).Decode(&env); err != nil {
if err := json.NewDecoder(io.LimitReader(r.Body, maxIngestBodyBytes)).Decode(&env); err != nil {
slog.Warn("ingest: bad request body", "err", err)
http.Error(w, "invalid JSON body", http.StatusBadRequest)
return
Expand All @@ -85,9 +92,15 @@ func handleIngest(w http.ResponseWriter, r *http.Request, mgr *Manager) {
http.Error(w, `"nodeId" is required`, http.StatusBadRequest)
return
}
if env.NodeID != callerUUID {
slog.Warn("ingest: envelope nodeId does not match the authenticated caller",
"caller", callerUUID)
http.Error(w, "nodeId does not match the authenticated caller", http.StatusBadRequest)
return
}

mgr.ReconcilePeer(env.NodeID, env.Errors)
slog.Debug("ingested peer snapshot", "nodeId", env.NodeID, "count", len(env.Errors))
mgr.ReconcilePeer(callerUUID, env.Errors)
slog.Debug("ingested peer snapshot", "nodeId", callerUUID, "count", len(env.Errors))
w.WriteHeader(http.StatusNoContent)
}

Expand Down
66 changes: 61 additions & 5 deletions services/nvpair-errors/peersync_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"encoding/json"
"net/http"
"slices"
"strings"
"testing"

"nvpair-shared/clustertrust"
Expand Down Expand Up @@ -156,14 +157,16 @@ func TestEvictNodeRemovesPeerEntries(t *testing.T) {
}

// TestHTTPIngestReconciles: the POST /v1/errors handler decodes a
// SyncEnvelope and merges it, returning 204.
// SyncEnvelope and merges it, returning 204. The envelope's nodeId must
// match the mTLS-authenticated caller UUID — the pin store keys peers by
// the same UUID the push side uses as nodeId.
func TestHTTPIngestReconciles(t *testing.T) {
m := managerForNode("node-a")
srv, client := servePinnedErrorsMux(t, m)

env := errors.SyncEnvelope{
NodeID: "node-b",
Errors: []ServiceError{localErr("b:one", "node-b", 1000)},
NodeID: "uuid-peer",
Errors: []ServiceError{localErr("b:one", "uuid-peer", 1000)},
}
body, _ := json.Marshal(env)
resp, err := client.Post(srv.URL+"/v1/errors", "application/json", bytes.NewReader(body))
Expand All @@ -176,8 +179,61 @@ func TestHTTPIngestReconciles(t *testing.T) {
}

got := m.snapshot()
if len(got) != 1 || got[0].ID != "b:one" || got[0].NodeID != "node-b" {
t.Fatalf("after ingest, snapshot = %+v, want single node-b entry", got)
if len(got) != 1 || got[0].ID != "b:one" || got[0].NodeID != "uuid-peer" {
t.Fatalf("after ingest, snapshot = %+v, want single uuid-peer entry", got)
}
}

// TestHTTPIngestRejectsSpoofedNodeID: a pinned caller cannot reconcile
// under another node's identity — the envelope's nodeId must equal the
// authenticated caller UUID, otherwise the push is a 400 and nothing is
// reconciled.
func TestHTTPIngestRejectsSpoofedNodeID(t *testing.T) {
m := managerForNode("node-a")
srv, client := servePinnedErrorsMux(t, m)

env := errors.SyncEnvelope{
NodeID: "uuid-victim",
Errors: []ServiceError{localErr("v:one", "uuid-victim", 1000)},
}
body, _ := json.Marshal(env)
resp, err := client.Post(srv.URL+"/v1/errors", "application/json", bytes.NewReader(body))
if err != nil {
t.Fatalf("POST: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusBadRequest {
t.Fatalf("POST status = %d, want 400", resp.StatusCode)
}
if got := m.snapshot(); len(got) != 0 {
t.Fatalf("spoofed ingest reconciled %+v, want empty snapshot", got)
}
}

// TestHTTPIngestRejectsOversizedBody: a body larger than the 1 MiB cap is
// refused and nothing is reconciled, rather than decoded unbounded.
func TestHTTPIngestRejectsOversizedBody(t *testing.T) {
m := managerForNode("node-a")
srv, client := servePinnedErrorsMux(t, m)

env := errors.SyncEnvelope{
NodeID: "uuid-peer",
Errors: []ServiceError{localErr(strings.Repeat("x", 1<<20), "uuid-peer", 1000)},
}
body, _ := json.Marshal(env)
if len(body) <= maxIngestBodyBytes {
t.Fatalf("test body %d bytes does not exceed the %d-byte cap", len(body), maxIngestBodyBytes)
}
resp, err := client.Post(srv.URL+"/v1/errors", "application/json", bytes.NewReader(body))
if err != nil {
t.Fatalf("POST: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusBadRequest {
t.Fatalf("oversized POST status = %d, want 400", resp.StatusCode)
}
if got := m.snapshot(); len(got) != 0 {
t.Fatalf("oversized ingest reconciled %+v, want empty snapshot", got)
}
}

Expand Down