test 4 milestones

internal/api/grpc/session/v2/session_integration_test.go

@@ -36,7 +36,7 @@ func TestMain(m *testing.M) {
 		defer Tester.Done()
 		Client = Tester.Client.SessionV2
 
-		CTX, _ = Tester.WithSystemAuthorization(ctx, integration.OrgOwner), errCtx
+		CTX, _ = Tester.WithAuthorization(ctx, integration.OrgOwner), errCtx
 		User = Tester.CreateHumanUser(CTX)
 		Tester.RegisterUserPasskey(CTX, User.GetUserId())
 		return m.Run()

internal/api/grpc/user/v2/user_integration_test.go

@@ -38,7 +38,7 @@ func TestMain(m *testing.M) {
 		Tester = integration.NewTester(ctx)
 		defer Tester.Done()
 
-		CTX, ErrCTX = Tester.WithSystemAuthorization(ctx, integration.OrgOwner), errCtx
+		CTX, ErrCTX = Tester.WithAuthorization(ctx, integration.OrgOwner), errCtx
 		Client = Tester.Client.UserV2
 		return m.Run()
 	}())

internal/integration/client.go

@@ -3,6 +3,7 @@ package integration
 import (
 	"context"
 	"fmt"
+	"github.com/zitadel/zitadel/pkg/grpc/system"
 	"testing"
 	"time"
 
@@ -29,6 +30,7 @@ type Client struct {
 	Mgmt      mgmt.ManagementServiceClient
 	UserV2    user.UserServiceClient
 	SessionV2 session.SessionServiceClient
+	System    system.SystemServiceClient
 }
 
 func newClient(cc *grpc.ClientConn) Client {
@@ -38,6 +40,7 @@ func newClient(cc *grpc.ClientConn) Client {
 		Mgmt:      mgmt.NewManagementServiceClient(cc),
 		UserV2:    user.NewUserServiceClient(cc),
 		SessionV2: session.NewSessionServiceClient(cc),
+		System:    system.NewSystemServiceClient(cc),
 	}
 }
 

internal/integration/config/system-user-key.pem

@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEowIBAAKCAQEAzi+FFSJL7f5yw4KTwzgMP34ePGycm/M+kT0M7V4Cgx5V3EaD
+IvTQKTLfBaEB45zb9LtjIXzDw0rXRoS2hO6th+CYQCz3KCvh09C0IzxZiB2IS3H/
+aT+5Bx9EFY+vnAkZjccbyG5YNRvmtOlnvIeIH7qZ0tEwkPfF5GEZNPJPtmy3UGV7
+iofdVQS1xRj73+aMw5rvH4D8IdyiAC3VekIbpt0Vj0SUX3DwKtog337BzTiPk3aX
+RF0sbFhQoqdJRI8NqgZjCwjq9yfI5tyxYswn+JGzHGdHvW3idODlmwEt5K2pasiR
+IWK2OGfq+w0EcltQHabuqEPgZlmhCkRdNfixBwIDAQABAoIBAA9jNoBkRdxmH/R9
+Wz+3gBqA9Aq4ZFuzJJk8QCm62V8ltWyyCnliYeKhPEm0QWrWOwghr/1AzW9Wt4g4
+wVJcabD5TwODF5L0626eZcM3bsscwR44TMJzEgD5EWC2j3mKqFCPaoBj08tq4KXh
+wW8tgjgz+eTk3cYD583qfTIZX1+SzSMBpetTBsssQtGhhOB/xPiuL7hi+fXmV2rh
+8mc9X6+wJ5u3zepsyK0vBeEDmurD4ZUIXFrZ0WCB/wNkSW9VKyoH+RC1asQAgqTz
+glJ/NPbDJSKGvSBQydoKkqoXx7MVJ8VObFddfgo4dtOoz6YCfUVBHt8qy+E5rz5y
+CICjL/kCgYEA9MnHntVVKNXtEFZPo02xgCwS3eG27ZwjYgJ1ZkCHM5BuL4MS7qbr
+743/POs1Ctaok0udHl1PFB4uAG0URnmkUnWzcoJYb6Plv03F0LRdsnfuhehfIxLP
+nWvxSm5n21H4ytfxm0BWY09JkLDnJZtXrgTILbuqb9Wy6TmAvUaF2YUCgYEA16Ec
+ywSaLVdqPaVpsTxi7XpRJAB2Isjp6RffNEecta4S0LL7s/IO3QXDH9SYpgmgCTah
+3aXhpT4hIFlpg3eBjVfbOwgqub8DgirnSQyQt99edUtHIK+K8nMdGxz6X6pfTKzK
+asSH7qPlt5tz1621vC0ocXSZR7zm99/FgwILwBsCgYBOsP8nJFV4By1qbxSy3qsN
+FR4LjiAMSoFlZHzxHhVYkjmZtH1FkwuNuwwuPT6T+WW/1DLyK/Tb9se7A1XdQgV9
+LLE/Qn/Dg+C7mvjYmuL0GHHpQkYzNDzh0m2DC/L/Il7kdn8I9anPyxFPHk9wW3vY
+SVlAum+T/BLDvuSP9DfbMQKBgCc1j7PG8XYfOB1fj7l/volqPYjrYI/wssAE7Dxo
+bTGIJrm2YhiVgmhkXNfT47IFfAlQ2twgBsjyZDmqqIoUWAVonV+9m29NMYkg3g+l
+bkdRIa74ckWaRgzSK8+7VDfDFjMuFFyXwhP9z460gLsORkaie4Et75Vg3yrhkNvC
+qnpTAoGBAMguDSWBbCewXnHlKGFpm+LH+OIvVKGEhtCSvfZojtNrg/JBeBebSL1n
+mmT1cONO+0O5bz7uVaRd3JdnH2JFevY698zFfhVsjVCrm+fz31i5cxAgC39G2Lfl
+YkTaa1AFLstnf348ZjuvBN3USUYZo3X3mxnS+uluVuRSGwIKsN0a
+-----END RSA PRIVATE KEY-----

internal/integration/config/zitadel.yaml

@@ -4,6 +4,11 @@ Log:
 TLS:
   Enabled: false
 
+Telemetry:
+  Enabled: true
+  Endpoints:
+  - http://localhost:8081
+
 FirstInstance:
   Org:
     Human:
@@ -28,10 +33,17 @@ Quotas:
     ExhaustedCookieMaxAge: "60s"
 
 Projections:
+  RequeueEvery: 1s
   Customizations:
     NotificationsQuotas:
       RequeueEvery: 1s
+    Telemetry:
+      RequeueEvery: 1s
 
 DefaultInstance:
   LoginPolicy:
     MfaInitSkipLifetime: "0"
+
+SystemAPIUsers:
+  - tester:
+      KeyData: "LS0tLS1CRUdJTiBQVUJMSUMgS0VZLS0tLS0KTUlJQklqQU5CZ2txaGtpRzl3MEJBUUVGQUFPQ0FROEFNSUlCQ2dLQ0FRRUF6aStGRlNKTDdmNXl3NEtUd3pnTQpQMzRlUEd5Y20vTStrVDBNN1Y0Q2d4NVYzRWFESXZUUUtUTGZCYUVCNDV6YjlMdGpJWHpEdzByWFJvUzJoTzZ0CmgrQ1lRQ3ozS0N2aDA5QzBJenhaaUIySVMzSC9hVCs1Qng5RUZZK3ZuQWtaamNjYnlHNVlOUnZtdE9sbnZJZUkKSDdxWjB0RXdrUGZGNUdFWk5QSlB0bXkzVUdWN2lvZmRWUVMxeFJqNzMrYU13NXJ2SDREOElkeWlBQzNWZWtJYgpwdDBWajBTVVgzRHdLdG9nMzM3QnpUaVBrM2FYUkYwc2JGaFFvcWRKUkk4TnFnWmpDd2pxOXlmSTV0eXhZc3duCitKR3pIR2RIdlczaWRPRGxtd0V0NUsycGFzaVJJV0syT0dmcSt3MEVjbHRRSGFidXFFUGdabG1oQ2tSZE5maXgKQndJREFRQUIKLS0tLS1FTkQgUFVCTElDIEtFWS0tLS0tCg=="

internal/integration/integration.go

@@ -8,6 +8,7 @@ import (
 	_ "embed"
 	"errors"
 	"fmt"
+	"math/rand"
 	"os"
 	"strings"
 	"sync"
@@ -15,6 +16,7 @@ import (
 
 	"github.com/spf13/viper"
 	"github.com/zitadel/logging"
+	"github.com/zitadel/oidc/v2/pkg/client"
 	"github.com/zitadel/oidc/v2/pkg/oidc"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
@@ -31,8 +33,13 @@ import (
 	"github.com/zitadel/zitadel/internal/query"
 	"github.com/zitadel/zitadel/internal/webauthn"
 	"github.com/zitadel/zitadel/pkg/grpc/admin"
+	"github.com/zitadel/zitadel/pkg/grpc/system"
 )
 
+func init() {
+	rand.Seed(time.Now().UnixNano())
+}
+
 var (
 	//go:embed config/zitadel.yaml
 	zitadelYAML []byte
@@ -40,6 +47,8 @@ var (
 	cockroachYAML []byte
 	//go:embed config/postgres.yaml
 	postgresYAML []byte
+	//go:embed config/system-user-key.pem
+	systemUserKey []byte
 )
 
 // UserType provides constants that give
@@ -53,6 +62,8 @@ type UserType int
 const (
 	Unspecified UserType = iota
 	OrgOwner
+	IAMOwner
+	SystemUser // SystemUser is a user with access to the system service.
 )
 
 // User information with a Personal Access Token.
@@ -80,11 +91,12 @@ func (s *Tester) Host() string {
 	return fmt.Sprintf("%s:%d", s.Config.ExternalDomain, s.Config.Port)
 }
 
-func (s *Tester) createClientConn(ctx context.Context) {
+func (s *Tester) createClientConn(ctx context.Context, opts ...grpc.DialOption) {
 	target := s.Host()
-	cc, err := grpc.DialContext(ctx, target,
-		grpc.WithBlock(), grpc.WithTransportCredentials(insecure.NewCredentials()),
-	)
+	cc, err := grpc.DialContext(ctx, target, append(opts,
+		grpc.WithBlock(),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)...)
 	if err != nil {
 		s.Shutdown <- os.Interrupt
 		s.wg.Wait()
@@ -124,10 +136,10 @@ func (s *Tester) pollHealth(ctx context.Context) (err error) {
 }
 
 const (
-	SystemUser = "integration"
+	MachineUser = "integration"
 )
 
-func (s *Tester) createSystemUser(ctx context.Context) {
+func (s *Tester) createMachineUser(ctx context.Context) {
 	var err error
 
 	s.Instance, err = s.Queries.InstanceByHost(ctx, s.Host())
@@ -137,7 +149,7 @@ func (s *Tester) createSystemUser(ctx context.Context) {
 	s.Organisation, err = s.Queries.OrgByID(ctx, true, s.Instance.DefaultOrganisationID())
 	logging.OnError(err).Fatal("query organisation")
 
-	query, err := query.NewUserUsernameSearchQuery(SystemUser, query.TextEquals)
+	query, err := query.NewUserUsernameSearchQuery(MachineUser, query.TextEquals)
 	logging.OnError(err).Fatal("user query")
 	user, err := s.Queries.GetUser(ctx, true, true, query)
 
@@ -146,8 +158,8 @@ func (s *Tester) createSystemUser(ctx context.Context) {
 			ObjectRoot: models.ObjectRoot{
 				ResourceOwner: s.Organisation.ID,
 			},
-			Username:        SystemUser,
-			Name:            SystemUser,
+			Username:        MachineUser,
+			Name:            MachineUser,
 			Description:     "who cares?",
 			AccessTokenType: domain.OIDCTokenTypeJWT,
 		})
@@ -176,10 +188,37 @@ func (s *Tester) createSystemUser(ctx context.Context) {
 	}
 }
 
-func (s *Tester) WithSystemAuthorization(ctx context.Context, u UserType) context.Context {
+func (s *Tester) WithAuthorization(ctx context.Context, u UserType) context.Context {
+	if u == SystemUser {
+		s.ensureSystemUser()
+	}
 	return metadata.AppendToOutgoingContext(ctx, "Authorization", fmt.Sprintf("Bearer %s", s.Users[u].Token))
 }
 
+func (s *Tester) ensureSystemUser() {
+	const ISSUER = "tester"
+
+	if _, ok := s.Users[SystemUser]; ok {
+		return
+	}
+	domain := viper.Get("ExternalDomain").(string)
+	port := viper.Get("ExternalPort").(int)
+	protocol := "http"
+	secure := viper.Get("ExternalSecure").(bool)
+	if secure {
+		protocol = "https"
+	}
+	audience := fmt.Sprintf("%s://%s:%d", protocol, domain, port)
+
+	signer, err := client.NewSignerFromPrivateKeyByte(systemUserKey, "")
+	logging.OnError(err).Fatal("system key signer")
+
+	jwt, err := client.SignedJWTProfileAssertion(ISSUER, []string{audience}, time.Hour, signer)
+	logging.OnError(err).Fatal("system key jwt")
+
+	s.Users[SystemUser] = User{Token: jwt}
+}
+
 // Done send an interrupt signal to cleanly shutdown the server.
 func (s *Tester) Done() {
 	err := s.Client.CC.Close()
@@ -237,7 +276,7 @@ func NewTester(ctx context.Context) *Tester {
 		logging.OnError(ctx.Err()).Fatal("waiting for integration tester server")
 	}
 	tester.createClientConn(ctx)
-	tester.createSystemUser(ctx)
+	tester.createMachineUser(ctx)
 	tester.WebAuthN = webauthn.NewClient(tester.Config.WebAuthNName, tester.Config.ExternalDomain, "https://"+tester.Host())
 
 	return tester
@@ -249,3 +288,37 @@ func Contexts(timeout time.Duration) (ctx, errCtx context.Context, cancel contex
 	ctx, cancel = context.WithTimeout(context.Background(), timeout)
 	return ctx, errCtx, cancel
 }
+
+var letterRunes = []rune("abcdefghijklmnopqrstuvwxyz")
+
+func randStringRunes(n int) string {
+	b := make([]rune, n)
+	for i := range b {
+		b[i] = letterRunes[rand.Intn(len(letterRunes))]
+	}
+	return string(b)
+}
+
+func (t *Tester) UseIsolatedInstance(ctx context.Context) (primaryDomain, instanceID string, systemCtx, iamOwnerCtx context.Context) {
+	systemCtx = t.WithAuthorization(ctx, SystemUser)
+	primaryDomain = randStringRunes(5) + ".integration"
+	instance, err := t.Client.System.CreateInstance(systemCtx, &system.CreateInstanceRequest{
+		InstanceName: "testinstance",
+		CustomDomain: primaryDomain,
+		Owner: &system.CreateInstanceRequest_Machine_{
+			Machine: &system.CreateInstanceRequest_Machine{
+				UserName:            "owner",
+				Name:                "owner",
+				PersonalAccessToken: &system.CreateInstanceRequest_PersonalAccessToken{},
+			},
+		},
+	})
+	if err != nil {
+		panic(err)
+	}
+	t.createClientConn(ctx, grpc.WithAuthority(primaryDomain))
+	t.Users[IAMOwner] = User{
+		Token: instance.GetPat(),
+	}
+	return primaryDomain, instance.GetInstanceId(), systemCtx, t.WithAuthorization(ctx, IAMOwner)
+}

internal/notification/handlers/handlers_integration_test.go

@@ -0,0 +1,40 @@
+//go:build integration
+
+package handlers_test
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/zitadel/zitadel/pkg/grpc/management"
+
+	"github.com/zitadel/zitadel/pkg/grpc/system"
+
+	"github.com/zitadel/zitadel/internal/integration"
+)
+
+var (
+	Tester                    *integration.Tester
+	SystemUserCTX             context.Context
+	SystemClient              system.SystemServiceClient
+	PrimaryDomain, InstanceID string
+	IAMOwnerCtx               context.Context
+	MgmtClient                management.ManagementServiceClient
+)
+
+func TestMain(m *testing.M) {
+	os.Exit(func() int {
+		ctx, _, cancel := integration.Contexts(5 * time.Minute)
+		defer cancel()
+		os.Setenv("INTEGRATION_DB_FLAVOR", "postgres")
+		os.Setenv("ZITADEL_MASTERKEY", "MasterkeyNeedsToHave32Characters")
+		Tester = integration.NewTester(ctx)
+		PrimaryDomain, InstanceID, SystemUserCTX, IAMOwnerCtx = Tester.UseIsolatedInstance(ctx)
+		MgmtClient = Tester.Client.Mgmt
+		SystemClient = Tester.Client.System
+		defer Tester.Done()
+		return m.Run()
+	}())
+}

internal/notification/handlers/telemetry_pusher_integration_test.go

@@ -3,129 +3,77 @@
 package handlers_test
 
 import (
-	"context"
-	"os"
+	"bytes"
+	"encoding/json"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/require"
-	"github.com/zitadel/zitadel/internal/integration"
-	object "github.com/zitadel/zitadel/pkg/grpc/object/v2alpha"
-	session "github.com/zitadel/zitadel/pkg/grpc/session/v2alpha"
-	user "github.com/zitadel/zitadel/pkg/grpc/user/v2alpha"
+	"github.com/zitadel/zitadel/pkg/grpc/management"
+	"github.com/zitadel/zitadel/pkg/grpc/system"
 )
 
-var (
-	CTX               context.Context
-	Tester            *integration.Tester
-	Client            session.SessionServiceClient
-	User              *user.AddHumanUserResponse
-	GenericOAuthIDPID string
-)
-
-func TestMain(m *testing.M) {
-	os.Exit(func() int {
-		ctx, errCtx, cancel := integration.Contexts(5 * time.Minute)
-		defer cancel()
-
-		Tester = integration.NewTester(ctx)
-		defer Tester.Done()
-		Client = Tester.Client.SessionV2
-
-		CTX, _ = Tester.WithSystemAuthorization(ctx, integration.OrgOwner), errCtx
-		User = Tester.CreateHumanUser(CTX)
-		Tester.RegisterUserPasskey(CTX, User.GetUserId())
-		return m.Run()
-	}())
+func TestServer_TelemetryPusher(t *testing.T) {
+	bodies := make(chan []byte, 0)
+	t.Log("testing against instance with primary domain", PrimaryDomain)
+	mockServer := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		if err != nil {
+			t.Error(err)
+		}
+		bodies <- body
+		w.WriteHeader(http.StatusOK)
+	}))
+	listener, err := net.Listen("tcp", "localhost:8081")
+	if err != nil {
+		t.Fatal(err)
+	}
+	mockServer.Listener = listener
+	mockServer.Start()
+	t.Cleanup(mockServer.Close)
+	awaitMilestone(t, bodies, "InstanceCreated")
+	project, err := MgmtClient.AddProject(IAMOwnerCtx, &management.AddProjectRequest{Name: "integration"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	awaitMilestone(t, bodies, "ProjectCreated")
+	if _, err = MgmtClient.AddOIDCApp(IAMOwnerCtx, &management.AddOIDCAppRequest{
+		ProjectId: project.GetId(),
+		Name:      "integration",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	awaitMilestone(t, bodies, "ApplicationCreated")
+	if _, err = SystemClient.RemoveInstance(SystemUserCTX, &system.RemoveInstanceRequest{InstanceId: InstanceID}); err != nil {
+		t.Fatal(err)
+	}
+	awaitMilestone(t, bodies, "InstanceDeleted")
 }
 
-func TestServer_TestInstanceCreatedMiletone(t *testing.T) {
-	tests := []struct {
-		name        string
-		req         *session.CreateSessionRequest
-		want        *session.CreateSessionResponse
-		wantErr     bool
-		wantFactors []wantFactor
-	}{
-		{
-			name: "empty session",
-			req: &session.CreateSessionRequest{
-				Metadata: map[string][]byte{"foo": []byte("bar")},
-			},
-			want: &session.CreateSessionResponse{
-				Details: &object.Details{
-					ResourceOwner: Tester.Organisation.ID,
-				},
-			},
-		},
-		{
-			name: "with user",
-			req: &session.CreateSessionRequest{
-				Checks: &session.Checks{
-					User: &session.CheckUser{
-						Search: &session.CheckUser_UserId{
-							UserId: User.GetUserId(),
-						},
-					},
-				},
-				Metadata: map[string][]byte{"foo": []byte("bar")},
-				Domain:   "domain",
-			},
-			want: &session.CreateSessionResponse{
-				Details: &object.Details{
-					ResourceOwner: Tester.Organisation.ID,
-				},
-			},
-			wantFactors: []wantFactor{wantUserFactor},
-		},
-		{
-			name: "password without user error",
-			req: &session.CreateSessionRequest{
-				Checks: &session.Checks{
-					Password: &session.CheckPassword{
-						Password: "Difficult",
-					},
-				},
-			},
-			wantErr: true,
-		},
-		{
-			name: "passkey without user error",
-			req: &session.CreateSessionRequest{
-				Challenges: []session.ChallengeKind{
-					session.ChallengeKind_CHALLENGE_KIND_PASSKEY,
-				},
-			},
-			wantErr: true,
-		},
-		{
-			name: "passkey without domain (not registered) error",
-			req: &session.CreateSessionRequest{
-				Checks: &session.Checks{
-					User: &session.CheckUser{
-						Search: &session.CheckUser_UserId{
-							UserId: User.GetUserId(),
-						},
-					},
-				},
-				Challenges: []session.ChallengeKind{
-					session.ChallengeKind_CHALLENGE_KIND_PASSKEY,
-				},
-			},
-			wantErr: true,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got, err := Client.CreateSession(CTX, tt.req)
-			if tt.wantErr {
-				require.Error(t, err)
+func awaitMilestone(t *testing.T, bodies chan []byte, expectMilestoneType string) {
+	for {
+		select {
+		case body := <-bodies:
+			plain := new(bytes.Buffer)
+			if err := json.Indent(plain, body, "", "  "); err != nil {
+				t.Fatal(err)
+			}
+			t.Log("received milestone", plain.String())
+			milestone := struct {
+				Type          string
+				PrimaryDomain string
+			}{}
+			if err := json.Unmarshal(body, &milestone); err != nil {
+				t.Error(err)
+			}
+			if milestone.Type == expectMilestoneType && milestone.PrimaryDomain == PrimaryDomain {
 				return
 			}
-			require.NoError(t, err)
-			integration.AssertDetails(t, tt.want, got)
-
-			verifyCurrentSession(t, got.GetSessionId(), got.GetSessionToken(), got.GetDetails().GetSequence(), time.Minute, tt.req.GetMetadata(), tt.wantFactors...)
-		})
+		case <-time.After(60 * time.Second):
+			t.Fatalf("timed out waiting for milestone")
+		}
 	}
 }