wip: sse working

k8-operator/config/crd/bases/secrets.infisical.com_infisicalsecrets.yaml

@@ -274,6 +274,9 @@ spec:
               hostAPI:
                 description: Infisical host to pull secrets from
                 type: string
+              instantUpdates:
+                default: true
+                type: boolean
               managedKubeConfigMapReferences:
                 items:
                   properties:
@@ -432,6 +435,7 @@ spec:
                 - secretNamespace
                 type: object
             required:
+            - instantUpdates
             - resyncInterval
             type: object
           status:

k8-operator/config/samples/crd/infisicalsecret/infisicalSecretCrd.yaml

@@ -7,8 +7,10 @@ metadata:
   annotations:
     example.com/annotation-to-be-passed-to-managed-secret: "sample-value"
 spec:
-  hostAPI: https://app.infisical.com/api
+  # hostAPI: https://app.infisical.com/api
+  hostAPI: http://localhost:8085/api
   resyncInterval: 10
+  instantUpdates: true
   # tls:
   #   caRef:
   #     secretName: custom-ca-certificate
@@ -19,19 +21,19 @@ spec:
     # If you have multiple authentication methods defined, it may cause issues.
 
     # (Deprecated) Service Token Auth
-    serviceToken:
-      serviceTokenSecretReference:
-        secretName: service-token
-        secretNamespace: default
-      secretsScope:
-        envSlug: <env-slug>
-        secretsPath: <secrets-path>
-        recursive: true
+    # serviceToken:
+    #   serviceTokenSecretReference:
+    #     secretName: service-token
+    #     secretNamespace: default
+    #   secretsScope:
+    #     envSlug: <env-slug>
+    #     secretsPath: <secrets-path>
+    #     recursive: true
 
     # Universal Auth
     universalAuth:
       secretsScope:
-        projectSlug: new-ob-em
+        projectSlug: test-j7-kx
         envSlug: dev # "dev", "staging", "prod", etc..
         secretsPath: "/" # Root is "/"
         recursive: true # Wether or not to use recursive mode (Fetches all secrets in an environment from a given secret path, and all folders inside the path) / defaults to false
@@ -39,67 +41,6 @@ spec:
         secretName: universal-auth-credentials
         secretNamespace: default
 
-    # Native Kubernetes Auth
-    kubernetesAuth:
-      serviceAccountRef:
-        name: <secret-name>
-        namespace: <secret-namespace>
-      identityId: <machine-identity-id>
-      serviceAccountTokenPath: "/path/to/your/service-account/token" # Optional, defaults to /var/run/secrets/kubernetes.io/serviceaccount/token
-
-      # secretsScope is identical to the secrets scope in the universalAuth field in this sample.
-      secretsScope:
-        projectSlug: your-project-slug
-        envSlug: prod
-        secretsPath: "/path"
-        recursive: true
-
-    # AWS IAM Auth
-    awsIamAuth:
-      identityId: <your-machine-identity-id>
-
-      # secretsScope is identical to the secrets scope in the universalAuth field in this sample.
-      secretsScope:
-        projectSlug: your-project-slug
-        envSlug: prod
-        secretsPath: "/path"
-        recursive: true
-
-    # Azure Auth
-    azureAuth:
-      identityId: <your-machine-identity-id>
-      resource: https://management.azure.com/&client_id=your_client_id # This field is optional, and will default to "https://management.azure.com/" if nothing is provided.
-
-      # secretsScope is identical to the secrets scope in the universalAuth field in this sample.
-      secretsScope:
-        projectSlug: your-project-slug
-        envSlug: prod
-        secretsPath: "/path"
-        recursive: true
-
-    # GCP ID Token Auth
-    gcpIdTokenAuth:
-      identityId: <your-machine-identity-id>
-
-      # secretsScope is identical to the secrets scope in the universalAuth field in this sample.
-      secretsScope:
-        projectSlug: your-project-slug
-        envSlug: prod
-        secretsPath: "/path"
-        recursive: true
-
-    # GCP IAM Auth
-    gcpIamAuth:
-      identityId: <your-machine-identity-id>
-      serviceAccountKeyFilePath: "/path/to-service-account-key-file-path.json"
-
-      # secretsScope is identical to the secrets scope in the universalAuth field in this sample.
-      secretsScope:
-        projectSlug: your-project-slug
-        envSlug: prod
-        secretsPath: "/path"
-        recursive: true
-
   managedKubeSecretReferences:
     - secretName: managed-secret
       secretNamespace: default

