Merge pull request #361 from aojea/samone

sambox: an agent serves only itself, over A2A, contracted statically

Author: aojea

.github/k8s/sam-box-canary-template.yaml

@@ -147,7 +147,7 @@ spec:
             echo "the sandbox's interfaces (expect lo and tun0 only): $(ip -o link show | cut -d: -f2 | tr -d ' ' | paste -sd,)"
             while true; do
               echo "allowlisted destination (expect 200): $(curl -s -o /dev/null -w '%{http_code}' http://example.com/)"
-              echo "the node's own API (expect 403): $(curl -s -o /dev/null -w '%{http_code}' http://mesh.sam.alt/sam/service/register)"
+              echo "the node's own API (expect 403): $(curl -s -o /dev/null -w '%{http_code}' http://mesh.sam.alt/sam/service/discover)"
               curl -s -o /dev/null http://blocked.example/ && echo "unlisted destination was NOT refused" || echo "unlisted destination refused (expected)"
               sleep 30
             done

cmd/sam-box/main.go

@@ -51,6 +51,7 @@ func main() {
 		logLevel      string
 
 		agentIngressSocket string
+		ingressListen      string
 	)
 
 	rootCmd := &cobra.Command{
@@ -78,12 +79,12 @@ func main() {
 			if err := verifyBundleCredential(cmd.Context(), bundlePath, issuer, audience, insecure); err != nil {
 				return err
 			}
-			ingress, err := resolveIngress(bundlePath, sidecarSocket, agentIngressSocket)
+			ingress, err := resolveIngress(bundlePath, ingressListen, agentIngressSocket)
 			if err != nil {
 				return err
 			}
 			if ingress != nil {
-				defer ingress.Close(context.WithoutCancel(cmd.Context()))
+				defer ingress.Close()
 			}
 
 			listener, err := sambox.ListenSandboxSocket(sandboxSocket)
@@ -107,7 +108,6 @@ func main() {
 					Router:        &sambox.Router{Egress: egress},
 					SidecarSocket: sidecarSocket,
 					AgentID:       agentID,
-					Ingress:       ingress,
 				},
 			}
 
@@ -136,6 +136,7 @@ func main() {
 	runCmd.Flags().BoolVar(&insecure, "insecure-unverified-bundle", false, "Trust the bundle's declared identity without a credential to back it, letting whoever can write the file decide which agent this sandbox is")
 	runCmd.Flags().StringVar(&metricsAddr, "metrics-addr", "", "Serve unauthenticated Prometheus metrics on this address, e.g. 127.0.0.1:9600; off by default")
 	runCmd.Flags().StringVar(&agentIngressSocket, "agent-ingress-socket", "", "Path to the sandbox's reverse channel, served by nano-init --ingress-socket; required to reach an agent that serves the mesh, because an isolated sandbox cannot be dialled")
+	runCmd.Flags().StringVar(&ingressListen, "ingress-listen", "127.0.0.1:7080", "Stable address the gateway's mesh-facing ingress listens on; the node's configuration declares services with this address as their backend")
 	runCmd.Flags().StringVar(&logLevel, "log-level", "info", "Log level (debug, info, warn, error)")
 	for _, required := range []string{"socket", "sidecar-socket"} {
 		if err := runCmd.MarkFlagRequired(required); err != nil {
@@ -222,31 +223,39 @@ func verifyBundleCredential(ctx context.Context, bundlePath, issuer, audience st
 
 // resolveIngress prepares what the agent is permitted to serve. Nil means
 // nothing, which is the case for a sandbox that only calls out.
-func resolveIngress(bundlePath, sidecarSocket, agentIngressSocket string) (*sambox.IngressManager, error) {
+func resolveIngress(bundlePath, ingressListen, agentIngressSocket string) (*sambox.IngressManager, error) {
 	if bundlePath == "" {
 		return nil, nil
 	}
 	bundle, err := sambox.LoadAgentBundle(bundlePath)
 	if err != nil {
 		return nil, err
 	}
-	if len(bundle.Ingress) == 0 {
+	if bundle.Serves == nil {
 		return nil, nil
 	}
 	if agentIngressSocket == "" {
 		// Refused rather than degraded. Without a channel into the sandbox the
 		// only address left is one in this process's network namespace, which
 		// is the pod's: the node's API and every sidecar are on that loopback,
 		// and the port would be the agent's to choose.
-		return nil, fmt.Errorf("agent %s may serve %d mesh service(s), but --agent-ingress-socket is not set. "+
+		return nil, fmt.Errorf("agent %s serves a2a://%s, but --agent-ingress-socket is not set. "+
 			"Point it at the path nano-init --ingress-socket serves; without it there is no way into the "+
 			"sandbox, and delivering to this process's own network namespace would reach the gateway's "+
-			"neighbours instead of the agent", bundle.Agent.ID, len(bundle.Ingress))
-	}
-	logger.Infof("Agent %s may serve %d mesh service(s) once it announces them", bundle.Agent.ID, len(bundle.Ingress))
-	return &sambox.IngressManager{
-		SidecarSocket: sidecarSocket,
-		Allowed:       bundle.Ingress,
-		AgentSocket:   agentIngressSocket,
-	}, nil
+			"neighbours instead of the agent", bundle.Agent.ID, bundle.Serves.Name)
+	}
+	manager := &sambox.IngressManager{
+		ListenAddr:  ingressListen,
+		Serves:      *bundle.Serves,
+		AgentSocket: agentIngressSocket,
+	}
+	// The routes are the bundle's contract and exist from startup; the node's
+	// config declares services backed by this address, and its backend probe
+	// fails until the agent actually binds its contracted port.
+	addr, err := manager.Start()
+	if err != nil {
+		return nil, fmt.Errorf("serving ingress on %s: %w", ingressListen, err)
+	}
+	logger.Infof("Agent %s serves a2a://%s; ingress at http://%s", bundle.Agent.ID, bundle.Serves.Name, addr)
+	return manager, nil
 }

internal/node/a2a_service.go

@@ -56,6 +56,36 @@ func (s *A2AService) Init(ctx context.Context) error {
 	return nil
 }
 
+// Probe asks the backend for its agent card, which is the protocol's own
+// definition of ready: an A2A agent is up exactly when it serves its card.
+// Gating advertisement on it lets a service be declared before its agent is
+// (a sandbox that has not bound its port yet probes as down and stays out of
+// discovery). Deliberately not cached, like the MCP probe.
+func (s *A2AService) Probe(ctx context.Context) error {
+	target, ok := s.backend.(*api.RegisterServiceRequest_TargetUrl)
+	if !ok {
+		return fmt.Errorf("a2a service %q has no URL backend to probe", s.info.GetName())
+	}
+	cardURL := strings.TrimSuffix(target.TargetUrl, "/") + "/.well-known/agent-card.json"
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, cardURL, nil)
+	if err != nil {
+		return err
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("fetch agent card of %q: %w", s.info.GetName(), err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("agent card of %q: %s", s.info.GetName(), resp.Status)
+	}
+	var card map[string]any
+	if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&card); err != nil {
+		return fmt.Errorf("agent card of %q is not JSON: %w", s.info.GetName(), err)
+	}
+	return nil
+}
+
 // a2aEgressGate runs the caller-side A2A checks on a raw egress request:
 // the fail-closed labels gate. On refusal it writes the HTTP error itself
 // and returns ok=false.

internal/node/a2a_service_test.go

@@ -65,6 +65,57 @@ func TestNewServiceFromRequestA2A(t *testing.T) {
 	}
 }
 
+// TestA2AServiceProbe pins advertisement gating on the agent card: an a2a
+// service may be declared before its agent is up (a sandbox that has not
+// bound its port), and must probe as down until the card is served.
+func TestA2AServiceProbe(t *testing.T) {
+	newSvc := func(target string) *A2AService {
+		return &A2AService{baseService: baseService{
+			info:    &api.ServiceInfo{Type: api.ServiceType_SERVICE_TYPE_A2A, Name: "agent"},
+			backend: &api.RegisterServiceRequest_TargetUrl{TargetUrl: target},
+		}}
+	}
+
+	t.Run("card served means ready", func(t *testing.T) {
+		backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if r.URL.Path != "/.well-known/agent-card.json" {
+				http.NotFound(w, r)
+				return
+			}
+			_, _ = w.Write([]byte(`{"name":"agent","version":"1.0.0"}`))
+		}))
+		defer backend.Close()
+		if err := newSvc(backend.URL).Probe(context.Background()); err != nil {
+			t.Fatalf("Probe with a served card: %v", err)
+		}
+	})
+
+	t.Run("no card means not ready", func(t *testing.T) {
+		backend := httptest.NewServer(http.NotFoundHandler())
+		defer backend.Close()
+		if err := newSvc(backend.URL).Probe(context.Background()); err == nil {
+			t.Fatal("Probe must fail while the card is not served")
+		}
+	})
+
+	t.Run("dead backend means not ready", func(t *testing.T) {
+		// A declared sandbox service whose agent never bound its port.
+		if err := newSvc("http://127.0.0.1:1").Probe(context.Background()); err == nil {
+			t.Fatal("Probe must fail when nothing listens")
+		}
+	})
+
+	t.Run("non-json card means not ready", func(t *testing.T) {
+		backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			_, _ = w.Write([]byte("<html>login page</html>"))
+		}))
+		defer backend.Close()
+		if err := newSvc(backend.URL).Probe(context.Background()); err == nil {
+			t.Fatal("Probe must fail on a non-JSON card")
+		}
+	})
+}
+
 func TestA2AEgressHookNonA2APassthrough(t *testing.T) {
 	rec := httptest.NewRecorder()
 	req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/mcp/svc/foo", nil)

internal/node/node.go

@@ -1583,6 +1583,15 @@ func (n *SamNode) HandleAuthHandshake(s network.Stream) {
 }
 
 func (n *SamNode) RegisterService(ctx context.Context, req *api.RegisterServiceRequest) error {
+	if req.Service == nil {
+		return fmt.Errorf("service field is required")
+	}
+	if req.Service.Name == "" || req.Service.Type == api.ServiceType_SERVICE_TYPE_UNSPECIFIED {
+		return fmt.Errorf("service name and type are required")
+	}
+	if req.Backend == nil {
+		return fmt.Errorf("service backend is required")
+	}
 	svc, err := NewServiceFromRequest(req)
 	if err != nil {
 		return err

internal/node/sidecar.go

@@ -23,7 +23,6 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
-	"io"
 	"net"
 	"net/http"
 	"net/http/httputil"
@@ -38,7 +37,6 @@ import (
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
-	"google.golang.org/protobuf/encoding/protojson"
 )
 
 // StartSidecarServer serves the node's local API on a TCP address, on a Unix
@@ -57,12 +55,10 @@ func StartSidecarServer(node *SamNode, addr, socketPath, token, certFile, keyFil
 
 	// Protected endpoints. allowAuthorizationFallback=true is safe here: none of
 	// these ever forward the inbound Authorization header to another service.
-	mux.Handle("/sam/service/register", withAuth(token, true, withMeshConnection(node, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		handleRegisterService(node, w, r)
-	}))))
-	mux.Handle("/sam/service/unregister", withAuth(token, true, withMeshConnection(node, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		handleUnregisterService(node, w, r)
-	}))))
+	// Services are declared in the node's configuration and registered at
+	// startup; there is deliberately no runtime registration surface, so no
+	// credential held by an agent can point the mesh at a new backend or
+	// withdraw a sibling service.
 	mux.Handle("/sam/service/discover", withAuth(token, true, withMeshConnection(node, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		handleDiscoverService(node, w, r)
 	}))))
