Merge pull request #2554 from keboola/hosan/PAT-1542

Add support for E2B access tokens in Kubernetes apps

Author: hosekpeter

internal/pkg/service/appsproxy/dataapps/k8sapp/appinfo.go

@@ -11,13 +11,20 @@ const (
 	Group    = "apps.keboola.com"
 	Version  = "v2"
 	Resource = "apps"
+
+	BackendTypeE2BSandbox = "e2bSandbox"
 )
 
 // AppGVR returns the GroupVersionResource for the App CRD.
 func AppGVR() schema.GroupVersionResource {
 	return schema.GroupVersionResource{Group: Group, Version: Version, Resource: Resource}
 }
 
+// SecretGVR returns the GroupVersionResource for core/v1 Secrets.
+func SecretGVR() schema.GroupVersionResource {
+	return schema.GroupVersionResource{Group: "", Version: "v1", Resource: "secrets"}
+}
+
 // AppActualState is the observed state of the app, read from .status.currentState.
 type AppActualState string
 
@@ -35,8 +42,17 @@ type appObject struct {
 }
 
 type appSpec struct {
-	AppID              string `json:"appId"`
-	AutoRestartEnabled *bool  `json:"autoRestartEnabled,omitempty"`
+	AppID              string     `json:"appId"`
+	AutoRestartEnabled *bool      `json:"autoRestartEnabled,omitempty"`
+	Runtime            appRuntime `json:"runtime"`
+}
+
+type appRuntime struct {
+	Backend appBackend `json:"backend"`
+}
+
+type appBackend struct {
+	Type string `json:"type,omitempty"`
 }
 
 // AppInfo is the cached state for an app, read from the K8s watcher.
