Merge branch 'main' into fix-postgres-opensearch-setup

Author: chaptersix

.github/workflows/extensibility.yaml

@@ -26,7 +26,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v6
         with:
-          go-version: '1.22'
-      - name: build
+          go-version-file: extensibility/go.mod
+      - name: build and test
         working-directory: extensibility
-        run: go build ./...
+        run: go test -count=1 -timeout 120s ./...

compose/.env

@@ -2,8 +2,8 @@ COMPOSE_PROJECT_NAME=temporal
 CASSANDRA_VERSION=3.11.9
 ELASTICSEARCH_VERSION=7.17.27
 MYSQL_VERSION=8
-TEMPORAL_VERSION=1.29.1
-TEMPORAL_ADMINTOOLS_VERSION=1.29.1-tctl-1.18.4-cli-1.5.0
+TEMPORAL_VERSION=1.30.1
+TEMPORAL_ADMINTOOLS_VERSION=1.30.1
 TEMPORAL_UI_VERSION=2.34.0
 POSTGRESQL_VERSION=16
 POSTGRES_PASSWORD=temporal

compose/scripts/validate-temporal.sh

@@ -1,4 +1,4 @@
-#!/bin/bash
+#!/bin/sh
 
 set -e
 

extensibility/authorizer/myAuthorizer_test.go

@@ -0,0 +1,69 @@
+package authorizer
+
+import (
+	"context"
+	"testing"
+
+	"go.temporal.io/server/common/authorization"
+)
+
+func TestMyAuthorizer(t *testing.T) {
+	a := NewMyAuthorizer()
+	ctx := context.Background()
+
+	tests := []struct {
+		name     string
+		claims   *authorization.Claims
+		target   *authorization.CallTarget
+		decision authorization.Decision
+	}{
+		{
+			name:     "allow temporal-system namespace",
+			claims:   nil,
+			target:   &authorization.CallTarget{Namespace: "temporal-system", APIName: "UpdateNamespace"},
+			decision: authorization.DecisionAllow,
+		},
+		{
+			name:     "allow system admin",
+			claims:   &authorization.Claims{System: authorization.RoleAdmin},
+			target:   &authorization.CallTarget{Namespace: "test", APIName: "UpdateNamespace"},
+			decision: authorization.DecisionAllow,
+		},
+		{
+			name:     "deny UpdateNamespace without claims",
+			claims:   nil,
+			target:   &authorization.CallTarget{Namespace: "test", APIName: "UpdateNamespace"},
+			decision: authorization.DecisionDeny,
+		},
+		{
+			name:     "deny UpdateNamespace with reader role",
+			claims:   &authorization.Claims{Namespaces: map[string]authorization.Role{"test": authorization.RoleReader}},
+			target:   &authorization.CallTarget{Namespace: "test", APIName: "UpdateNamespace"},
+			decision: authorization.DecisionDeny,
+		},
+		{
+			name:     "allow UpdateNamespace with namespace writer role",
+			claims:   &authorization.Claims{Namespaces: map[string]authorization.Role{"test": authorization.RoleWriter}},
+			target:   &authorization.CallTarget{Namespace: "test", APIName: "UpdateNamespace"},
+			decision: authorization.DecisionAllow,
+		},
+		{
+			name:     "allow other calls without claims",
+			claims:   nil,
+			target:   &authorization.CallTarget{Namespace: "test", APIName: "ListNamespaces"},
+			decision: authorization.DecisionAllow,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := a.Authorize(ctx, tt.claims, tt.target)
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if result.Decision != tt.decision {
+				t.Fatalf("expected %v, got %v", tt.decision, result.Decision)
+			}
+		})
+	}
+}

extensibility/authorizer/server/main.go

@@ -34,22 +34,28 @@ import (
 	"github.com/temporalio/service-samples/authorizer"
 )
 
