feat(runners): internal read variants

.github/workflows/e2e.yml

@@ -9,6 +9,13 @@ concurrency:
   group: e2e-${{ github.event_name == 'pull_request' && github.event.pull_request.number || github.sha }}
   cancel-in-progress: true
 
+env:
+  AGN_INIT_IMAGE: ghcr.io/agynio/agent-init-agn:0.4
+  CODEX_INIT_IMAGE: ghcr.io/agynio/agent-init-codex:0.13
+  CLAUDE_INIT_IMAGE: ghcr.io/agynio/agent-init-claude:0.1
+  AGN_EXPOSE_INIT_IMAGE: ghcr.io/agynio/agent-init-agn:0.4
+  AGYN_AGENT_INIT_IMAGE: ghcr.io/agynio/agent-init:v1.0.0
+
 jobs:
   e2e:
     runs-on: ubuntu-latest
@@ -46,5 +53,11 @@ jobs:
 
       - name: Run E2E tests
         uses: agynio/e2e/.github/actions/run-tests@main
+        env:
+          AGN_INIT_IMAGE: ${{ env.AGN_INIT_IMAGE }}
+          CODEX_INIT_IMAGE: ${{ env.CODEX_INIT_IMAGE }}
+          CLAUDE_INIT_IMAGE: ${{ env.CLAUDE_INIT_IMAGE }}
+          AGN_EXPOSE_INIT_IMAGE: ${{ env.AGN_EXPOSE_INIT_IMAGE }}
+          AGYN_AGENT_INIT_IMAGE: ${{ env.AGYN_AGENT_INIT_IMAGE }}
         with:
           service: runners

README.md