@@ -46,11 +62,20 @@ type AppInfo struct {
 	// UpstreamTarget is the pre-parsed URL from .status.appsProxy.upstreamUrl.
 	// Nil when the field is absent or unparseable.
 	UpstreamTarget *url.URL
+	// E2BAccessToken is the access token loaded from the K8s Secret
+	// referenced by .status.e2bSandbox.accessTokenSecretName.
+	// Empty when the app is not an E2B sandbox or the secret is unavailable.
+	E2BAccessToken string
 }
 
 type appStatus struct {
 	CurrentState AppActualState `json:"currentState"`
 	AppsProxy    appsProxy      `json:"appsProxy"`
+	E2BSandbox   e2bSandbox     `json:"e2bSandbox"`
+}
+
+type e2bSandbox struct {
+	AccessTokenSecretName string `json:"accessTokenSecretName,omitempty"`
 }
 
 type appsProxy struct {

internal/pkg/service/appsproxy/dataapps/k8sapp/watcher.go

@@ -2,10 +2,12 @@ package k8sapp
 
 import (
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"net/url"
 	"sync"
 
+	"golang.org/x/sync/singleflight"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -19,6 +21,7 @@ import (
 	"github.com/keboola/keboola-as-code/internal/pkg/log"
 	"github.com/keboola/keboola-as-code/internal/pkg/service/appsproxy/dataapps/api"
 	"github.com/keboola/keboola-as-code/internal/pkg/service/common/servicectx"
+	"github.com/keboola/keboola-as-code/internal/pkg/utils/errors"
 )
 
 // entry stores the K8s object name and last observed state for an app.
@@ -27,16 +30,18 @@ type entry struct {
 	state              AppActualState
 	autoRestartEnabled bool
 	upstreamTarget     *url.URL // pre-parsed; nil when appsProxy.upstreamUrl absent/invalid
+	e2bAccessToken     string   // loaded from K8s Secret; empty for non-E2B apps
+	e2bSecretName      string   // Secret name for lazy token loading; empty for non-E2B apps
 }
 
 // StateWatcher watches App CRDs in Kubernetes and provides a local cache of app states.
 type StateWatcher struct {
-	client    dynamic.Interface
-	namespace string
-	logger    log.Logger
-	hasSynced cache.InformerSynced
-	// apps: AppID → entry
-	apps sync.Map
+	client         dynamic.Interface
+	namespace      string
+	logger         log.Logger
+	hasSynced      cache.InformerSynced
+	apps           sync.Map           // AppID → entry
+	tokenLoadGroup singleflight.Group // coalesces concurrent lazy-load K8s API calls per secret
 }
 
 type dependencies interface {
@@ -69,10 +74,10 @@ func NewStateWatcher(d dependencies, client dynamic.Interface, namespace string)
 	})
 
 	lw := &cache.ListWatch{
-		ListFunc: func(opts metav1.ListOptions) (runtime.Object, error) {
+		ListWithContextFunc: func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
 			return client.Resource(AppGVR()).Namespace(namespace).List(ctx, opts)
 		},
-		WatchFunc: func(opts metav1.ListOptions) (watch.Interface, error) {
+		WatchFuncWithContext: func(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error) {
 			return client.Resource(AppGVR()).Namespace(namespace).Watch(ctx, opts)
 		},
 	}
@@ -114,16 +119,34 @@ func NewStateWatcher(d dependencies, client dynamic.Interface, namespace string)
 }
 
 // GetState returns the cached AppInfo for the app. Returns (AppInfo{}, false) if not yet cached.
+// If the E2B access token is missing but a secret name is known, it attempts to load the token lazily.
 func (w *StateWatcher) GetState(appID api.AppID) (AppInfo, bool) {
 	v, ok := w.apps.Load(appID)
 	if !ok {
 		return AppInfo{}, false
 	}
 	e := v.(entry)
+
+	// Lazy-load E2B token: the Secret may not have existed when the App CRD event was processed.
+	// a singleflight coalesces concurrent requests for the same secret into a single K8s API call.
+	if e.e2bAccessToken == "" && e.e2bSecretName != "" {
+		token, err, _ := w.tokenLoadGroup.Do(e.e2bSecretName, func() (any, error) {
+			return w.loadSecretToken(context.Background(), e.e2bSecretName)
+		})
+		if err != nil {
+			w.logger.Warnf(context.Background(), "App %s: failed to lazy-load E2B access token from secret %q: %s", appID, e.e2bSecretName, err)
+		} else if t, ok := token.(string); t != "" && ok {
+			e.e2bAccessToken = t
+			w.apps.Store(appID, e)
+			w.logger.Infof(context.Background(), "App %s: lazy-loaded E2B access token from secret %q", appID, e.e2bSecretName)
+		}
+	}
+
 	return AppInfo{
 		ActualState:        e.state,
 		AutoRestartEnabled: e.autoRestartEnabled,
 		UpstreamTarget:     e.upstreamTarget,
+		E2BAccessToken:     e.e2bAccessToken,
 	}, true
 }
 
@@ -194,12 +217,26 @@ func (w *StateWatcher) handleUpsert(ctx context.Context, obj any) {
 		}
 	}
 
+	var e2bAccessToken string
+	var e2bSecretName string
+	if appObj.Spec.Runtime.Backend.Type == BackendTypeE2BSandbox {
+		e2bSecretName = appObj.Status.E2BSandbox.AccessTokenSecretName
+		if e2bSecretName != "" {
+			token, err := w.loadSecretToken(ctx, e2bSecretName)
+			if err == nil {
+				e2bAccessToken = token
+			}
+		}
+	}
+
 	appID := api.AppID(appObj.Spec.AppID)
 	w.apps.Store(appID, entry{
 		k8sName:            k8sName,
 		state:              appObj.Status.CurrentState,
 		autoRestartEnabled: autoRestartEnabled,
 		upstreamTarget:     upstreamTarget,
+		e2bAccessToken:     e2bAccessToken,
+		e2bSecretName:      e2bSecretName,
 	})
 	w.logger.Debugf(ctx, "App CRD %q (appID=%s) state updated: actualState=%q autoRestartEnabled=%v upstreamTarget=%v", k8sName, appID, appObj.Status.CurrentState, autoRestartEnabled, upstreamTarget != nil)
 }
@@ -229,3 +266,32 @@ func (w *StateWatcher) handleDelete(ctx context.Context, obj any) {
 		return true
 	})
 }
