Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
6 changes: 6 additions & 0 deletions api/ingress.go
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,12 @@ const (
// MaxIngressAllowEntries bounds the number of caller principal patterns a
// single ServiceIngress.allow list may carry in directory announcements.
MaxIngressAllowEntries = 64

// MaxIngressBodyBytes caps the body of one external ingress request. The
// router applies it on its public listener and the node applies it again
// on /mesh/ingress/1.0.0, so a router that forwards without the cap gains
// nothing.
MaxIngressBodyBytes int64 = 32 << 20
)

// IsInboundServiceType reports whether t is an inbound service type that a
Expand Down
2 changes: 1 addition & 1 deletion internal/identity/biscuit.go
Original file line number Diff line number Diff line change
Expand Up @@ -889,7 +889,7 @@ func RequireIngress(b *biscuit.Biscuit, key ed25519.PublicKey, svcType, svcName,
authorizer.AddPolicy(p)
}
if err := authorizer.Authorize(); err != nil {
return fmt.Errorf("credential does not grant %s ingress for %s://%s", mode, svcType, svcName)
return fmt.Errorf("credential does not grant %s ingress for %s://%s: %w", mode, svcType, svcName, err)
}
return nil
}
Expand Down
58 changes: 57 additions & 1 deletion internal/node/ingress.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@ package node