@@ -38,3 +38,11 @@ devspace run test-e2e --tag svc_runners
 
 E2E coverage is centralized in [agynio/e2e](https://github.com/agynio/e2e) under the go-core suite.
 See [E2E Testing](https://github.com/agynio/architecture/blob/main/architecture/operations/e2e-testing.md).
+
+## Helm chart defaults
+
+The chart ships with a DENY-based Istio AuthorizationPolicy. By default,
+`authorizationPolicy.identityServiceAccounts` allows in-mesh callers that
+forward `x-identity-id` (`gateway`, `expose`, `notifications`, `chat`).
+Override the list if your deployment uses different service accounts (for
+example, add `agents-orchestrator-e2e` for E2E runs).

charts/runners/templates/authorizationpolicy.yaml

@@ -0,0 +1,61 @@
+{{- if .Values.authorizationPolicy.enabled }}
+{{- $orchestratorNamespace := .Values.authorizationPolicy.orchestratorServiceAccount.namespace | default .Release.Namespace }}
+{{- $orchestratorName := .Values.authorizationPolicy.orchestratorServiceAccount.name }}
+{{- $orchestratorPrincipal := printf "cluster.local/ns/%s/sa/%s" $orchestratorNamespace $orchestratorName }}
+{{- $identityServiceAccounts := .Values.authorizationPolicy.identityServiceAccounts | default list }}
+{{- $identityPrincipals := list $orchestratorPrincipal }}
+{{- range $identityServiceAccounts }}
+{{- $identityNamespace := .namespace | default $.Release.Namespace }}
+{{- $identityPrincipals = append $identityPrincipals (printf "cluster.local/ns/%s/sa/%s" $identityNamespace .name) }}
+{{- end }}
+{{- $gatewayPaths := list "/agynio.api.runners.v1.RunnersService/RegisterRunner" "/agynio.api.runners.v1.RunnersService/GetRunner" "/agynio.api.runners.v1.RunnersService/ListRunners" "/agynio.api.runners.v1.RunnersService/UpdateRunner" "/agynio.api.runners.v1.RunnersService/DeleteRunner" "/agynio.api.runners.v1.RunnersService/GetWorkload" "/agynio.api.runners.v1.RunnersService/ListWorkloadsByThread" "/agynio.api.runners.v1.RunnersService/ListWorkloads" "/agynio.api.runners.v1.RunnersService/TouchWorkload" "/agynio.api.runners.v1.RunnersService/StreamWorkloadLogs" "/agynio.api.runners.v1.RunnersService/GetVolume" "/agynio.api.runners.v1.RunnersService/ListVolumes" "/agynio.api.runners.v1.RunnersService/ListVolumesByThread" }}
+{{- $orchestratorPaths := list "/agynio.api.runners.v1.RunnersService/CreateWorkload" "/agynio.api.runners.v1.RunnersService/UpdateWorkload" "/agynio.api.runners.v1.RunnersService/UpdateWorkloadStatus" "/agynio.api.runners.v1.RunnersService/DeleteWorkload" "/agynio.api.runners.v1.RunnersService/BatchUpdateWorkloadSampledAt" "/agynio.api.runners.v1.RunnersService/CreateVolume" "/agynio.api.runners.v1.RunnersService/UpdateVolume" "/agynio.api.runners.v1.RunnersService/BatchUpdateVolumeSampledAt" }}
+apiVersion: security.istio.io/v1beta1
+kind: AuthorizationPolicy
+metadata:
+  name: {{ include "service-base.fullname" . }}-internal
+  labels:
+    {{- include "service-base.labels" . | nindent 4 }}
+spec:
+  selector:
+    matchLabels:
+      {{- include "service-base.selectorLabels" . | nindent 6 }}
+  action: DENY
+  rules:
+    - from:
+        - source:
+            notPrincipals:
+              - {{ $orchestratorPrincipal | quote }}
+      to:
+        - operation:
+            paths:
+              {{ range $orchestratorPaths }}
+              - {{ . | quote }}
+              {{ end }}
+    - from:
+        - source:
+            notPrincipals:
+              {{ range $identityPrincipals }}
+              - {{ . | quote }}
+              {{ end }}
+      to:
+        - operation:
+            paths:
+              {{ range $gatewayPaths }}
+              - {{ . | quote }}
+              {{ end }}
+    - from:
+        - source:
+            notPrincipals:
+              - {{ $orchestratorPrincipal | quote }}
+      when:
+        - key: request.headers[x-identity-id]
+          notValues:
+            - "*"
+      to:
+        - operation:
+            paths:
+              {{ range $gatewayPaths }}
+              - {{ . | quote }}
+              {{ end }}
+{{- end }}

charts/runners/values.yaml

@@ -25,6 +25,21 @@ serviceAccount:
 
 automountServiceAccountToken: true
 
+authorizationPolicy:
+  enabled: true
+  orchestratorServiceAccount:
+    name: agents-orchestrator
+    namespace: ""
+  identityServiceAccounts:
+    - name: gateway
+      namespace: ""
+    - name: expose
+      namespace: ""
+    - name: notifications
+      namespace: ""
+    - name: chat
+      namespace: ""
+
 rbac:
   create: false
   clusterWide: false

internal/server/authz.go

@@ -18,11 +18,30 @@ func identityFromMetadata(ctx context.Context) (uuid.UUID, error) {
 	}
 	values := md.Get(identityMetadata)
 	if len(values) != 1 {
-		return uuid.UUID{}, fmt.Errorf("expected single value")
+		return uuid.UUID{}, fmt.Errorf("metadata %s: expected single value, got %d", identityMetadata, len(values))
 	}
 	return parseUUID(values[0])
 }
 
+func identityFromMetadataOptional(ctx context.Context) (*uuid.UUID, error) {
+	md, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		return nil, nil
+	}
+	values := md.Get(identityMetadata)
+	if len(values) == 0 {
+		return nil, nil
+	}
+	if len(values) != 1 {
+		return nil, fmt.Errorf("metadata %s: expected single value, got %d", identityMetadata, len(values))
+	}
+	parsed, err := parseUUID(values[0])
+	if err != nil {
+		return nil, err
+	}
+	return &parsed, nil
+}
+
 func (s *Server) requireClusterAdmin(ctx context.Context, identityID uuid.UUID) error {
 	return s.requireRelation(ctx, identityID, clusterAdminRelation, clusterObject)
 }

internal/server/runners.go