-func main() {
-
-	cfg, err := config.LoadConfig("development", "./config", "")
+func newServer(configFile string, opts ...temporal.ServerOption) (temporal.Server, error) {
+	cfg, err := config.Load(config.WithConfigFile(configFile))
 	if err != nil {
-		log.Fatal(err)
+		return nil, err
 	}
 
-	s, err := temporal.NewServer(
+	defaults := []temporal.ServerOption{
 		temporal.ForServices(temporal.DefaultServices),
 		temporal.WithConfig(cfg),
-		temporal.InterruptOn(temporal.InterruptCh()),
 		temporal.WithClaimMapper(func(cfg *config.Config) authorization.ClaimMapper {
 			return authorizer.NewMyClaimMapper(cfg)
 		}),
 		temporal.WithAuthorizer(authorizer.NewMyAuthorizer()),
-	)
+	}
+
+	return temporal.NewServer(append(defaults, opts...)...)
+}
+
+func main() {
+	// InterruptOn is passed here rather than in newServer so tests can call s.Stop() directly.
+	// Include this in production scenarios to enable graceful shutdown on SIGINT/SIGTERM.
+	s, err := newServer("./config/development.yaml", temporal.InterruptOn(temporal.InterruptCh()))
 	if err != nil {
 		log.Fatal(err)
 	}

extensibility/authorizer/server/main_test.go

@@ -0,0 +1,93 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"go.temporal.io/api/operatorservice/v1"
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/server/common/config"
+	_ "go.temporal.io/server/common/persistence/sql/sqlplugin/sqlite"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/durationpb"
+)
+
+func TestAuthorizerDeniesUpdateNamespace(t *testing.T) {
+	s, err := newServer("testdata/config.yaml")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := s.Start(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = s.Stop() })
+
+	cfg, err := config.Load(config.WithConfigFile("testdata/config.yaml"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	frontendAddr := fmt.Sprintf("127.0.0.1:%d", cfg.Services["frontend"].RPC.GRPCPort)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	conn, err := grpc.NewClient(frontendAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	wfClient := workflowservice.NewWorkflowServiceClient(conn)
+	opClient := operatorservice.NewOperatorServiceClient(conn)
+
+	// Wait for server to be ready
+	for {
+		_, err := wfClient.GetSystemInfo(ctx, &workflowservice.GetSystemInfoRequest{})
+		if err == nil {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			t.Fatal("timed out waiting for server")
+		case <-time.After(200 * time.Millisecond):
+		}
+	}
+
+	// Create a namespace (should succeed)
+	ns := "test-authorizer"
+	_, err = wfClient.RegisterNamespace(ctx, &workflowservice.RegisterNamespaceRequest{
+		Namespace:                        ns,
+		WorkflowExecutionRetentionPeriod: durationpb.New(24 * time.Hour),
+	})
+	if err != nil {
+		t.Fatalf("RegisterNamespace should succeed: %v", err)
+	}
+
+	// Wait for namespace to propagate
+	time.Sleep(2 * time.Second)
+
+	// UpdateNamespace should be denied by the authorizer
+	_, err = wfClient.UpdateNamespace(ctx, &workflowservice.UpdateNamespaceRequest{
+		Namespace: ns,
+	})
+	if err == nil {
+		t.Fatal("UpdateNamespace should have been denied")
+	}
+	if s, ok := status.FromError(err); !ok || s.Code() != codes.PermissionDenied {
+		t.Fatalf("expected PermissionDenied, got: %v", err)
+	}
+
+	// DeleteNamespace should still be allowed
+	_, err = opClient.DeleteNamespace(ctx, &operatorservice.DeleteNamespaceRequest{
+		Namespace: ns,
+	})
+	if err != nil {
+		t.Fatalf("DeleteNamespace should succeed: %v", err)
+	}
+}

extensibility/authorizer/server/testdata/config.yaml

@@ -0,0 +1,82 @@
+log:
+  stdout: true
+  level: error
+
+persistence:
+  defaultStore: sqlite-default
+  visibilityStore: sqlite-visibility
+  numHistoryShards: 1
+  datastores:
+    sqlite-default:
+      sql:
+        pluginName: "sqlite"
+        databaseName: "default"
+        connectAddr: "localhost"
+        connectProtocol: "tcp"
+        connectAttributes:
+          mode: "memory"
+          cache: "private"
+        maxConns: 1
+        maxIdleConns: 1
+        maxConnLifetime: "1h"
+    sqlite-visibility:
+      sql:
+        pluginName: "sqlite"
+        databaseName: "default"
+        connectAddr: "localhost"
+        connectProtocol: "tcp"
+        connectAttributes:
+          mode: "memory"
+          cache: "private"
+        maxConns: 1
+        maxIdleConns: 1
+        maxConnLifetime: "1h"
+
+global:
+  membership:
+    maxJoinDuration: 30s
+    broadcastAddress: "127.0.0.1"
+  metrics:
+    prometheus:
+      framework: "tally"
+      listenAddress: ":0"
+
+services:
+  frontend:
+    rpc:
+      grpcPort: 17233
+      membershipPort: 17234
+      bindOnLocalHost: true
+  matching:
+    rpc:
+      grpcPort: 17235
+      membershipPort: 17236
+      bindOnLocalHost: true
+  history:
+    rpc:
+      grpcPort: 17237
+      membershipPort: 17238
+      bindOnLocalHost: true
+  worker:
+    rpc:
+      grpcPort: 17239
+      membershipPort: 17240
+      bindOnLocalHost: true
+
+clusterMetadata:
+  enableGlobalNamespace: false
+  failoverVersionIncrement: 10
+  masterClusterName: "active"
+  currentClusterName: "active"
+  clusterInformation:
+    active:
+      enabled: true
+      initialFailoverVersion: 1
+      rpcName: "frontend"
+      rpcAddress: "localhost:17233"
+
+dcRedirectionPolicy:
+  policy: "noop"
+
+publicClient:
+  hostPort: "localhost:17233"

extensibility/config/development.yaml

@@ -1,3 +1,7 @@
+log:
+  stdout: true
+  level: info
+
 persistence:
   defaultStore: cass-default
   visibilityStore: es-visibility