+
+// loadSecretToken fetches a K8s Secret by name and returns the value of the "token" key.
+// The dynamic client returns Secret data values as base64-encoded strings.
+func (w *StateWatcher) loadSecretToken(ctx context.Context, secretName string) (string, error) {
+	obj, err := w.client.Resource(SecretGVR()).Namespace(w.namespace).Get(ctx, secretName, metav1.GetOptions{})
+	if err != nil {
+		return "", err
+	}
+
+	data, found, err := unstructured.NestedMap(obj.Object, "data")
+	if err != nil {
+		return "", errors.Errorf("secret %q: failed to read data field: %s", secretName, err)
+	}
+	if !found {
+		return "", errors.Errorf("secret %q has no data field", secretName)
+	}
+
+	token, ok := data["token"].(string)
+	if !ok || token == "" {
+		return "", errors.Errorf("secret %q has no \"token\" key in data", secretName)
+	}
+
+	tokenBytes, err := base64.StdEncoding.DecodeString(token)
+	if err != nil {
+		return "", errors.Errorf("secret %q: failed to base64-decode token: %s", secretName, err)
+	}
+
+	return string(tokenBytes), nil
+}

internal/pkg/service/appsproxy/dataapps/k8sapp/watcher_test.go

@@ -2,6 +2,7 @@ package k8sapp_test
 
 import (
 	"context"
+	"encoding/base64"
 	"testing"
 	"time"
 
@@ -43,11 +44,12 @@ func newTestDeps(t *testing.T) *watcherDeps {
 func (d *watcherDeps) Logger() log.Logger           { return d.logger }
 func (d *watcherDeps) Process() *servicectx.Process { return d.proc }
 
-// newFakeClient creates a fake dynamic client with the App list kind registered.
+// newFakeClient creates a fake dynamic client with the App and Secret list kinds registered.
 func newFakeClient() *k8sfake.FakeDynamicClient {
 	scheme := runtime.NewScheme()
 	return k8sfake.NewSimpleDynamicClientWithCustomListKinds(scheme, map[schema.GroupVersionResource]string{
-		k8sapp.AppGVR(): "AppList",
+		k8sapp.AppGVR():    "AppList",
+		k8sapp.SecretGVR(): "SecretList",
 	})
 }
 
@@ -211,3 +213,117 @@ func TestStateWatcher_GetState_UpstreamTarget_AbsentWhenMissing(t *testing.T) {
 	require.True(t, ok)
 	assert.Nil(t, info.UpstreamTarget)
 }
+
+// newE2BAppObject creates an App CRD with spec.runtime.backend.type = "e2bSandbox"
+// and status.e2bSandbox.accessTokenSecretName set.
+func newE2BAppObject(k8sName, appID string, state k8sapp.AppActualState, secretName string) *unstructured.Unstructured {
+	obj := newAppObject(k8sName, appID, state)
+	obj.Object["spec"].(map[string]any)["runtime"] = map[string]any{
+		"backend": map[string]any{
+			"type": k8sapp.BackendTypeE2BSandbox,
+		},
+	}
+	obj.Object["status"].(map[string]any)["e2bSandbox"] = map[string]any{
+		"accessTokenSecretName": secretName,
+	}
+	return obj
+}
+
+// newSecretObject creates an unstructured K8s Secret with a base64-encoded "token" key.
+func newSecretObject(name, namespace, tokenValue string) *unstructured.Unstructured {
+	return &unstructured.Unstructured{
+		Object: map[string]any{
+			"apiVersion": "v1",
+			"kind":       "Secret",
+			"metadata": map[string]any{
+				"name":      name,
+				"namespace": namespace,
+			},
+			"data": map[string]any{
+				"token": base64.StdEncoding.EncodeToString([]byte(tokenValue)),
+			},
+		},
+	}
+}
+
+func TestStateWatcher_GetState_E2BAccessToken(t *testing.T) {
+	t.Parallel()
+
+	fakeClient := newFakeClient()
+	d := newTestDeps(t)
+
+	// Create the secret first.
+	secret := newSecretObject("my-e2b-secret", testNamespace, "my-e2b-token")
+	_, err := fakeClient.Resource(k8sapp.SecretGVR()).Namespace(testNamespace).Create(
+		t.Context(), secret, metav1.CreateOptions{},
+	)
+	require.NoError(t, err)
+
+	// Create E2B app referencing the secret.
+	appObj := newE2BAppObject("my-e2b-app-k8s", "app-e2b", k8sapp.AppActualStateRunning, "my-e2b-secret")
+	_, err = fakeClient.Resource(k8sapp.AppGVR()).Namespace(testNamespace).Create(
+		t.Context(), appObj, metav1.CreateOptions{},
+	)
+	require.NoError(t, err)
+
+	watcher := k8sapp.NewStateWatcher(d, fakeClient, testNamespace)
+
+	var info k8sapp.AppInfo
+	assert.Eventually(t, func() bool {
+		var ok bool
+		info, ok = watcher.GetState(api.AppID("app-e2b"))
+		return ok && info.E2BAccessToken != ""
+	}, 5*time.Second, 50*time.Millisecond)
+
+	assert.Equal(t, "my-e2b-token", info.E2BAccessToken)
+}
+
+func TestStateWatcher_GetState_E2BAccessToken_MissingSecret(t *testing.T) {
+	t.Parallel()
+
+	fakeClient := newFakeClient()
+	d := newTestDeps(t)
+
+	// Create E2B app referencing a non-existent secret.
+	appObj := newE2BAppObject("my-e2b-app-k8s", "app-e2b", k8sapp.AppActualStateRunning, "missing-secret")
+	_, err := fakeClient.Resource(k8sapp.AppGVR()).Namespace(testNamespace).Create(
+		t.Context(), appObj, metav1.CreateOptions{},
+	)
+	require.NoError(t, err)
+
+	watcher := k8sapp.NewStateWatcher(d, fakeClient, testNamespace)
+
+	assert.Eventually(t, func() bool {
+		_, ok := watcher.GetState(api.AppID("app-e2b"))
+		return ok
+	}, 5*time.Second, 50*time.Millisecond)
+
+	info, ok := watcher.GetState(api.AppID("app-e2b"))
+	require.True(t, ok)
+	assert.Empty(t, info.E2BAccessToken)
+}
+
+func TestStateWatcher_GetState_NonE2BApp_NoToken(t *testing.T) {
+	t.Parallel()
+
+	fakeClient := newFakeClient()
+	d := newTestDeps(t)
+
+	// Create a regular (non-E2B) app.
+	appObj := newAppObject("my-app-k8s", "app-regular", k8sapp.AppActualStateRunning)
+	_, err := fakeClient.Resource(k8sapp.AppGVR()).Namespace(testNamespace).Create(
+		t.Context(), appObj, metav1.CreateOptions{},
+	)
+	require.NoError(t, err)
+
+	watcher := k8sapp.NewStateWatcher(d, fakeClient, testNamespace)
+
+	assert.Eventually(t, func() bool {
+		_, ok := watcher.GetState(api.AppID("app-regular"))
+		return ok
+	}, 5*time.Second, 50*time.Millisecond)
+
+	info, ok := watcher.GetState(api.AppID("app-regular"))
+	require.True(t, ok)
+	assert.Empty(t, info.E2BAccessToken)
+}

internal/pkg/service/appsproxy/dependencies/mocked.go

@@ -86,7 +86,8 @@ func newMockedServiceScope(tb testing.TB, ctx context.Context, cfg config.Config
 	// avoiding a race with the watch channel setup.
 	scheme := runtime.NewScheme()
 	fakeClient := k8sfake.NewSimpleDynamicClientWithCustomListKinds(scheme, map[schema.GroupVersionResource]string{
-		k8sapp.AppGVR(): "AppList",
+		k8sapp.AppGVR():    "AppList",
+		k8sapp.SecretGVR(): "SecretList",
 	}, initialK8sObjects...)
 
 	// Validate config