peer: reject invalid X-DART-Hop values at both servers

docs/engine.md

@@ -360,6 +360,7 @@ go test ./internal/engine/ -cover -count=1
 | `TestEngineP2PPeerMissFallsBackToOrigin` | empty peer (404) → origin fallback, correct bytes |
 | `TestEngineP2PSelfOwnerUsesOrigin` | self is owner → peers not consulted, origin used |
 | `TestPeerSourceRelayFetchOnBehalf` | relay Source fetches-on-behalf and serves a block it did not hold |
+| **`TestRelayHopBoundary`** | **both peer-facing sources: hop == maxHop-1 still relays; hop == maxHop declines without touching the origin** |
 | `TestTreeMultiHopRelay` | 3-node fanout=1 chain: a tail request populates every node via the relay chain |
 | `TestPeerStreamSourceLocalHit` | locally-held block streamed from the store |
 | `TestPeerStreamSourceRootFetchesOrigin` | tree root satisfies a relay request from origin and caches it |

docs/peer.md

@@ -31,10 +31,19 @@ X-DART-Hop:    <n>              (relay depth, for loop safety)
   `store.BlockKey.Block` in base-10.
 - `X-DART-Origin` lets a relay-capable Source fetch a block it does not hold
   (via its own parent/origin); `X-DART-Hop` bounds relay recursion.
+- `X-DART-Hop` is optional (absent = depth 0) and must be a non-negative
+  decimal int; any other value — malformed, negative, or out of range — is
+  rejected with `400` before the Source is invoked. Accepting a negative hop
+  would weaken the engine's relay-loop bound (`hop >= maxHop`, incremented at
+  each relay): a negative start would delay the cutoff, a huge negative one
+  effectively forever. The bound itself (`maxHop`) is enforced by the engine,
+  not here, so any non-negative value — including one at or past the bound —
+  is a valid wire value at this layer.
 - Responses: `200` + block bytes (with `X-DART-Node`; `Content-Length` when
   the source knows the size up front, chunked otherwise — e.g. on the
   cut-through relay path), `404` if the peer cannot provide the block, `400`