k8-operator/config/samples/infisical-config.yaml

@@ -9,4 +9,4 @@ metadata:
   name: infisical-config
   namespace: infisical-operator-system
 data:
-  hostAPI: "https://example.com/api"
+  hostAPI: "http://localhost:8085/api"

k8-operator/config/samples/k8s-auth/sample.yaml

@@ -7,19 +7,19 @@ metadata:
     annotations:
         example.com/annotation-to-be-passed-to-managed-secret: "sample-value"
 spec:
-    hostAPI: https://app.infisical.com/api
+    hostAPI: http://localhost:8085/api
     resyncInterval: 10
     authentication:
         # Native Kubernetes Auth
         kubernetesAuth:
-            identityId: <>
+            identityId: 25d05878-0a2f-47c5-b889-46b70fb2dbfc
             serviceAccountRef:
               name: infisical-auth
               namespace: default
 
             # secretsScope is identical to the secrets scope in the universalAuth field in this sample.
             secretsScope:
-                projectSlug: dsf-gpb-t
+                projectSlug: test-j7-kx
                 envSlug: dev
                 secretsPath: "/"
                 recursive: true

k8-operator/config/samples/machineIdentitySecret.yaml

@@ -4,5 +4,5 @@ metadata:
     name: universal-auth-credentials
 type: Opaque
 stringData:
-    clientId: <machine-identity-client-id>
-    clientSecret: <machine-identity-client-secret>
+    clientId: 86c17800-00b9-4556-89a8-ea11ac6f8371
+    clientSecret: 6792edff19e0b7a43647501958c9a0d1a7964b2db1ff2401fbdd555cfd1ae757

k8-operator/config/samples/serviceTokenSecret.yaml

@@ -1,7 +1,7 @@
-apiVersion: v1
-kind: Secret
-metadata:
-  name: service-token
-type: Opaque
-data:
-  infisicalToken: <base64 infisical token here>
+# apiVersion: v1
+# kind: Secret
+# metadata:
+#   name: service-token
+# type: Opaque
+# data:
+#   infisicalToken: <base64 infisical token here>

k8-operator/controllers/infisicaldynamicsecret/infisicaldynamicsecret_controller.go