import (
"context"
"errors"
"fmt"
"net/http"
"slices"
"sort"
Expand All @@ -37,6 +39,10 @@ const (
externalIngressOutcomeNotFound = "not_found"
externalIngressOutcomeRateLimited = "rate_limited"
externalIngressOutcomeInvalid = "invalid"
// externalIngressOutcomeUndecided is an authorizer that stopped at a run
// limit before it reached a verdict; the request is neither allowed nor
// denied.
externalIngressOutcomeUndecided = "undecided"
)

var externalIngressRequestsTotal = promauto.NewCounterVec(
Expand All @@ -56,6 +62,7 @@ func init() {
externalIngressOutcomeNotFound,
externalIngressOutcomeRateLimited,
externalIngressOutcomeInvalid,
externalIngressOutcomeUndecided,
} {
externalIngressRequestsTotal.WithLabelValues(mode, outcome)
}
Expand Down Expand Up @@ -180,7 +187,9 @@ func (n *AgentMeshNode) StartExternalIngressServer(ctx context.Context) error {
func (n *AgentMeshNode) externalIngressHandler() http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
rc := &readCloserWithCount{ReadCloser: r.Body}
r.Body = rc
// The router caps the body on its public listener; cap it here too so
// the node does not take the router's word for it.
r.Body = http.MaxBytesReader(w, rc, api.MaxIngressBodyBytes)
wc := &responseWriterWithCount{ResponseWriter: w, statusCode: http.StatusOK}
w = wc

Expand Down Expand Up @@ -334,6 +343,26 @@ func (n *AgentMeshNode) externalIngressHandler() http.Handler {
target = serviceTypeStr + "://" + serviceName
backendPath := "/" + upstreamPath

// The router checked this node's role ceiling before it forwarded; check
// it here too, against this node's own credential and the mesh policy
// it holds, so a router that skips the check exposes nothing.
if err := n.requireOwnIngressGrant(serviceTypeStr, serviceName, modeStr); err != nil {
if stopped, transient := authorizerLimit(err); stopped {
externalIngressRequestsTotal.WithLabelValues(modeMetric, externalIngressOutcomeUndecided).Inc()
logger.Warnf("[ExternalIngress] Own ingress grant for %s undecided via router %s: %v", target, remotePeer, err)
status := http.StatusInternalServerError
if transient {
status = http.StatusGatewayTimeout
}
refuse(w, status, "Authorization undecided", proxyStatusDestinationUnreachable)
return
}
externalIngressRequestsTotal.WithLabelValues(modeMetric, externalIngressOutcomeDeny).Inc()
logger.Warnf("[ExternalIngress] Refusing %s ingress to %s via router %s: %v", modeStr, target, remotePeer, err)
refuse(w, http.StatusForbidden, "Forbidden: this node's role does not grant external ingress for the service", proxyStatusDenied)
return
}

if effectiveMode == api.IngressAuthMode_INGRESS_AUTH_MODE_PASSTHROUGH {
api.SanitizeExternalIngressHeaders(r.Header, false)
r.Header.Set(api.HeaderPeerID, remotePeer.String())
Expand Down Expand Up @@ -476,6 +505,33 @@ func (n *AgentMeshNode) externalIngressHandler() http.Handler {
})
}

// requireOwnIngressGrant checks that this node's own credential, under the
// mesh policy it holds, grants exposing svcType://svcName on external ingress
// in mode: the same check the router runs before it forwards, repeated here
// so that the node does not depend on the router having run it.
func (n *AgentMeshNode) requireOwnIngressGrant(svcType, svcName, mode string) error {
token := n.GetIdentity()
if len(token) == 0 {
return errors.New("node holds no credential")
}
keys := n.trustedPublicKeys()
if len(keys) == 0 {
return errors.New("no trusted control plane keys loaded")
}
pid, err := n.localPeerID()
if err != nil {
return err
}
b, key, err := identity.VerifyBiscuitAndGetKey(token, pid, keys, n.BiscuitTimeout)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Performance Optimization: Cache Verified Node Identity\n\nCryptographically verifying the node's own Biscuit token (identity.VerifyBiscuitAndGetKey) on every single external ingress request is CPU-intensive and can become a performance bottleneck under high request volumes. Since the node's own identity token and the trusted control plane keys change very infrequently (only during credential refresh), the verified Biscuit b and verifying key should be cached on the AgentMeshNode struct (e.g., protected by a mutex or stored in an atomic.Pointer / sync.Once pattern) and invalidated/re-verified only when the identity token is updated.\n\nThis avoids redundant cryptographic signature verification and Datalog parsing on the hot path of every ingress request.

if err != nil {
return fmt.Errorf("own credential: %w", err)
}
n.MeshPolicyMu.RLock()
meshRules := n.MeshPolicyRules
n.MeshPolicyMu.RUnlock()
return identity.RequireIngress(b, key, svcType, svcName, mode, meshRules, n.BiscuitTimeout)
}

func (n *AgentMeshNode) buildExternalIngressEvalParams(serviceType, serviceName, method, path string, allow []string) (identity.IngressEvalParams, error) {
pol, err := n.meshPolicy()
if err != nil {
Expand Down
146 changes: 146 additions & 0 deletions internal/node/ingress_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import (
"crypto/rand"
"encoding/base64"
"errors"
"io"
"net/http"
"net/http/httptest"
"os"
Expand All @@ -30,6 +31,7 @@ import (
"time"

"github.com/eclipse-biscuit/biscuit-go/v2"
"github.com/eclipse-biscuit/biscuit-go/v2/datalog"
"github.com/eclipse-biscuit/biscuit-go/v2/parser"
"github.com/golang-jwt/jwt/v5"
"github.com/google/agentmesh/api"
Expand Down Expand Up @@ -355,6 +357,18 @@ func TestExternalIngressHandler(t *testing.T) {
AllowedServices: []string{"a2a://assistant", "mcp://files"},
AllowedTargets: []string{"email:owner@example.com"},
},
{
// The serving node's role: the ingress ceiling the control plane
// renders as granted_ingress_* rules. Nothing grants a2a://ungranted.
Name: "provider",
AllowedIngress: []*api.IngressGrant{
{Services: []string{"a2a://assistant", "mcp://files"}},
{
Services: []string{"a2a://webhook", "a2a://sink"},
AllowedAuthModes: []api.IngressAuthMode{api.IngressAuthMode_INGRESS_AUTH_MODE_PASSTHROUGH},
},
},
},
}
policyBindings := []*api.PolicyBinding{
{
Expand Down Expand Up @@ -473,6 +487,19 @@ func TestExternalIngressHandler(t *testing.T) {
},
handler: recordHandler,
},
{
// Opted in by the service owner, but the node's role does not
// grant it: the router's ceiling check is what keeps it out, and
// the node must not rely on the router having run it.
info: &api.ServiceInfo{
Name: "ungranted",
Type: api.ServiceType_SERVICE_TYPE_A2A,
Ingress: &api.ServiceIngress{
AuthMode: api.IngressAuthMode_INGRESS_AUTH_MODE_PASSTHROUGH,
},
},
handler: recordHandler,
},
} {
if err := reg.Register(context.Background(), svc); err != nil {
t.Fatalf("Register %s: %v", svc.info.Name, err)
Expand Down Expand Up @@ -695,6 +722,62 @@ func TestExternalIngressHandler(t *testing.T) {
}
})

t.Run("service the node's role does not grant is refused even from a router", func(t *testing.T) {
before := counterValue(t, externalIngressRequestsTotal.WithLabelValues(api.IngressAuthModeStringPassthrough, externalIngressOutcomeDeny))
req := httptest.NewRequest(http.MethodPost, "http://mesh/a2a/ungranted/events", strings.NewReader(`{"event":"ping"}`))
req.RemoteAddr = routerPID.String()
req.Header.Set(api.HeaderMeshBiscuit, routerB64)
req.Header.Set(api.HeaderMeshIngress, api.IngressAuthModeStringPassthrough)
req.Header.Set("Authorization", "******")
capturedPath = ""
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if rec.Code != http.StatusForbidden {
t.Fatalf("ungranted passthrough status = %d, want 403: %s", rec.Code, rec.Body.String())
}
if capturedPath != "" {
t.Errorf("backend was called at %q, want no call", capturedPath)
}
after := counterValue(t, externalIngressRequestsTotal.WithLabelValues(api.IngressAuthModeStringPassthrough, externalIngressOutcomeDeny))
if after != before+1 {
t.Errorf("deny counter = %v, want %v", after, before+1)
}
})

t.Run("body above the ingress cap is cut at the node", func(t *testing.T) {
var readErr error
sink := &fakeService{
info: &api.ServiceInfo{
Name: "sink",
Type: api.ServiceType_SERVICE_TYPE_A2A,
Ingress: &api.ServiceIngress{AuthMode: api.IngressAuthMode_INGRESS_AUTH_MODE_PASSTHROUGH},
},
handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_, readErr = io.Copy(io.Discard, r.Body)
w.WriteHeader(http.StatusOK)
}),
}
if err := reg.Register(context.Background(), sink); err != nil {
t.Fatalf("Register sink: %v", err)
}
t.Cleanup(func() { _ = reg.Unregister(context.Background(), "sink") })

body := io.LimitReader(zeroReader{}, api.MaxIngressBodyBytes+1)
req := httptest.NewRequest(http.MethodPost, "http://mesh/a2a/sink/", body)
req.RemoteAddr = routerPID.String()
req.Header.Set(api.HeaderMeshBiscuit, routerB64)
req.Header.Set(api.HeaderMeshIngress, api.IngressAuthModeStringPassthrough)
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200: %s", rec.Code, rec.Body.String())
}
var maxErr *http.MaxBytesError
if !errors.As(readErr, &maxErr) {
t.Fatalf("backend read error = %v, want *http.MaxBytesError", readErr)
}
})

t.Run("mode mismatch header rejected fail-closed", func(t *testing.T) {
req := httptest.NewRequest(http.MethodPost, "http://mesh/a2a/assistant/", strings.NewReader(`{}`))
req.RemoteAddr = routerPID.String()
Expand Down Expand Up @@ -941,3 +1024,66 @@ func TestServiceRegistry_UnannouncedAndExternalIngressGate(t *testing.T) {
}
})
}

// zeroReader yields zero bytes without end; LimitReader sizes it.
type zeroReader struct{}

func (zeroReader) Read(p []byte) (int, error) {
clear(p)
return len(p), nil
}

// The node's own ingress grant is read from its credential under the mesh
// policy it holds. A grant passes; a role without one is refused; an
// authorizer that stops at a run limit is reported as such, not as a refusal,
// so the handler can answer 500 or 504 instead of 403.
func TestRequireOwnIngressGrant(t *testing.T) {
cpPub, cpPriv, err := ed25519.GenerateKey(nil)
if err != nil {
t.Fatal(err)
}
pid, priv := mustTestPeerID(t)
exp := time.Now().Add(time.Hour)
trusted := []TrustedKey{{Key: cpPub, ReceivedAt: time.Now()}}

granting := &api.PolicyRole{Name: api.RoleNode, AllowedIngress: []*api.IngressGrant{{Services: []string{"a2a://assistant"}}}}
withGrant, err := identity.MintBootstrapBiscuitToken(cpPriv, pid, api.RoleNode, exp, []*api.PolicyRole{granting}, nil)
if err != nil {
t.Fatal(err)
}
withoutGrant, err := identity.MintBootstrapBiscuitToken(cpPriv, pid, api.RoleNode, exp, []*api.PolicyRole{{Name: api.RoleNode}}, nil)
if err != nil {
t.Fatal(err)
}

newNode := func(token []byte, rules []biscuit.Rule) *AgentMeshNode {
n := &AgentMeshNode{config: Options{PrivKey: priv}, trustedKeys: trusted, BiscuitTimeout: 500 * time.Millisecond, MeshPolicyRules: rules}
n.SetIdentityCache(token)
return n
}

if err := newNode(withGrant, nil).requireOwnIngressGrant("a2a", "assistant", api.IngressAuthModeStringEdge); err != nil {
t.Fatalf("edge ingress on a granted service: %v, want nil", err)
}
if err := newNode(withGrant, nil).requireOwnIngressGrant("a2a", "assistant", api.IngressAuthModeStringPassthrough); err == nil {
t.Fatal("passthrough ingress without an explicit mode grant: nil, want a refusal")
}
err = newNode(withoutGrant, nil).requireOwnIngressGrant("a2a", "assistant", api.IngressAuthModeStringEdge)
if err == nil {
t.Fatal("ingress on a role without a grant: nil, want a refusal")
}
if stopped, _ := authorizerLimit(err); stopped {
t.Fatalf("refusal reported as a run limit: %v", err)
}
if (&AgentMeshNode{}).requireOwnIngressGrant("a2a", "assistant", api.IngressAuthModeStringEdge) == nil {
t.Fatal("node without a credential: nil, want an error")
}

err = newNode(withGrant, authorizerFactLimitRules()).requireOwnIngressGrant("a2a", "assistant", api.IngressAuthModeStringEdge)
if !errors.Is(err, datalog.ErrWorldRunLimitMaxFacts) {
t.Fatalf("over the facts limit = %v, want an error wrapping %v", err, datalog.ErrWorldRunLimitMaxFacts)
}
if stopped, transient := authorizerLimit(err); !stopped || transient {
t.Fatalf("authorizerLimit(%v) = (%v, %v), want (true, false)", err, stopped, transient)
}
}
3 changes: 1 addition & 2 deletions internal/router/ingress.go
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,6 @@ const (
ingressOutcomeInvalid = "invalid"
ingressOutcomeUnavailable = "unavailable"

maxIngressBodyBytes = 32 << 20
maxIngressCardBodyBytes = 1 << 20
a2aAgentCardSubpath = ".well-known/agent-card.json"
ingressDialTimeout = 10 * time.Second
Expand Down Expand Up @@ -303,7 +302,7 @@ func (r *Router) handleIngressRequest(w http.ResponseWriter, req *http.Request,
}

clientIP := clientIPFromRemoteAddr(req.RemoteAddr)
req.Body = http.MaxBytesReader(w, req.Body, maxIngressBodyBytes)
req.Body = http.MaxBytesReader(w, req.Body, api.MaxIngressBodyBytes)

if api.HasDotSegment(req.URL.Path) {
routerIngressRequestsTotal.WithLabelValues(ingressModeUnknown, ingressOutcomeInvalid).Inc()
Expand Down
2 changes: 1 addition & 1 deletion sdk/meshnet/meshnet_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -823,7 +823,7 @@ func TestCredentialLapseExponentialBackoffAndSignal(t *testing.T) {
})
mux.HandleFunc("/policies", func(w http.ResponseWriter, _ *http.Request) {
compiled, _ := api.BuildPolicyRules([]*api.PolicyRole{
{Name: api.RoleNode, AllowedTargets: []string{"*"}, AllowedServices: []string{"*"}, AllowedPublish: []string{"*"}},
{Name: api.RoleNode, AllowedTargets: []string{"*"}, AllowedServices: []string{"*"}, AllowedPublish: []string{"*"}, AllowedIngress: []*api.IngressGrant{{Services: []string{"*"}}}},
}, nil)
texts := make([]string, 0, len(compiled))
for _, r := range compiled {
Expand Down
7 changes: 7 additions & 0 deletions sdk/meshnet/statedir_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -617,6 +617,13 @@ func newFakeControlPlaneHTTPWithKeys(t *testing.T, pub ed25519.PublicKey, priv e
AllowedTargets: []string{"*"},
AllowedServices: []string{"*"},
AllowedPublish: []string{"*"},
// Edge-only ingress on every service, as agentmesh-one's
// development policy grants, plus the explicit passthrough
// grant one test's pass-agent needs; the node checks both.
AllowedIngress: []*api.IngressGrant{
{Services: []string{"*"}},
{Services: []string{"a2a://pass-agent"}, AllowedAuthModes: []api.IngressAuthMode{api.IngressAuthMode_INGRESS_AUTH_MODE_PASSTHROUGH}},
},
},
}, nil)
if len(warnings) > 0 {
Expand Down
15 changes: 12 additions & 3 deletions site/content/docs/guides/external-ingress.md
Original file line number Diff line number Diff line change
Expand Up @@ -193,14 +193,23 @@ serves internal mesh peers over `/libp2p-http`.
3. The router verifies the Biscuit signature, authority revocation ID, banned
peer status, `TaskAuthorizationRule` chain (including MCP `tools/call` tool
names), and `ServiceIngress.allow` locally **before** dialing any provider
node. Once admitted, both the router and the serving node (defense in depth
node, and checks that the provider's own credential grants
`allowed_ingress` for the service and auth mode under the current mesh
policy. Once admitted, both the router and the serving node (defense in depth
over `/mesh/ingress/1.0.0`) enforce the full intersection of mesh Datalog
RBAC, `TaskAuthorizationRule`, `ServiceIngress.allow`, and local
`agentmesh-node.yaml` `attenuation`. The node serves `/mesh/ingress/1.0.0`
`agentmesh-node.yaml` `attenuation`. The serving node repeats the
`allowed_ingress` check against its own credential and the mesh policy it
holds, so a router that forwards a service the node's role does not grant
gets `403` and a `deny` in
`agentmesh_node_external_ingress_requests_total`; an authorizer that stops
at a run limit before a verdict answers `504` (time) or `500` (facts or
iterations), counted as `undecided`. The node serves `/mesh/ingress/1.0.0`
only to a peer that authenticated as `mesh:role:router`: a stream from any
other peer, authenticated or not, is reset at the libp2p gate before the
HTTP server reads it and counted in
`agentmesh_node_streams_refused_total{scope="not_router"}`.
`agentmesh_node_streams_refused_total{scope="not_router"}`. Request bodies
are capped at 32 MiB on the router's public listener and again on the node.
4. The external `Authorization` header is stripped before reaching the local
backend (replaced only if `target_auth_path` is configured on the node).
5. For `a2a` services, unauthenticated `GET` requests to the service root or
Expand Down
Loading