-  for a malformed path, `405` for non-GET, `500` on a source error, and `502` +
+  for a malformed path or an invalid `X-DART-Hop`, `405` for non-GET, `500`
+  on a source error, and `502` +
   `X-DART-Upstream-Status: <origin code>` when a relay's origin fetch was
   refused (§3.6 — the peer is fine; the caller's credential is not).
 - The path has no embedded URLs, so it is parsed from `URL.Path` (no `//` trap).
@@ -330,6 +339,8 @@ go test ./internal/peer/ -race -count=1
 | `TestClientMiss` | a not-held block yields `held=false, err=nil` (404) |
 | `TestServerNodeHeader` | `X-DART-Node` echoed on responses |
 | `TestServerBadPathAndMethod` | malformed paths → 400/404; non-GET → 405 |
+| **`TestServerHopValidation`** | **both servers: absent hop = 0; non-negative hops (incl. ≥ the engine's relay bound) accepted; negative/malformed/overflow hops → 400 and the Source is never invoked** |
+| `TestParseHop` | decoder edges: empty, leading plus, `-0`, overflow |
 | `TestServerSourceError` | a source error → 500 |
 | `TestClientConnError` | a closed peer port surfaces a transport error |
 | `TestParseBlockPath` | path parsing incl. max uint64 and rejects |

internal/engine/hop_bound_test.go

@@ -0,0 +1,88 @@
+package engine
+
+import (
+	"bytes"
+	"context"
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	"github.com/data-accelerator/dart/internal/chunk"
+	"github.com/data-accelerator/dart/internal/cluster"
+	"github.com/data-accelerator/dart/internal/peer"
+	"github.com/data-accelerator/dart/internal/store"
+)
+
+// TestRelayHopBoundary pins the maxHop gate on both peer-facing sources
+// (PeerSource and PeerStreamSource): at hop == maxHop-1 a miss still relays
+// (falls through to origin here, this node being a lone root), while at
+// hop == maxHop the request is declined without the origin being touched at
+// all. Together with the peer transport rejecting negative/malformed hops
+// (internal/peer), this is the whole loop-safety contract: hop enters the
+// engine in [0, maxHop), grows by one per relay, and is cut off at maxHop.
+func TestRelayHopBoundary(t *testing.T) {
+	content := blob(100)
+
+	cases := []struct {
+		name      string
+		hop       int
+		wantHeld  bool
+		wantFetch bool
+	}{
+		{"one below the bound relays", maxHop - 1, true, true},
+		{"at the bound declines", maxHop, false, false},
+	}
+
+	sources := map[string]func(e *Engine, req peer.BlockRequest, buf *bytes.Buffer) (bool, error){
+		"buffered": func(e *Engine, req peer.BlockRequest, buf *bytes.Buffer) (bool, error) {
+			data, held, err := e.PeerSource()(context.Background(), req)
+			if held {
+				buf.Write(data)
+			}
+			return held, err
+		},
+		"stream": func(e *Engine, req peer.BlockRequest, buf *bytes.Buffer) (bool, error) {
+			_, held, err := e.PeerStreamSource()(context.Background(), req, buf, func(int64) {})
+			return held, err
+		},
+	}
+
+	for srcName, call := range sources {
+		for _, tc := range cases {
+			t.Run(srcName+"/"+tc.name, func(t *testing.T) {
+				var originHits int64
+				origin := countingOrigin(t, content, &originHits)
+
+				prov := cluster.NewStaticProvider(
+					cluster.Member{ID: "R", Addr: "127.0.0.1:1", Weight: 1, State: cluster.Ready})
+				e, err := New(Options{
+					Chunk: testCfg(), Store: openStoreAt(t), Fetcher: newFetcher(),
+					Cluster: prov, Peer: peer.NewClient(), SelfID: "R",
+				})
+				if err != nil {
+					t.Fatalf("New: %v", err)
+				}
+
+				url := origin.URL + "/blob"
+				oid, _ := chunk.ObjectID(url)
+				key := store.BlockKey{Chunk: chunk.ChunkKey("dart", oid, 0), Block: 0}
+				req := peer.BlockRequest{Key: key, URL: url, Hop: tc.hop}
+
+				var buf bytes.Buffer
+				held, err := call(e, req, &buf)
+				if err != nil {
+					t.Fatalf("source error: %v", err)
+				}
+				if held != tc.wantHeld {
+					t.Errorf("held = %v, want %v", held, tc.wantHeld)
+				}
+				if gotFetch := atomic.LoadInt64(&originHits) > 0; gotFetch != tc.wantFetch {
+					t.Errorf("origin contacted = %v, want %v", gotFetch, tc.wantFetch)
+				}
+				if tc.wantHeld && !bytes.Equal(buf.Bytes(), content[:16]) {
+					t.Errorf("served bytes mismatch: got %d bytes %q", buf.Len(), strings.TrimSpace(buf.String()))
+				}
+			})
+		}
+	}
+}

internal/peer/hop_test.go

@@ -0,0 +1,120 @@
+package peer
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+// TestServerHopValidation pins the X-DART-Hop domain on both peer server
+// implementations (buffered Server and cut-through StreamServer): the header is
+// optional (absent = depth 0) and any non-negative decimal is accepted — the
+// relay-depth bound itself is enforced by the engine, not the transport — but a
+// malformed, negative, or out-of-range value is rejected with 400 before the
+// source is ever invoked. A negative hop used to sail through: the engine only
+// declines at hop >= maxHop and every relay increments the value, so a negative
+// start delayed the loop-safety cutoff (a huge negative one effectively
+// forever) exactly when membership skew creates a relay cycle.
+func TestServerHopValidation(t *testing.T) {
+	blockPath := "/peer/v1/block/abcdef/7"
+
+	servers := map[string]func(record func(hop int)) http.Handler{
+		"buffered": func(record func(hop int)) http.Handler {
+			return &Server{NodeID: "n", Src: func(_ context.Context, req BlockRequest) ([]byte, bool, error) {
+				record(req.Hop)
+				return []byte("data"), true, nil
+			}}
+		},
+		"stream": func(record func(hop int)) http.Handler {
+			return &StreamServer{NodeID: "n", Src: func(_ context.Context, req BlockRequest, w io.Writer, sizer func(int64)) (int64, bool, error) {
+				record(req.Hop)
+				sizer(4)
+				n, err := w.Write([]byte("data"))
+				return int64(n), true, err
+			}}
+		},
+	}
+
+	cases := []struct {
+		name       string
+		hop        string // header value; set=false means the header is absent
+		set        bool
+		wantStatus int
+		wantHop    int // hop the source must observe (valid cases only)
+	}{
+		{"absent means depth zero", "", false, http.StatusOK, 0},
+		{"zero", "0", true, http.StatusOK, 0},
+		{"below relay bound", "63", true, http.StatusOK, 63},
+		// The transport accepts any non-negative depth; declining at the relay
+		// bound (maxHop) is the engine's job, so 64 and beyond are still valid
+		// wire values here.
+		{"at relay bound", "64", true, http.StatusOK, 64},
+		{"large", "4096", true, http.StatusOK, 4096},
+		{"negative one", "-1", true, http.StatusBadRequest, 0},
+		{"negative large", "-4096", true, http.StatusBadRequest, 0},
+		{"min int64", "-9223372036854775808", true, http.StatusBadRequest, 0},
+		{"overflow", "9223372036854775808", true, http.StatusBadRequest, 0},
+		{"non-numeric", "abc", true, http.StatusBadRequest, 0},
+		{"float", "1.5", true, http.StatusBadRequest, 0},
+		{"leading space", " 1", true, http.StatusBadRequest, 0},
+		{"grouped digits", "1_000", true, http.StatusBadRequest, 0},
+	}
+
+	for srvName, build := range servers {
+		for _, tc := range cases {
+			t.Run(srvName+"/"+tc.name, func(t *testing.T) {
+				var calls int
+				var gotHop int
+				h := build(func(hop int) { calls++; gotHop = hop })
+
+				r := httptest.NewRequest(http.MethodGet, blockPath, nil)
+				if tc.set {
+					r.Header.Set(HeaderHop, tc.hop)
+				}
+				rec := httptest.NewRecorder()
+				h.ServeHTTP(rec, r)
+
+				if rec.Code != tc.wantStatus {
+					t.Fatalf("status = %d, want %d (body %q)", rec.Code, tc.wantStatus, rec.Body.String())
+				}
+				if tc.wantStatus == http.StatusOK {
+					if calls != 1 {
+						t.Fatalf("source called %d times, want 1", calls)
+					}
+					if gotHop != tc.wantHop {
+						t.Errorf("source observed hop %d, want %d", gotHop, tc.wantHop)
+					}
+				} else if calls != 0 {
+					t.Errorf("source called %d times on a rejected request, want 0", calls)
+				}
+			})
+		}
+	}
+}
+
+// TestParseHop exercises the decoder directly, including the values strconv
+// accepts that the wire never produces (a leading plus).
+func TestParseHop(t *testing.T) {
+	cases := []struct {
+		in      string
+		wantHop int
+		wantOK  bool
+	}{
+		{"", 0, true},
+		{"0", 0, true},
+		{"7", 7, true},
+		{"+5", 5, true}, // strconv.Atoi semantics; harmless and accepted
+		{"-1", 0, false},
+		{"-0", 0, true}, // Atoi("-0") == 0, which is in domain
+		{"abc", 0, false},
+		{"9223372036854775808", 0, false},
+	}
+	for _, tc := range cases {
+		hop, ok := parseHop(tc.in)
+		if ok != tc.wantOK || (ok && hop != tc.wantHop) {
+			t.Errorf("parseHop(%q) = (%d, %v), want (%d, %v)", tc.in, hop, ok, tc.wantHop, tc.wantOK)
+		}
+	}
+}

internal/peer/peer.go

@@ -62,7 +62,8 @@ type BlockRequest struct {
 	// (the peer should serve only what it already holds).
 	URL string
 	// Hop is the relay depth (X-DART-Hop); incremented at each relay to bound
-	// recursion and detect loops.
+	// recursion and detect loops. Always non-negative on the wire: the servers
+	// reject a malformed or negative header with 400 before calling the Source.
 	Hop int
 }
 
@@ -100,7 +101,11 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "bad block path", http.StatusBadRequest)
 		return
 	}