@@ -168,7 +168,6 @@ func (r *InfisicalDynamicSecretReconciler) Reconcile(ctx context.Context, req ct
 }
 
 func (r *InfisicalDynamicSecretReconciler) SetupWithManager(mgr ctrl.Manager) error {
-
 	// Custom predicate that allows both spec changes and deletions
 	specChangeOrDelete := predicate.Funcs{
 		UpdateFunc: func(e event.UpdateEvent) bool {

k8-operator/controllers/infisicalpushsecret/infisicalpushsecret_controller.go

@@ -208,7 +208,6 @@ func (r *InfisicalPushSecretReconciler) Reconcile(ctx context.Context, req ctrl.
 }
 
 func (r *InfisicalPushSecretReconciler) SetupWithManager(mgr ctrl.Manager) error {
-
 	// Custom predicate that allows both spec changes and deletions
 	specChangeOrDelete := predicate.Funcs{
 		UpdateFunc: func(e event.UpdateEvent) bool {

k8-operator/controllers/infisicalsecret/infisicalsecret_controller.go

@@ -201,6 +201,8 @@ func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 }
 
 func (r *InfisicalSecretReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	r.SourceCh = make(chan event.GenericEvent)
+
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&secretsv1alpha1.InfisicalSecret{}, builder.WithPredicates(predicate.Funcs{
 			UpdateFunc: func(e event.UpdateEvent) bool {

k8-operator/controllers/infisicalsecret/infisicalsecret_helper.go

@@ -3,6 +3,7 @@ package controllers
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"strings"
@@ -466,7 +467,7 @@ func (r *InfisicalSecretReconciler) getResourceVariables(infisicalSecret v1alpha
 			InfisicalClient:   client,
 			CancelCtx:         cancel,
 			AuthDetails:       util.AuthenticationDetails{},
-			EventStreamClient: sse.NewClient(api.API_HOST_URL, api.USER_AGENT_NAME),
+			EventStreamClient: sse.NewClient(api.API_HOST_URL),
 		}
 
 		resourceVariables = infisicalSecretResourceVariablesMap[string(infisicalSecret.UID)]
@@ -578,26 +579,68 @@ func (r *InfisicalSecretReconciler) EnsureEventStream(ctx context.Context, logge
 
 	resourceVariables := r.getResourceVariables(*secret)
 	infiscalClient := resourceVariables.InfisicalClient
-	// sseClient := resourceVariables.EventStreamClient
-
 	projectSlug := resourceVariables.AuthDetails.MachineIdentityScope.ProjectSlug
 
-	proj, err := util.GetProjectByID(infiscalClient.Auth().GetAccessToken(), projectSlug)
+	token := infiscalClient.Auth().GetAccessToken()
 
-	logger.Info("Project", proj)
+	proj, err := util.GetProjectBySlug(token, projectSlug)
+
+	logger.Info("Project", "project", proj, "slug", projectSlug)
 	if err != nil {
 		return fmt.Errorf("failed to get project [err=%s]", err)
 	}
 
-	api.CallSubscribeProjectEvents(secret.Spec.Authentication.ServiceAccount.ProjectId, api.SubProjectEventsRequest{
+	client := sse.NewClient(fmt.Sprintf("%s/v1/events/subscribe/project-events", api.API_HOST_URL))
+
+	registers := []api.SubProjectEventsRequestRegister{
+		api.SubProjectEventsRequestRegister{
+			Event: "secret:delete",
+			Conditions: &api.SubProjectEventsRequestCondition{
+				SecretPath:      "/**",
+				EnvironmentSlug: secret.Spec.Authentication.UniversalAuth.SecretsScope.EnvSlug,
+			},
+		},
+	}
+
+	b, err := json.Marshal(api.SubProjectEventsRequest{
 		ProjectID: proj.ID,
+		Register:  registers,
 	})
 
-	return nil
+	if err != nil {
+		return fmt.Errorf("CallSubscribeProjectEvents: Unable to marshal body [err=%s]", err)
+	}
+
+	headers := map[string]string{
+		"User-Agent":    api.USER_AGENT_NAME,
+		"Authorization": fmt.Sprint("Bearer ", token),
+	}
+
+	events, errors, err := client.Connect("POST", headers, strings.NewReader(string(b)))
+
+	if err != nil {
+		return fmt.Errorf("Unable to connect to SSE server [err=%s]", err)
+	}
+
+	if err != nil {
+		return fmt.Errorf("failed to subscribe to project events [err=%s]", err)
+	}
+
+	for {
+		select {
+		case event := <-events:
+			logger.Info("Received event", "event", event)
+		case err := <-errors:
+			logger.Error(err, "Error occurred")
+		case <-ctx.Done():
+			logger.Info("Context done")
+			return nil
+		}
+	}
 }
 
 func (r *InfisicalSecretReconciler) CloseEventStream(ctx context.Context, logger logr.Logger, infisicalSecretCRD *secretsv1alpha1.InfisicalSecret) error {
-	logger.Info("Event watcher enabled")
+	logger.Info("Event watcher disabled")
 	// ensure event watcher is running
 
 	return nil

k8-operator/packages/api/api.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/Infisical/infisical/k8-operator/packages/model"
 	"github.com/Infisical/infisical/k8-operator/packages/util/sse"
 	"github.com/go-resty/resty/v2"
 )
@@ -150,17 +151,40 @@ func CallGetProjectByID(httpClient *resty.Client, request GetProjectByIDRequest)
 
 }
 
-func CallSubscribeProjectEvents(projectID string, body SubProjectEventsRequest) (<-chan sse.SSEEvent, <-chan error, error) {
-	client := sse.NewClient(fmt.Sprintf("%s/api/v1/events/subscribe/project-events", API_HOST_URL), USER_AGENT_NAME)
+func CallGetProjectByIDv2(httpClient *resty.Client, request GetProjectByIDRequest) (model.Project, error) {
+	var projectResponse model.Project
+
+	response, err := httpClient.
+		R().SetResult(&projectResponse).
+		SetHeader("User-Agent", USER_AGENT_NAME).
+		Get(fmt.Sprintf("%s/v2/workspace/%s", API_HOST_URL, request.ProjectID))
+
+	if err != nil {
+		return model.Project{}, fmt.Errorf("CallGetProject: Unable to complete api request [err=%s]", err)
+	}
+
+	if response.IsError() {
+		return model.Project{}, fmt.Errorf("CallGetProject: Unsuccessful response: [response=%s]", response)
+	}
+
+	return projectResponse, nil
+
+}
+
+func CallSubscribeProjectEvents(projectID, token string, body SubProjectEventsRequest) (<-chan sse.SSEEvent, <-chan error, error) {
+	client := sse.NewClient(fmt.Sprintf("%s/v1/events/subscribe/project-events", API_HOST_URL))
 
 	b, err := json.Marshal(body)
 	if err != nil {
 		return nil, nil, fmt.Errorf("CallSubscribeProjectEvents: Unable to marshal body [err=%s]", err)
 	}
 
-	events, errors, err := client.Connect("POST", map[string]string{
-		"User-Agent": USER_AGENT_NAME,
-	}, strings.NewReader(string(b)))
+	headers := map[string]string{
+		"User-Agent":    USER_AGENT_NAME,
+		"Authorization": fmt.Sprint("Bearer ", token),
+	}
+
+	events, errors, err := client.Connect("POST", headers, strings.NewReader(string(b)))
 
 	if err != nil {
 		return nil, nil, fmt.Errorf("CallSubscribeProjectEvents: Unable to connect to SSE server [err=%s]", err)

k8-operator/packages/api/models.go

@@ -208,11 +208,13 @@ type GetProjectByIDResponse struct {
 }
 
 type SubProjectEventsRequestRegister struct {
-	EventType  string `json:"eventType"`
-	Conditions *struct {
-		EnvironmentSlug string `json:"environmentSlug"`
-		SecretPath      string `json:"secretPath"`
-	} `json:"conditions"`
+	Event      string                            `json:"event"`
+	Conditions *SubProjectEventsRequestCondition `json:"conditions"`
+}
+
+type SubProjectEventsRequestCondition struct {
+	EnvironmentSlug string `json:"environmentSlug"`
+	SecretPath      string `json:"secretPath"`
 }
 
 type SubProjectEventsRequest struct {

k8-operator/packages/util/sse/sse.go

@@ -16,16 +16,14 @@ type SSEEvent struct {
 
 // SSEClient handles SSE connections
 type SSEClient struct {
-	URL       string
-	UserAgent string
-	Client    *http.Client
+	URL    string
+	Client *http.Client
 }
 
 // NewSSEClient creates a new SSE client
-func NewClient(url, ua string) *SSEClient {
+func NewClient(url string) *SSEClient {
 	return &SSEClient{
-		URL:       url,
-		UserAgent: ua,
+		URL: url,
 		Client: &http.Client{
 			Timeout: 0, // No timeout for streaming
 		},
@@ -42,6 +40,7 @@ func (c *SSEClient) Connect(method string, headers map[string]string, body io.Re
 	// Set required headers for SSE
 	req.Header.Set("Cache-Control", "no-cache")
 	req.Header.Set("Connection", "keep-alive")
+	req.Header.Set("Content-Type", "application/json")
 
 	for key, value := range headers {
 		req.Header.Set(key, value)

k8-operator/packages/util/workspace.go

@@ -9,7 +9,6 @@ import (
 )
 
 func GetProjectByID(accessToken string, projectId string) (model.Project, error) {
-
 	httpClient := resty.New()
 	httpClient.
 		SetAuthScheme("Bearer").
@@ -25,3 +24,21 @@ func GetProjectByID(accessToken string, projectId string) (model.Project, error)
 
 	return projectDetails.Project, nil
 }
+
+func GetProjectBySlug(accessToken string, projectSlug string) (model.Project, error) {
+	httpClient := resty.New()
+	httpClient.
+		SetAuthScheme("Bearer").
+		SetAuthToken(accessToken).
+		SetHeader("Accept", "application/json")
+
+	project, err := api.CallGetProjectByIDv2(httpClient, api.GetProjectByIDRequest{
+		ProjectID: projectSlug,
+	})
+
+	if err != nil {
+		return model.Project{}, fmt.Errorf("unable to get project by slug. [err=%v]", err)
+	}
+
+	return project, nil
+}