@@ -448,90 +444,10 @@ func constantTimeEqual(got, want string) bool {
 	return subtle.ConstantTimeCompare(gotHash[:], wantHash[:]) == 1
 }
 
-type ServiceRequest struct {
-	ServiceName string `json:"service_name"`
-}
-
 // maxRequestBodyBytes caps request bodies read into memory to guard
 // against memory-exhaustion from oversized payloads.
 const maxRequestBodyBytes = 1 << 20 // 1 MiB
 
-func handleRegisterService(node *SamNode, w http.ResponseWriter, r *http.Request) {
-	if r.Method != http.MethodPost {
-		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
-		return
-	}
-
-	r.Body = http.MaxBytesReader(w, r.Body, maxRequestBodyBytes)
-	body, err := io.ReadAll(r.Body)
-	if err != nil {
-		http.Error(w, "Failed to read request body", http.StatusInternalServerError)
-		return
-	}
-	_ = r.Body.Close()
-
-	var req api.RegisterServiceRequest
-	if err := protojson.Unmarshal(body, &req); err != nil {
-		http.Error(w, fmt.Sprintf("Invalid request body: %v", err), http.StatusBadRequest)
-		return
-	}
-
-	if req.Service == nil {
-		http.Error(w, "service field is required", http.StatusBadRequest)
-		return
-	}
-
-	if req.Service.Name == "" || req.Service.Type == api.ServiceType_SERVICE_TYPE_UNSPECIFIED {
-		http.Error(w, "name and type are required", http.StatusBadRequest)
-		return
-	}
-
-	if req.Backend == nil {
-		http.Error(w, "backend is required", http.StatusBadRequest)
-		return
-	}
-
-	if err := node.RegisterService(r.Context(), &req); err != nil {
-		logger.Errorf("Failed to register service: %v", err)
-		http.Error(w, fmt.Sprintf("Failed to register service: %v", err), http.StatusInternalServerError)
-		return
-	}
-
-	w.WriteHeader(http.StatusOK)
-	if _, err := w.Write([]byte("Service registered")); err != nil {
-		logger.Errorf("Failed to write response: %v", err)
-	}
-}
-
-func handleUnregisterService(node *SamNode, w http.ResponseWriter, r *http.Request) {
-	if r.Method != http.MethodPost {
-		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
-		return
-	}
-
-	var req api.ServiceInfo
-	r.Body = http.MaxBytesReader(w, r.Body, maxRequestBodyBytes)
-	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
-		http.Error(w, "Invalid request body", http.StatusBadRequest)
-		return
-	}
-
-	if req.Name == "" {
-		http.Error(w, "name is required", http.StatusBadRequest)
-		return
-	}
-
-	if err := node.UnregisterService(r.Context(), req.Name); err != nil {
-		http.Error(w, fmt.Sprintf("Failed to unregister service: %v", err), http.StatusInternalServerError)
-		return
-	}
-
-	w.WriteHeader(http.StatusOK)
-	if _, err := w.Write([]byte("Service unregistered")); err != nil {
-		logger.Errorf("Failed to write response: %v", err)
-	}
-}
-
 func handleDiscoverService(node *SamNode, w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodGet {
 		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)