feat(api): clean workflow run secrets

engine/api/api.go

@@ -66,7 +66,8 @@ type Configuration struct {
 	} `toml:"url" comment:"#####################\n CDS URLs Settings \n####################" json:"url"`
 	HTTP    service.HTTPRouterConfiguration `toml:"http" json:"http"`
 	Secrets struct {
-		SkipProjectSecretsOnRegion []string `toml:"skipProjectSecretsOnRegion" json:"-"`
+		SkipProjectSecretsOnRegion []string `toml:"skipProjectSecretsOnRegion" json:"skipProjectSecretsOnRegion" comment:"For given region, CDS will not automatically inject project's secrets when running a job."`
+		SnapshotRetentionDelay     int64    `toml:"snapshotRetentionDelay" json:"snapshotRetentionDelay" comment:"Retention delay for workflow run secrets snapshot (in days), set to 0 will keep secrets until workflow run deletion. Removing secrets will activate the read only mode on a workflow run."`
 	} `toml:"secrets" json:"secrets"`
 	Database database.DBConfiguration `toml:"database" comment:"################################\n Postgresql Database settings \n###############################" json:"database"`
 	Cache    struct {
@@ -748,6 +749,11 @@ func (a *API) Serve(ctx context.Context) error {
 	a.GoRoutines.RunWithRestart(ctx, "workflow.ResyncWorkflowRunResultsRoutine", func(ctx context.Context) {
 		workflow.ResyncWorkflowRunResultsRoutine(ctx, a.mustDB)
 	})
+	if a.Config.Secrets.SnapshotRetentionDelay > 0 {
+		a.GoRoutines.RunWithRestart(ctx, "workflow.CleanSecretsSnapshot", func(ctx context.Context) {
+			a.cleanWorkflowRunSecrets(ctx, time.Minute*10)
+		})
+	}
 
 	log.Info(ctx, "Bootstrapping database...")
 	defaultValues := sdk.DefaultValues{

engine/api/workflow/dao_run.go

@@ -353,27 +353,27 @@ func LoadRunsSummaries(ctx context.Context, db gorp.SqlExecutor, projectkey, wor
 				SELECT workflow.id FROM workflow
 				JOIN project ON project.id = workflow.project_id
 				WHERE workflow.name = $2 AND project.projectkey = $1
-			), 
+			),
 			runs as (
-				SELECT %s 
+				SELECT %s
 				FROM workflow_run wr
 				JOIN workflowID ON workflowID.id = wr.workflow_id
 				WHERE wr.to_delete = false
 			),
 			tags as (
-				SELECT workflow_run_id, tag || '=' || value "all_tags" 
+				SELECT workflow_run_id, tag || '=' || value "all_tags"
 				FROM workflow_run_tag
 				JOIN runs ON runs.id = workflow_run_id
 			),
 			aggTags as (
-				SELECT workflow_run_id, string_agg(all_tags, ',') as tags 
+				SELECT workflow_run_id, string_agg(all_tags, ',') as tags
 				FROM tags
 				GROUP BY workflow_run_id
 			)
 			SELECT runs.*
 			FROM runs
 			JOIN aggTags ON aggTags.workflow_run_id = runs.id
-			WHERE string_to_array($5, ',') <@ string_to_array(aggTags.tags, ',') 
+			WHERE string_to_array($5, ',') <@ string_to_array(aggTags.tags, ',')
 			ORDER BY runs.start DESC OFFSET $4 LIMIT $3`, selectedColumn)
 		var tags []string
 		for k, v := range tagFilter {
@@ -1095,3 +1095,28 @@ func stopRunsBlocked(ctx context.Context, db *gorp.DbMap) error {
 	}
 	return nil
 }
+
+// LoadRunsIDsCreatedBefore returns the first 100 workflow runs created before given date.
+func LoadRunsIDsCreatedBefore(ctx context.Context, db gorp.SqlExecutor, date time.Time) ([]int64, error) {
+	var ids []int64
+	query := `
+    SELECT id
+    FROM workflow_run
+		WHERE read_only = false AND start < $1
+		ORDER BY start ASC
+    LIMIT 100
+  `
+	if _, err := db.Select(&ids, query, date); err != nil {
+		return nil, sdk.WithStack(err)
+	}
+	return ids, nil
+}
+
+// SetRunReadOnly set read only flag of a workflow run, this run cannot be restarted anymore.
+func SetRunReadOnlyByID(ctx context.Context, db gorpmapper.SqlExecutorWithTx, workflowRunID int64) error {
+	query := `UPDATE workflow_run SET read_only = true WHERE id = $1`
+	if _, err := db.Exec(query, workflowRunID); err != nil {
+		return sdk.WrapError(err, "unable to set read only for workflow run with id %d", workflowRunID)
+	}
+	return nil
+}

engine/api/workflow/dao_run_secret.go

@@ -49,3 +49,20 @@ func loadRunSecretWithDecryption(ctx context.Context, db gorp.SqlExecutor, runID
 	}
 	return secrets, nil
 }
+
+func CountRunSecretsByWorkflowRunID(ctx context.Context, db gorp.SqlExecutor, workflowRunID int64) (int64, error) {
+	query := `SELECT COUNT(1) FROM workflow_run_secret WHERE workflow_run_id = $1`
+	count, err := db.SelectInt(query, workflowRunID)
+	if err != nil {
+		return 0, sdk.WrapError(err, "unable to count workflow run secret for workflow run id %d", workflowRunID)
+	}
+	return count, nil
+}
+
+func DeleteRunSecretsByWorkflowRunID(ctx context.Context, db gorpmapper.SqlExecutorWithTx, workflowRunID int64) error {
+	query := `DELETE FROM workflow_run_secret WHERE workflow_run_id = $1`
+	if _, err := db.Exec(query, workflowRunID); err != nil {
+		return sdk.WrapError(err, "unable to delete workflow run secret for workflow run id %d", workflowRunID)
+	}
+	return nil
+}

engine/api/workflow_run_secrets.go

@@ -0,0 +1,52 @@
+package api
+
+import (
+	"context"
+	"time"
+
+	"github.com/go-gorp/gorp"
+	"github.com/rockbears/log"
+
+	"github.com/ovh/cds/engine/api/workflow"
+	"github.com/ovh/cds/sdk"
+)
+
+func (api *API) cleanWorkflowRunSecrets(ctx context.Context, delay time.Duration) {
+	// Load workflow run older than now - snapshot retention delay
+	maxRetentionDate := time.Now().Add(-time.Hour * time.Duration(24*api.Config.Secrets.SnapshotRetentionDelay))
+
+	db := api.mustDB()
+
+	ticker := time.NewTicker(delay)
+
+	for range ticker.C {
+		runIDs, err := workflow.LoadRunsIDsCreatedBefore(ctx, db, maxRetentionDate)
+		if err != nil {
+			log.ErrorWithStackTrace(ctx, err)
+			continue
+		}
+		for _, id := range runIDs {
+			if err := api.cleanWorkflowRunSecretsForRun(ctx, db, id); err != nil {
+				log.ErrorWithStackTrace(ctx, err)
+			}
+		}
+	}
+}
+
+func (api *API) cleanWorkflowRunSecretsForRun(ctx context.Context, db *gorp.DbMap, workflowRunID int64) error {
+	tx, err := db.Begin()
+	if err != nil {
+		return sdk.WithStack(err)
+	}
+	defer tx.Rollback() // nolint
+	if err := workflow.SetRunReadOnlyByID(ctx, tx, workflowRunID); err != nil {
+		return sdk.WithStack(err)
+	}
+	if err := workflow.DeleteRunSecretsByWorkflowRunID(ctx, tx, workflowRunID); err != nil {
+		return sdk.WithStack(err)
+	}
+	if err := tx.Commit(); err != nil {
+		return sdk.WithStack(err)
+	}
+	return nil
+}

engine/api/workflow_run_secrets_test.go

@@ -0,0 +1,65 @@
+package api
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/ovh/cds/engine/api/authentication"
+	"github.com/ovh/cds/engine/api/project"
+	"github.com/ovh/cds/engine/api/test/assets"
+	"github.com/ovh/cds/engine/api/workflow"
+	"github.com/ovh/cds/sdk"
+)
+
+func Test_cleanSecretsSnapshotForRun(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	api, db, _ := newTestAPI(t)
+
+	u, _ := assets.InsertAdminUser(t, db)
+	consumer, _ := authentication.LoadConsumerByTypeAndUserID(context.TODO(), db, sdk.ConsumerLocal, u.ID, authentication.LoadConsumerOptions.WithAuthentifiedUser)
+	projectKey := sdk.RandomString(10)
+	p := assets.InsertTestProject(t, db, api.Cache, projectKey, projectKey)
+
+	require.NoError(t, project.InsertVariable(db, p.ID, &sdk.ProjectVariable{
+		Type:  sdk.SecretVariable,
+		Name:  "my-secret",
+		Value: "my-value",
+	}, u))
+
+	w := assets.InsertTestWorkflow(t, db, api.Cache, p, sdk.RandomString(10))
+	wr, err := workflow.CreateRun(db.DbMap, w, sdk.WorkflowRunPostHandlerOption{
+		Hook: &sdk.WorkflowNodeRunHookEvent{},
+	})
+	require.NoError(t, err)
+	api.initWorkflowRun(ctx, p.Key, w, wr, sdk.WorkflowRunPostHandlerOption{
+		Manual:         &sdk.WorkflowNodeRunManual{},
+		AuthConsumerID: consumer.ID,
+	})
+
+	runIDs, err := workflow.LoadRunsIDsCreatedBefore(ctx, db, time.Now())
+	require.NoError(t, err)
+	require.Contains(t, runIDs, wr.ID)
+
+	runIDs, err = workflow.LoadRunsIDsCreatedBefore(ctx, db, wr.Start)
+	require.NoError(t, err)
+	require.NotContains(t, runIDs, wr.ID)
+
+	count, err := workflow.CountRunSecretsByWorkflowRunID(ctx, db, wr.ID)
+	require.NoError(t, err)
+	require.Equal(t, int64(1), count)
+
+	require.NoError(t, api.cleanWorkflowRunSecretsForRun(ctx, db.DbMap, wr.ID))
+
+	result, err := workflow.LoadRunByID(ctx, db, wr.ID, workflow.LoadRunOptions{})
+	require.NoError(t, err)
+	require.True(t, result.ReadOnly)
+
+	count, err = workflow.CountRunSecretsByWorkflowRunID(ctx, db, wr.ID)
+	require.NoError(t, err)
+	require.Equal(t, int64(0), count)
+}

engine/sql/api/251_clean_workflow_run_secrets.sql

@@ -0,0 +1,5 @@
+-- +migrate Up
+SELECT create_index('workflow_run', 'idx_workflow_run_start', 'read_only,start');
+
+-- +migrate Down
+DROP INDEX idx_workflow_run_start;