@@ -177,7 +177,7 @@ func (s *Server) RegisterRunner(ctx context.Context, req *runnersv1.RegisterRunn
 }
 
 func (s *Server) GetRunner(ctx context.Context, req *runnersv1.GetRunnerRequest) (*runnersv1.GetRunnerResponse, error) {
-	callerID, err := identityFromMetadata(ctx)
+	callerID, err := identityFromMetadataOptional(ctx)
 	if err != nil {
 		return nil, status.Errorf(codes.Unauthenticated, "unauthenticated: %v", err)
 	}
@@ -189,8 +189,8 @@ func (s *Server) GetRunner(ctx context.Context, req *runnersv1.GetRunnerRequest)
 	if err != nil {
 		return nil, toStatusError(err)
 	}
-	if runner.OrganizationID != nil {
-		if err := s.requireOrgMember(ctx, callerID, *runner.OrganizationID); err != nil {
+	if callerID != nil && runner.OrganizationID != nil {
+		if err := s.requireOrgMember(ctx, *callerID, *runner.OrganizationID); err != nil {
 			return nil, err
 		}
 	}
@@ -259,7 +259,7 @@ func (s *Server) UpdateRunner(ctx context.Context, req *runnersv1.UpdateRunnerRe
 }
 
 func (s *Server) ListRunners(ctx context.Context, req *runnersv1.ListRunnersRequest) (*runnersv1.ListRunnersResponse, error) {
-	callerID, err := identityFromMetadata(ctx)
+	callerID, err := identityFromMetadataOptional(ctx)
 	if err != nil {
 		return nil, status.Errorf(codes.Unauthenticated, "unauthenticated: %v", err)
 	}
@@ -271,8 +271,8 @@ func (s *Server) ListRunners(ctx context.Context, req *runnersv1.ListRunnersRequ
 		}
 		organizationID = &parsed
 	}
-	if organizationID != nil {
-		if err := s.requireOrgMember(ctx, callerID, *organizationID); err != nil {
+	if callerID != nil && organizationID != nil {
+		if err := s.requireOrgMember(ctx, *callerID, *organizationID); err != nil {
 			return nil, err
 		}
 	}
@@ -286,15 +286,15 @@ func (s *Server) ListRunners(ctx context.Context, req *runnersv1.ListRunnersRequ
 		return nil, status.Errorf(codes.Internal, "list runners: %v", err)
 	}
 
-	if organizationID == nil {
+	if callerID != nil && organizationID == nil {
 		memberCache := map[uuid.UUID]bool{}
 		filtered := make([]runnerRecord, 0, len(runners))
 		for _, runner := range runners {
 			if runner.OrganizationID == nil {
 				filtered = append(filtered, runner)
 				continue
 			}
-			allowed, err := s.memberAllowed(ctx, callerID, *runner.OrganizationID, memberCache)
+			allowed, err := s.memberAllowed(ctx, *callerID, *runner.OrganizationID, memberCache)
 			if err != nil {
 				return nil, err
 			}

internal/server/runners_test.go

@@ -139,6 +139,7 @@ func (f fakeIdentityClient) ResolveNickname(ctx context.Context, req *identityv1
 func (f fakeIdentityClient) BatchGetNicknames(ctx context.Context, req *identityv1.BatchGetNicknamesRequest, opts ...grpc.CallOption) (*identityv1.BatchGetNicknamesResponse, error) {
 	return nil, status.Error(codes.Unimplemented, "not implemented")
 }
+
 type fakeAgentsClient struct {
 	getAgent              func(ctx context.Context, req *agentsv1.GetAgentRequest) (*agentsv1.GetAgentResponse, error)
 	getVolume             func(ctx context.Context, req *agentsv1.GetVolumeRequest) (*agentsv1.GetVolumeResponse, error)
@@ -189,6 +190,7 @@ func (f fakeAgentsClient) GetHook(ctx context.Context, req *agentsv1.GetHookRequ
 	}
 	return f.getHook(ctx, req)
 }
+
 type fakeAuthorizationClient struct {
 	check func(ctx context.Context, req *authorizationv1.CheckRequest) (*authorizationv1.CheckResponse, error)
 	write func(ctx context.Context, req *authorizationv1.WriteRequest) (*authorizationv1.WriteResponse, error)
@@ -491,6 +493,52 @@ func TestGetRunnerRequiresMember(t *testing.T) {
 	}
 }
 
+func TestGetRunnerInternalNoIdentity(t *testing.T) {
+	mockPool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to create mock pool: %v", err)
+	}
+
+	runnerID := uuid.New()
+	organizationID := uuid.New()
+	identityID := uuid.New()
+	now := time.Now().UTC()
+	labelsJSON := []byte("{}")
+	capabilitiesJSON := []byte("[]")
+	rows := pgxmock.NewRows([]string{"id", "name", "organization_id", "identity_id", "ziti_identity_id", "ziti_service_id", "ziti_service_name", "status", "labels", "capabilities", "created_at", "updated_at"}).
+		AddRow(runnerID, "runner-1", pgtype.UUID{Bytes: organizationID, Valid: true}, identityID, "", "service-id", "runner-service", runnerStatusOffline, labelsJSON, capabilitiesJSON, now, now)
+
+	query := fmt.Sprintf(`SELECT %s FROM runners WHERE id = $1`, runnerColumns)
+	mockPool.ExpectQuery(regexp.QuoteMeta(query)).WithArgs(runnerID).WillReturnRows(rows)
+
+	checkCalls := 0
+	authorizationClient := fakeAuthorizationClient{
+		check: func(ctx context.Context, req *authorizationv1.CheckRequest) (*authorizationv1.CheckResponse, error) {
+			checkCalls++
+			return &authorizationv1.CheckResponse{Allowed: false}, nil
+		},
+	}
+
+	srv := New(Options{Pool: mockPool, AuthorizationClient: authorizationClient})
+	resp, err := srv.GetRunner(context.Background(), &runnersv1.GetRunnerRequest{Id: runnerID.String()})
+	if err != nil {
+		t.Fatalf("GetRunner failed: %v", err)
+	}
+	if resp.GetRunner() == nil {
+		t.Fatal("expected runner in response")
+	}
+	if resp.GetRunner().GetOrganizationId() != organizationID.String() {
+		t.Fatalf("expected organization id %q, got %q", organizationID.String(), resp.GetRunner().GetOrganizationId())
+	}
+	if checkCalls != 0 {
+		t.Fatalf("expected no authorization checks, got %d", checkCalls)
+	}
+
+	if err := mockPool.ExpectationsWereMet(); err != nil {
+		t.Fatalf("unmet expectations: %v", err)
+	}
+}
+
 func TestListRunnersRequiresMember(t *testing.T) {
 	mockPool, err := pgxmock.NewPool()
 	if err != nil {
@@ -519,6 +567,53 @@ func TestListRunnersRequiresMember(t *testing.T) {
 	}
 }
 
+func TestListRunnersInternalNoIdentity(t *testing.T) {
+	mockPool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to create mock pool: %v", err)
+	}
+
+	runnerID := uuid.New()
+	organizationID := uuid.New()
+	identityID := uuid.New()
+	now := time.Now().UTC()
+	labelsJSON := []byte("{}")
+	capabilitiesJSON := []byte("[]")
+	rows := pgxmock.NewRows([]string{"id", "name", "organization_id", "identity_id", "ziti_identity_id", "ziti_service_id", "ziti_service_name", "status", "labels", "capabilities", "created_at", "updated_at"}).
+		AddRow(runnerID, "runner-1", pgtype.UUID{Bytes: organizationID, Valid: true}, identityID, "", "service-id", "runner-service", runnerStatusOffline, labelsJSON, capabilitiesJSON, now, now)
+
+	limit := normalizePageSize(0)
+	query := fmt.Sprintf("SELECT %s FROM runners ORDER BY id ASC LIMIT $1", runnerColumns)
+	mockPool.ExpectQuery(regexp.QuoteMeta(query)).WithArgs(int(limit) + 1).WillReturnRows(rows)
+
+	checkCalls := 0
+	authorizationClient := fakeAuthorizationClient{
+		check: func(ctx context.Context, req *authorizationv1.CheckRequest) (*authorizationv1.CheckResponse, error) {
+			checkCalls++
+			return &authorizationv1.CheckResponse{Allowed: false}, nil
+		},
+	}
+
+	srv := New(Options{Pool: mockPool, AuthorizationClient: authorizationClient})
+	resp, err := srv.ListRunners(context.Background(), &runnersv1.ListRunnersRequest{})
+	if err != nil {
+		t.Fatalf("ListRunners failed: %v", err)
+	}
+	if len(resp.GetRunners()) != 1 {
+		t.Fatalf("expected 1 runner, got %d", len(resp.GetRunners()))
+	}
+	if resp.GetRunners()[0].GetOrganizationId() != organizationID.String() {
+		t.Fatalf("expected organization id %q, got %q", organizationID.String(), resp.GetRunners()[0].GetOrganizationId())
+	}
+	if checkCalls != 0 {
+		t.Fatalf("expected no authorization checks, got %d", checkCalls)
+	}
+
+	if err := mockPool.ExpectationsWereMet(); err != nil {
+		t.Fatalf("unmet expectations: %v", err)
+	}
+}
+
 func TestUpdateRunnerUpdatesLabels(t *testing.T) {
 	mockPool, err := pgxmock.NewPool()
 	if err != nil {