From 092a61b44ed9012481bad580b5d05af62f16ddf9 Mon Sep 17 00:00:00 2001 From: QuentinBisson Date: Thu, 24 Sep 2026 23:18:14 +0200 Subject: [PATCH] feat(substrate): the compiled ActorTemplate carries the revision's egress policy as its default ActorTemplateForRevision sets default_egress_policy from the revision's egress destinations and credentials, the same allowlist the AgentInstance workflow creates for each actor. Substrate copies it onto every actor it creates from the template, including the golden actor the template reconciler creates, which no client call could give a policy before it boots: without one the golden boot is denied all egress (403) and a workload that fetches skills or plugins before readyz never gets its snapshot. The policy builder moves to the substrate package so both callers share it; the workflow's EnsureActorEgressPolicy stays as the idempotent check. The revision digest carries an actorTemplateFormat marker so templates compiled before this change are re-created instead of conflicting. Depends on ActorTemplate.default_egress_policy in agent-substrate/substrate (agent-substrate/substrate#1324). Fixes #2604 Signed-off-by: QuentinBisson --- .../service/agentinstance/workflow.go | 50 +------------------ .../service/agentinstance/workflow_test.go | 10 ++-- go/core/internal/substrate/actor_template.go | 23 ++++++--- .../internal/substrate/actor_template_test.go | 46 ++++++++++++++++- go/core/internal/substrate/egress.go | 50 +++++++++++++++++++ go/core/internal/translator/revision.go | 10 ++++ 6 files changed, 127 insertions(+), 62 deletions(-) diff --git a/go/core/internal/service/agentinstance/workflow.go b/go/core/internal/service/agentinstance/workflow.go index 9915bb2770..e635b17b9f 100644 --- a/go/core/internal/service/agentinstance/workflow.go +++ b/go/core/internal/service/agentinstance/workflow.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "net/netip" "slices" "strings" "time" @@ -13,12 +12,10 @@ import ( "github.com/google/uuid" apiv1alpha1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1alpha1" "github.com/kagent-dev/kagent/go/core/internal/database" - "github.com/kagent-dev/kagent/go/core/internal/egress" "github.com/kagent-dev/kagent/go/core/internal/substrate" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" - "k8s.io/apimachinery/pkg/util/validation" ) type workflowStore interface { @@ -164,7 +161,7 @@ func (w *ActorWorkflow) run(ctx context.Context, instanceID string, requestedKin atespace, name := revision.ActorTemplateAtespace, substrate.ActorName(instance.Id) var policy *ateapipb.EgressPolicy if kind == apiv1alpha1.AgentInstanceOperation_AGENT_INSTANCE_OPERATION_CREATE { - policy, err = actorEgressPolicy(atespace, revision.EgressDestinations, revision.Credentials) + policy, err = substrate.ActorEgressPolicy(atespace, revision.EgressDestinations, revision.Credentials) if err != nil { return w.failPreparation(ctx, operation, fmt.Errorf("build Actor %s/%s egress policy: %w", atespace, name, err)) } @@ -304,48 +301,3 @@ func validActorIdentity(actor *ateapipb.Actor, revision *database.RuntimeRevisio return metadata.GetName() == name && metadata.GetAtespace() == revision.ActorTemplateAtespace && metadata.GetUid() != "" && ref.GetAtespace() == revision.ActorTemplateAtespace && ref.GetName() == revision.ActorTemplateName } - -// actorEgressPolicy compiles destinations into an actor's default allowlist. -// Credential bindings are already canonicalized by the store. -func actorEgressPolicy(atespace string, destinations []string, credentials []egress.Credential) (*ateapipb.EgressPolicy, error) { - var hostnames, cidrs []string - for _, destination := range destinations { - if ip, err := netip.ParseAddr(destination); err == nil && ip.Zone() == "" { - ip = ip.Unmap() - cidrs = append(cidrs, netip.PrefixFrom(ip, ip.BitLen()).String()) - continue - } - hostname := strings.TrimSuffix(strings.ToLower(destination), ".") - if len(validation.IsDNS1123Subdomain(hostname)) != 0 { - return nil, fmt.Errorf("invalid egress destination %q", destination) - } - hostnames = append(hostnames, hostname) - } - policy := &ateapipb.EgressPolicy{Metadata: &ateapipb.ResourceMetadata{Atespace: atespace, Name: "default"}} - for _, binding := range credentials { - if !slices.Contains(hostnames, binding.Hostname) { - return nil, fmt.Errorf("credential destination %q is not allowed", binding.Hostname) - } - var rule *ateapipb.HostnameRule - if len(policy.Rules) > 0 { - rule = policy.Rules[len(policy.Rules)-1].GetHostnames() - } - if rule == nil || rule.Patterns[0] != binding.Hostname { - rule = &ateapipb.HostnameRule{Patterns: []string{binding.Hostname}, Effects: &ateapipb.EgressRuleEffects{}} - policy.Rules = append(policy.Rules, &ateapipb.EgressRule{Hostnames: rule}) - } - rule.Effects.InjectStaticHeaders = append(rule.Effects.InjectStaticHeaders, &ateapipb.CredentialHeaderInjection{Header: binding.Header, Prefix: binding.Prefix, CredentialUri: binding.URI}) - } - if len(hostnames) > 0 { - slices.Sort(hostnames) - policy.Rules = append(policy.Rules, &ateapipb.EgressRule{Hostnames: &ateapipb.HostnameRule{Patterns: slices.Compact(hostnames)}}) - } - if len(cidrs) > 0 { - slices.Sort(cidrs) - policy.Rules = append(policy.Rules, &ateapipb.EgressRule{Cidrs: &ateapipb.CIDRRule{Cidrs: slices.Compact(cidrs)}}) - } - if len(policy.Rules) > 256 { - return nil, fmt.Errorf("egress policy exceeds 256 rules") - } - return policy, nil -} diff --git a/go/core/internal/service/agentinstance/workflow_test.go b/go/core/internal/service/agentinstance/workflow_test.go index 233f7c98d8..14fffe1513 100644 --- a/go/core/internal/service/agentinstance/workflow_test.go +++ b/go/core/internal/service/agentinstance/workflow_test.go @@ -318,18 +318,18 @@ func TestActorCreationRetainsEgressPolicyFailure(t *testing.T) { } func TestActorEgressPolicy(t *testing.T) { - policy, err := actorEgressPolicy("team-a", []string{"API.Example.com.", "api.example.com", "192.0.2.1", "2001:db8::1", "::ffff:192.0.2.1"}, nil) + policy, err := substrate.ActorEgressPolicy("team-a", []string{"API.Example.com.", "api.example.com", "192.0.2.1", "2001:db8::1", "::ffff:192.0.2.1"}, nil) require.NoError(t, err) require.Equal(t, &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "default"}, policy.Metadata) require.Len(t, policy.Rules, 2) require.Equal(t, []string{"api.example.com"}, policy.Rules[0].GetHostnames().GetPatterns()) require.Equal(t, []string{"192.0.2.1/32", "2001:db8::1/128"}, policy.Rules[1].GetCidrs().GetCidrs()) - policy, err = actorEgressPolicy("team-a", nil, nil) + policy, err = substrate.ActorEgressPolicy("team-a", nil, nil) require.NoError(t, err) require.Empty(t, policy.Rules, "no destinations must deny all egress") for _, destination := range []string{"", "*", "https://api.example.com", "api.example.com:443", "192.0.2.0/24", "fe80::1%eth0"} { t.Run(destination, func(t *testing.T) { - _, err := actorEgressPolicy("team-a", []string{destination}, nil) + _, err := substrate.ActorEgressPolicy("team-a", []string{destination}, nil) require.Error(t, err) }) } @@ -340,9 +340,9 @@ func TestActorEgressCredentialsRequireAllowedDestination(t *testing.T) { {Hostname: "api.example.com", Header: "authorization", Prefix: "Bearer ", URI: "ate-secret://kubernetes.io/team/auth/token"}, {Hostname: "api.example.com", Header: "x-api-key", URI: "ate-secret://kubernetes.io/team/auth/key"}, } - _, err := actorEgressPolicy("team", []string{"other.example.com"}, bindings) + _, err := substrate.ActorEgressPolicy("team", []string{"other.example.com"}, bindings) require.ErrorContains(t, err, "is not allowed") - policy, err := actorEgressPolicy("team", []string{"api.example.com", "other.example.com"}, bindings) + policy, err := substrate.ActorEgressPolicy("team", []string{"api.example.com", "other.example.com"}, bindings) require.NoError(t, err) require.Len(t, policy.Rules, 2) require.Equal(t, []string{"api.example.com"}, policy.Rules[0].GetHostnames().GetPatterns()) diff --git a/go/core/internal/substrate/actor_template.go b/go/core/internal/substrate/actor_template.go index 97e07eaf08..50bcf01d9d 100644 --- a/go/core/internal/substrate/actor_template.go +++ b/go/core/internal/substrate/actor_template.go @@ -75,6 +75,13 @@ func ActorTemplateForRevision(spec *translator.Revision, revisionID translator.R if len(actorEnv) > 32 { return nil, fmt.Errorf("runtime revision has %d environment variables; Substrate supports at most 32", len(actorEnv)) } + // The template carries the revision's allowlist so the golden actor, which + // Substrate creates without any client call, is admitted to the same + // destinations as the instances that boot from its snapshot. + egressPolicy, err := ActorEgressPolicy(spec.Namespace, spec.EgressDestinations, spec.Credentials) + if err != nil { + return nil, fmt.Errorf("compile egress policy: %w", err) + } template := &ateapipb.ActorTemplate{ Metadata: &ateapipb.ResourceMetadata{Atespace: spec.Namespace, Name: name}, @@ -95,7 +102,8 @@ func ActorTemplateForRevision(spec *translator.Revision, revisionID translator.R }, TimeoutSeconds: 30}, VolumeMounts: []*ateapipb.VolumeMount{{Name: durableDataVolume, MountPath: durableDataMount}, {Name: egressTrustVolume, MountPath: egressTrustMount}}, }}, - WorkerSelector: workerSelectorForPool(workerKey), + WorkerSelector: workerSelectorForPool(workerKey), + DefaultEgressPolicy: &ateapipb.EgressPolicyTemplate{Rules: egressPolicy.GetRules()}, SnapshotsConfig: &ateapipb.SnapshotsConfig{ StorageLocation: spec.SnapshotLocation, OnPause: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL, @@ -127,12 +135,13 @@ func actorTemplateSpec(template *ateapipb.ActorTemplate) *ateapipb.ActorTemplate Atespace: template.GetMetadata().GetAtespace(), Name: template.GetMetadata().GetName(), }, - WorkerSelector: template.GetWorkerSelector(), - Containers: template.GetContainers(), - Volumes: template.GetVolumes(), - SnapshotsConfig: template.GetSnapshotsConfig(), - SandboxConfig: template.GetSandboxConfig(), - Resources: template.GetResources(), + WorkerSelector: template.GetWorkerSelector(), + Containers: template.GetContainers(), + Volumes: template.GetVolumes(), + SnapshotsConfig: template.GetSnapshotsConfig(), + SandboxConfig: template.GetSandboxConfig(), + Resources: template.GetResources(), + DefaultEgressPolicy: template.GetDefaultEgressPolicy(), } } diff --git a/go/core/internal/substrate/actor_template_test.go b/go/core/internal/substrate/actor_template_test.go index 65211e1772..0239fe2f79 100644 --- a/go/core/internal/substrate/actor_template_test.go +++ b/go/core/internal/substrate/actor_template_test.go @@ -9,6 +9,7 @@ import ( a2apb "github.com/a2aproject/a2a-go/v2/a2apb/v1" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "github.com/kagent-dev/kagent/go/core/internal/egress" "github.com/kagent-dev/kagent/go/core/internal/translator" "google.golang.org/protobuf/proto" corev1 "k8s.io/api/core/v1" @@ -23,7 +24,9 @@ func TestActorTemplateForRevision(t *testing.T) { WorkerPoolName: "default", SnapshotLocation: "snapshots", ConfigJSON: []byte(`{"instruction":"help"}`), AgentCard: &a2apb.AgentCard{Name: "helper", Version: "v1", Capabilities: &a2apb.AgentCapabilities{Streaming: new(true)}, SupportedInterfaces: []*a2apb.AgentInterface{{Url: "http://127.0.0.1:80", ProtocolBinding: "GRPC", ProtocolVersion: "1.0"}}, DefaultInputModes: []string{"text"}, DefaultOutputModes: []string{"text"}}, - Environment: []corev1.EnvVar{{Name: "API_KEY", Value: translator.CredentialPlaceholder}}, + Environment: []corev1.EnvVar{{Name: "API_KEY", Value: translator.CredentialPlaceholder}}, + EgressDestinations: []string{"github.com", "api.anthropic.com"}, + Credentials: []egress.Credential{{Hostname: "api.anthropic.com", Header: "x-api-key", URI: "ate-secret://kubernetes.io/agents/anthropic/key"}}, } revisionID, err := spec.Digest() if err != nil { @@ -72,6 +75,42 @@ func TestActorTemplateForRevision(t *testing.T) { if environment["KAGENT_CONFIG_JSON"].Value != string(spec.ConfigJSON) { t.Fatal("config was not embedded as a non-secret literal") } + wantPolicy, err := ActorEgressPolicy("agents", spec.EgressDestinations, spec.Credentials) + if err != nil { + t.Fatal(err) + } + if !proto.Equal(&ateapipb.EgressPolicy{Rules: template.GetDefaultEgressPolicy().GetRules()}, &ateapipb.EgressPolicy{Rules: wantPolicy.GetRules()}) { + t.Fatalf("template default egress policy = %v, want %v", template.GetDefaultEgressPolicy(), wantPolicy) + } + if injected := template.GetDefaultEgressPolicy().GetRules()[0].GetHostnames().GetEffects().GetInjectStaticHeaders(); len(injected) != 1 || injected[0].GetHeader() != "x-api-key" { + t.Fatalf("credential injection missing from the template policy: %v", template.GetDefaultEgressPolicy()) + } +} + +func TestActorTemplateForRevisionDeniesAllEgressWithoutDestinations(t *testing.T) { + spec := &translator.Revision{ + Namespace: "agents", AgentTemplateName: "helper", HarnessName: "kagent", + Image: "agent.example/image@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + WorkerPoolName: "default", SnapshotLocation: "snapshots", + ConfigJSON: []byte(`{}`), AgentCard: &a2apb.AgentCard{Name: "helper", Version: "v1", Capabilities: &a2apb.AgentCapabilities{}, + SupportedInterfaces: []*a2apb.AgentInterface{{Url: "http://127.0.0.1:80", ProtocolBinding: "GRPC", ProtocolVersion: "1.0"}}, DefaultInputModes: []string{"text"}, DefaultOutputModes: []string{"text"}}, + } + revisionID, err := spec.Digest() + if err != nil { + t.Fatal(err) + } + template, err := ActorTemplateForRevision(spec, revisionID) + if err != nil { + t.Fatal(err) + } + if template.GetDefaultEgressPolicy() == nil || len(template.GetDefaultEgressPolicy().GetRules()) != 0 { + t.Fatalf("a revision without destinations must carry an empty (deny-all) policy, got %v", template.GetDefaultEgressPolicy()) + } + invalid := *spec + invalid.EgressDestinations = []string{"https://github.com"} + if _, err := ActorTemplateForRevision(&invalid, revisionID); err == nil { + t.Fatal("an invalid destination must fail template compilation") + } } func TestActorTemplateStampsTheRevisionOnTheResource(t *testing.T) { @@ -120,4 +159,9 @@ func TestActorTemplateSpecEqualIgnoresServerFields(t *testing.T) { if ActorTemplateSpecEqual(left, right) { t.Fatal("different container image was accepted") } + right = proto.CloneOf(left) + right.DefaultEgressPolicy = &ateapipb.EgressPolicyTemplate{Rules: []*ateapipb.EgressRule{{Hostnames: &ateapipb.HostnameRule{Patterns: []string{"github.com"}}}}} + if ActorTemplateSpecEqual(left, right) { + t.Fatal("different default egress policy was accepted") + } } diff --git a/go/core/internal/substrate/egress.go b/go/core/internal/substrate/egress.go index dd7d0be647..d1b0205dee 100644 --- a/go/core/internal/substrate/egress.go +++ b/go/core/internal/substrate/egress.go @@ -3,11 +3,16 @@ package substrate import ( "context" "fmt" + "net/netip" + "slices" + "strings" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "github.com/kagent-dev/kagent/go/core/internal/egress" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" + "k8s.io/apimachinery/pkg/util/validation" ) // EnsureActorEgressPolicy can be retried after a lost create response. A prepared @@ -32,3 +37,48 @@ func (c *Client) EnsureActorEgressPolicy(ctx context.Context, atespace, name str } return nil } + +// ActorEgressPolicy compiles destinations into an actor's default allowlist. +// Credential bindings are already canonicalized by the store. +func ActorEgressPolicy(atespace string, destinations []string, credentials []egress.Credential) (*ateapipb.EgressPolicy, error) { + var hostnames, cidrs []string + for _, destination := range destinations { + if ip, err := netip.ParseAddr(destination); err == nil && ip.Zone() == "" { + ip = ip.Unmap() + cidrs = append(cidrs, netip.PrefixFrom(ip, ip.BitLen()).String()) + continue + } + hostname := strings.TrimSuffix(strings.ToLower(destination), ".") + if len(validation.IsDNS1123Subdomain(hostname)) != 0 { + return nil, fmt.Errorf("invalid egress destination %q", destination) + } + hostnames = append(hostnames, hostname) + } + policy := &ateapipb.EgressPolicy{Metadata: &ateapipb.ResourceMetadata{Atespace: atespace, Name: "default"}} + for _, binding := range credentials { + if !slices.Contains(hostnames, binding.Hostname) { + return nil, fmt.Errorf("credential destination %q is not allowed", binding.Hostname) + } + var rule *ateapipb.HostnameRule + if len(policy.Rules) > 0 { + rule = policy.Rules[len(policy.Rules)-1].GetHostnames() + } + if rule == nil || rule.Patterns[0] != binding.Hostname { + rule = &ateapipb.HostnameRule{Patterns: []string{binding.Hostname}, Effects: &ateapipb.EgressRuleEffects{}} + policy.Rules = append(policy.Rules, &ateapipb.EgressRule{Hostnames: rule}) + } + rule.Effects.InjectStaticHeaders = append(rule.Effects.InjectStaticHeaders, &ateapipb.CredentialHeaderInjection{Header: binding.Header, Prefix: binding.Prefix, CredentialUri: binding.URI}) + } + if len(hostnames) > 0 { + slices.Sort(hostnames) + policy.Rules = append(policy.Rules, &ateapipb.EgressRule{Hostnames: &ateapipb.HostnameRule{Patterns: slices.Compact(hostnames)}}) + } + if len(cidrs) > 0 { + slices.Sort(cidrs) + policy.Rules = append(policy.Rules, &ateapipb.EgressRule{Cidrs: &ateapipb.CIDRRule{Cidrs: slices.Compact(cidrs)}}) + } + if len(policy.Rules) > 256 { + return nil, fmt.Errorf("egress policy exceeds 256 rules") + } + return policy, nil +} diff --git a/go/core/internal/translator/revision.go b/go/core/internal/translator/revision.go index b55c981372..d13dc35de8 100644 --- a/go/core/internal/translator/revision.go +++ b/go/core/internal/translator/revision.go @@ -65,6 +65,11 @@ type Revision struct { EgressDestinations []string } +// actorTemplateFormat is bumped when ActorTemplateForRevision changes what it +// renders from unchanged revision inputs. 2 = default_egress_policy on the +// template. +const actorTemplateFormat = 2 + // Digest returns the immutable identity of every input that affects runtime // behavior. The full digest is the database key; Kubernetes names use a short // prefix only for readability. @@ -83,11 +88,16 @@ func (r *Revision) Digest() (RevisionID, error) { Provenance json.RawMessage `json:"provenance"` Credentials []egress.Credential `json:"credentials,omitempty"` EgressDestinations []string `json:"egressDestinations"` + // ActorTemplateFormat changes when the ActorTemplate compiled from the + // same inputs changes shape, so the immutable template is re-created + // under a new name instead of conflicting with the stored one. + ActorTemplateFormat int `json:"actorTemplateFormat"` }{ Namespace: r.Namespace, AgentTemplateName: r.AgentTemplateName, HarnessName: r.HarnessName, Image: r.Image, Command: r.Command, Args: r.Args, Environment: r.Environment, ConfigJSON: r.ConfigJSON, WorkerPoolName: r.WorkerPoolName, SnapshotLocation: r.SnapshotLocation, Provenance: r.Provenance, Credentials: r.Credentials, EgressDestinations: r.EgressDestinations, + ActorTemplateFormat: actorTemplateFormat, }) if err != nil { return RevisionID{}, fmt.Errorf("marshal runtime revision inputs: %w", err)