-	hop, _ := strconv.Atoi(r.Header.Get(HeaderHop))
+	hop, ok := parseHop(r.Header.Get(HeaderHop))
+	if !ok {
+		http.Error(w, "bad "+HeaderHop+" header", http.StatusBadRequest)
+		return
+	}
 	req := BlockRequest{Key: key, URL: r.Header.Get(HeaderOrigin), Hop: hop}
 
 	data, held, err := s.Src(r.Context(), req)
@@ -128,6 +133,24 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	_, _ = w.Write(data)
 }
 
+// parseHop decodes the X-DART-Hop header. The header is optional: absent means
+// depth 0 (a direct, non-relayed request). Any value outside the valid
+// domain — malformed, negative, or out of int range — is rejected: the engine
+// bounds relay recursion with `hop >= maxHop`, and each relay increments the
+// value, so a negative start would delay that cutoff (a huge negative one
+// effectively forever), weakening the loop-safety bound exactly when
+// membership skew creates a relay cycle.
+func parseHop(h string) (int, bool) {
+	if h == "" {
+		return 0, true
+	}
+	hop, err := strconv.Atoi(h)
+	if err != nil || hop < 0 {
+		return 0, false
+	}
+	return hop, true
+}
+
 // parseBlockPath parses "/peer/v1/block/<chunkHex>/<blockIndex>".
 func parseBlockPath(p string) (store.BlockKey, bool) {
 	rest, ok := strings.CutPrefix(p, blockPath)

internal/peer/stream.go

@@ -92,7 +92,11 @@ func (s *StreamServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "bad block path", http.StatusBadRequest)
 		return
 	}
-	hop, _ := strconv.Atoi(r.Header.Get(HeaderHop))
+	hop, ok := parseHop(r.Header.Get(HeaderHop))
+	if !ok {
+		http.Error(w, "bad "+HeaderHop+" header", http.StatusBadRequest)
+		return
+	}
 	req := BlockRequest{Key: key, URL: r.Header.Get(HeaderOrigin), Hop: hop}
 
 	// Buffer the head of the stream so a "not held" or an immediate error can