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)