package engine
import (
"context"
"errors"
"fmt"
"io"
"log/slog"
"os"
"path/filepath"
"slices"
"strings"
"sync"
"testing"
"time"
"tangled.org/core/notifier"
"tangled.org/core/spindle/config"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/models"
"tangled.org/core/spindle/observability"
"tangled.org/core/spindle/secrets"
"tangled.org/core/workflow"
)
const testRunToken = "eyJhbGciOiJFUzI1NiJ9.eyJzdWIiOiJkaWQ6cGxjOmJvbHRsZXNzIn0.c2lnbmF0dXJl"
type staticVault struct {
secrets []secrets.UnlockedSecret
}
func (v staticVault) AddSecret(context.Context, secrets.UnlockedSecret) error { return nil }
func (v staticVault) RemoveSecret(context.Context, secrets.Secret[any]) error { return nil }
func (v staticVault) RemoveAllSecrets(context.Context, secrets.RepoIdentifier) error {
return nil
}
func (v staticVault) GetSecretsLocked(context.Context, secrets.RepoIdentifier) ([]secrets.LockedSecret, error) {
return nil, nil
}
func (v staticVault) GetSecretsUnlocked(context.Context, secrets.RepoIdentifier) ([]secrets.UnlockedSecret, error) {
return v.secrets, nil
}
type echoStep struct{}
func (echoStep) Name() string { return "echo" }
func (echoStep) Command() string { return "env" }
func (echoStep) Kind() models.StepKind { return models.StepKindUser }
// echoEngine prints every credential it is handed, the way `set -x` prints the
// environment of a real step. With a gate it holds each workflow at its slot so
// a test can observe what happens while a run waits for capacity.
type slotWait struct {
name string
grant chan struct{}
}
type echoEngine struct {
gate chan slotWait
timeout time.Duration
slotBudget time.Duration
mu sync.Mutex
secrets map[string][]secrets.UnlockedSecret
ran map[string]chan struct{}
ranClosed map[string]bool
order []string
nixCache map[string]*models.NixCacheGrant
}
func newEchoEngine() *echoEngine {
return &echoEngine{
secrets: map[string][]secrets.UnlockedSecret{},
ran: map[string]chan struct{}{},
ranClosed: map[string]bool{},
nixCache: map[string]*models.NixCacheGrant{},
}
}
func (e *echoEngine) InitWorkflow(twf workflow.WorkflowSpec, _ workflow.PipelineSpec) (*models.Workflow, error) {
e.mu.Lock()
defer e.mu.Unlock()
if e.ran[twf.Name] == nil {
e.ran[twf.Name] = make(chan struct{})
}
return &models.Workflow{Name: twf.Name, Steps: []models.Step{echoStep{}}}, nil
}
func (e *echoEngine) AcquireWorkflowSlot(ctx context.Context, _ models.WorkflowId, wf *models.Workflow, _ AcquireMode) (WorkflowSlot, error) {
e.note("slot:" + wf.Name)
if e.slotBudget > 0 {
return budgetSlot(e.slotBudget), nil
}
if e.gate == nil {
return NoopSlot{}, nil
}
grant := make(chan struct{})
select {
case e.gate <- slotWait{name: wf.Name, grant: grant}:
case <-ctx.Done():
return nil, ctx.Err()
}
select {
case <-grant:
return NoopSlot{}, nil
case <-ctx.Done():
return nil, ctx.Err()
}
}
type budgetSlot time.Duration
func (budgetSlot) Release() {}
func (s budgetSlot) RunBudget() time.Duration { return time.Duration(s) }
func (e *echoEngine) SetupWorkflow(_ context.Context, wid models.WorkflowId, wf *models.Workflow, _ models.WorkflowLogger) error {
e.mu.Lock()
defer e.mu.Unlock()
e.nixCache[wid.Name] = wf.NixCache
return nil
}
func (e *echoEngine) nixCacheOf(name string) *models.NixCacheGrant {
e.mu.Lock()
defer e.mu.Unlock()
return e.nixCache[name]
}
func (e *echoEngine) WorkflowTimeout() time.Duration {
if e.timeout == 0 {
return time.Minute
}
return e.timeout
}
func (e *echoEngine) DestroyWorkflow(context.Context, models.WorkflowId) error { return nil }
func (e *echoEngine) RunStep(_ context.Context, wid models.WorkflowId, w *models.Workflow, idx int, unlocked []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error {
e.mu.Lock()
e.secrets[wid.Name] = unlocked
e.order = append(e.order, "steps:"+wid.Name)
if !e.ranClosed[wid.Name] {
e.ranClosed[wid.Name] = true
close(e.ran[wid.Name])
}
e.mu.Unlock()
for _, s := range unlocked {
fmt.Fprintf(wfLogger.DataWriter(idx, "stdout"), "TANGLED_ID_TOKEN=%s\n", s.Value)
}
if _, ok := w.Environment[models.IDTokenEnvVar]; ok {
return errors.New("the id token was merged into the workflow env")
}
return nil
}
func (e *echoEngine) note(event string) {
e.mu.Lock()
defer e.mu.Unlock()
e.order = append(e.order, event)
}
func (e *echoEngine) events() []string {
e.mu.Lock()
defer e.mu.Unlock()
return append([]string(nil), e.order...)
}
func (e *echoEngine) ranWorkflow(name string) <-chan struct{} {
e.mu.Lock()
defer e.mu.Unlock()
if e.ran[name] == nil {
e.ran[name] = make(chan struct{})
}
return e.ran[name]
}
func (e *echoEngine) runSecrets(name string) []secrets.UnlockedSecret {
e.mu.Lock()
defer e.mu.Unlock()
return e.secrets[name]
}
func secretValue(all []secrets.UnlockedSecret, key string) string {
for _, s := range all {
if s.Key == key {
return s.Value
}
}
return ""
}
// startEchoPipeline runs StartWorkflows to completion. It reports errors instead
// of failing a test so it can be driven from another goroutine.
func startEchoPipeline(eng *echoEngine, pipeline *models.Pipeline, names ...string) (logPath string, cleanup func(), err error) {
records := make([]workflow.WorkflowSpec, len(names))
for i, name := range names {
records[i] = workflow.WorkflowSpec{Name: name}
}
return startEchoPipelineRecords(eng, pipeline, records...)
}
func startEchoPipelineRecords(eng *echoEngine, pipeline *models.Pipeline, records ...workflow.WorkflowSpec) (logPath string, cleanup func(), err error) {
ctx := observability.WithMetrics(context.Background(), observability.NewMetrics())
logDir, err := os.MkdirTemp("", "spindle-logs")
if err != nil {
return "", func() {}, err
}
repoDir, err := os.MkdirTemp("", "spindle-repos")
if err != nil {
os.RemoveAll(logDir)
return "", func() {}, err
}
dbDir, err := os.MkdirTemp("", "spindle-db")
if err != nil {
os.RemoveAll(logDir)
os.RemoveAll(repoDir)
return "", func() {}, err
}
cleanup = func() {
os.RemoveAll(logDir)
os.RemoveAll(repoDir)
os.RemoveAll(dbDir)
}
d, err := db.Make(ctx, filepath.Join(dbDir, "spindle.db"))
if err != nil {
cleanup()
return "", func() {}, err
}
cfg := &config.Config{}
cfg.Server.LogDir = logDir
cfg.Server.RepoDir = repoDir
n := notifier.New()
pipelineId := models.PipelineId("3mrkp6iz6os2o")
workflows := map[models.Engine][]models.Workflow{}
for _, record := range records {
wf, err := eng.InitWorkflow(record, workflow.PipelineSpec{})
if err != nil {
cleanup()
return "", func() {}, err
}
if record.Audience != nil {
wf.Audience = *record.Audience
}
if err := d.StatusPending(models.WorkflowId{PipelineId: pipelineId, Name: record.Name}, &n); err != nil {
cleanup()
return "", func() {}, err
}
workflows[eng] = append(workflows[eng], *wf)
}
pipeline.Workflows = workflows
StartWorkflows(
slog.New(slog.NewTextHandler(io.Discard, nil)),
staticVault{secrets: []secrets.UnlockedSecret{{Key: "VAULT_SECRET", Value: "vault-value"}}},
cfg, nil, nil, d, &n, nil, nil, ctx, pipeline, pipelineId,
)
if err := d.Close(); err != nil {
cleanup()
return "", func() {}, err
}
return models.LogFilePath(logDir, models.WorkflowId{PipelineId: pipelineId, Name: records[0].Name}), cleanup, nil
}
func runEchoPipeline(t *testing.T, eng *echoEngine, pipeline *models.Pipeline, names ...string) string {
t.Helper()
logPath, cleanup, err := startEchoPipeline(eng, pipeline, names...)
if err != nil {
cleanup()
t.Fatal(err)
}
t.Cleanup(cleanup)
return logPath
}
func runEchoPipelineRecords(t *testing.T, eng *echoEngine, pipeline *models.Pipeline, records ...workflow.WorkflowSpec) string {
t.Helper()
logPath, cleanup, err := startEchoPipelineRecords(eng, pipeline, records...)
if err != nil {
cleanup()
t.Fatal(err)
}
t.Cleanup(cleanup)
return logPath
}
func audienceRecord(name, audience string) workflow.WorkflowSpec {
return workflow.WorkflowSpec{Name: name, Audience: &audience}
}
func TestWorkflowLogMasksTheRunIDToken(t *testing.T) {
eng := newEchoEngine()
logPath := runEchoPipelineRecords(t, eng, &models.Pipeline{
RepoDid: "did:plc:boltless",
TrustedSource: true,
MintIDToken: func(time.Duration, string) (string, error) { return testRunToken, nil },
}, audienceRecord("build", "vault"))
if value := secretValue(eng.runSecrets("build"), models.IDTokenEnvVar); value != testRunToken {
t.Fatalf("%s handed to the step = %q, want the run's id token", models.IDTokenEnvVar, value)
}
logged, err := os.ReadFile(logPath)
if err != nil {
t.Fatal(err)
}
if strings.Contains(string(logged), testRunToken) {
t.Fatal("the persisted log kept the id token in the clear")
}
if !strings.Contains(string(logged), "***") {
t.Fatalf("the persisted log has no masked entry: %s", logged)
}
}
func TestForkRunsGetNoIDTokenEvenWhenOneWasMinted(t *testing.T) {
eng := newEchoEngine()
logPath := runEchoPipelineRecords(t, eng, &models.Pipeline{
RepoDid: "did:plc:boltless",
TrustedSource: false,
MintIDToken: func(time.Duration, string) (string, error) { return testRunToken, nil },
}, audienceRecord("build", "vault"))
if value := secretValue(eng.runSecrets("build"), models.IDTokenEnvVar); value != "" {
t.Fatalf("%s handed to a fork run = %q, want none", models.IDTokenEnvVar, value)
}
if value := secretValue(eng.runSecrets("build"), "VAULT_SECRET"); value != "" {
t.Fatalf("VAULT_SECRET handed to a fork run = %q, want none", value)
}
logged, err := os.ReadFile(logPath)
if err != nil {
t.Fatal(err)
}
if strings.Contains(string(logged), testRunToken) {
t.Fatalf("a fork run leaked the id token: %s", logged)
}
}
func TestWorkflowProceedsWhenTheIssuerWithholdsTheToken(t *testing.T) {
eng := newEchoEngine()
// what an identity the issuer refuses to sign looks like by the time a
// workflow runs: a minter that yields no token and no error
runEchoPipelineRecords(t, eng, &models.Pipeline{RepoDid: "did:plc:boltless",
TrustedSource: true,
MintIDToken: func(time.Duration, string) (string, error) { return "", nil },
}, audienceRecord("build", "vault"))
if value := secretValue(eng.runSecrets("build"), models.IDTokenEnvVar); value != "" {
t.Fatalf("%s = %q, want none", models.IDTokenEnvVar, value)
}
if value := secretValue(eng.runSecrets("build"), "VAULT_SECRET"); value != "vault-value" {
t.Fatalf("VAULT_SECRET = %q, want the run to keep its other credentials", value)
}
}
func TestIDTokenIsMintedPerWorkflowOnceItHoldsItsSlot(t *testing.T) {
eng := newEchoEngine()
eng.gate = make(chan slotWait)
eng.timeout = 3 * time.Hour
var mu sync.Mutex
var budgets []time.Duration
var audiences []string
minter := func(budget time.Duration, audience string) (string, error) {
mu.Lock()
defer mu.Unlock()
budgets = append(budgets, budget)
audiences = append(audiences, audience)
return fmt.Sprintf("%s.%d", testRunToken, len(budgets)), nil
}
type result struct {
cleanup func()
err error
}
done := make(chan result, 1)
go func() {
_, cleanup, err := startEchoPipelineRecords(eng, &models.Pipeline{
RepoDid: "did:plc:boltless",
TrustedSource: true,
MintIDToken: minter,
}, audienceRecord("first", "aud-first"), audienceRecord("second", "aud-second"))
done <- result{cleanup: cleanup, err: err}
}()
mintedCount := func() int {
mu.Lock()
defer mu.Unlock()
return len(budgets)
}
// each workflow is released only after the test has seen that no token was
// minted for it while it waited
waiting := make([]slotWait, 0, 2)
for range 2 {
select {
case w := <-eng.gate:
if got := mintedCount(); got != len(waiting) {
t.Fatalf("minted %d tokens while %d workflows held a slot, before releasing %q: a token must not spend its lifetime waiting for capacity", got, len(waiting), w.name)
}
waiting = append(waiting, w)
close(w.grant)
case <-time.After(20 * time.Second):
t.Fatal("a workflow never waited for its slot")
}
}
tokens := map[string]string{}
for _, name := range []string{"first", "second"} {
select {
case <-eng.ranWorkflow(name):
case <-time.After(20 * time.Second):
t.Fatalf("%s never ran its steps", name)
}
if token := secretValue(eng.runSecrets(name), models.IDTokenEnvVar); token != "" {
tokens[name] = token
}
}
res := <-done
if res.cleanup != nil {
defer res.cleanup()
}
if res.err != nil {
t.Fatal(res.err)
}
if got := mintedCount(); got != 2 {
t.Fatalf("minted %d tokens for 2 workflows, want one each", got)
}
for _, budget := range budgets {
if budget != 3*time.Hour {
t.Fatalf("mint budget = %s, want the workflow time budget 3h", budget)
}
}
if len(tokens) != 2 {
t.Fatalf("workflows carried tokens %v, want one freshly minted token each", tokens)
}
if tokens["first"] == tokens["second"] {
t.Fatalf("both workflows carry %q, want a fresh token per workflow", tokens["first"])
}
if !slices.Contains(audiences, "aud-first") || !slices.Contains(audiences, "aud-second") {
t.Fatalf("minted for audiences %v, want each workflow's own", audiences)
}
}
func TestIDTokenLivesAsLongAsTheSlotsRunBudget(t *testing.T) {
for _, tt := range []struct {
name string
slotBudget time.Duration
wantMinted time.Duration
}{
{"shorter than the engine timeout", 15 * time.Minute, 15 * time.Minute},
{"longer than the engine timeout", 2 * time.Hour, time.Hour},
} {
t.Run(tt.name, func(t *testing.T) {
eng := newEchoEngine()
eng.timeout = time.Hour
eng.slotBudget = tt.slotBudget
var minted time.Duration
runEchoPipelineRecords(t, eng, &models.Pipeline{
RepoDid: "did:plc:boltless",
TrustedSource: true,
MintIDToken: func(budget time.Duration, _ string) (string, error) {
minted = budget
return testRunToken, nil
},
}, audienceRecord("build", "vault"))
if minted != tt.wantMinted {
t.Fatalf("id token minted for %s, want %s", minted, tt.wantMinted)
}
})
}
}
func TestWorkflowsWithoutAnAudienceGetNoIDToken(t *testing.T) {
eng := newEchoEngine()
runEchoPipeline(t, eng, &models.Pipeline{
RepoDid: "did:plc:boltless",
TrustedSource: true,
MintIDToken: func(time.Duration, string) (string, error) {
t.Error("the minter ran for a workflow that declared no audience")
return testRunToken, nil
},
}, "build")
if value := secretValue(eng.runSecrets("build"), models.IDTokenEnvVar); value != "" {
t.Fatalf("%s = %q, want none: the audience declaration is the trigger", models.IDTokenEnvVar, value)
}
if value := secretValue(eng.runSecrets("build"), "VAULT_SECRET"); value != "vault-value" {
t.Fatalf("VAULT_SECRET = %q, want the run to keep its other credentials", value)
}
}
func nixCacheGrant() *models.NixCacheGrant {
return &models.NixCacheGrant{Cache: "plc-boltless", PublicKey: "plc-boltless:AAAA=", PullToken: "pull", PushToken: "push"}
}
func TestNixCacheGrantLivesAsLongAsTheSlotsRunBudget(t *testing.T) {
eng := newEchoEngine()
eng.timeout = time.Hour
eng.slotBudget = 15 * time.Minute
var budget time.Duration
var run models.WorkflowId
runEchoPipelineRecords(t, eng, &models.Pipeline{
RepoDid: "did:plc:boltless",
TrustedSource: true,
GrantNixCache: func(_ context.Context, wid models.WorkflowId, b time.Duration) (*models.NixCacheGrant, error) {
run, budget = wid, b
return nixCacheGrant(), nil
},
}, audienceRecord("build", "vault"))
got := eng.nixCacheOf("build")
if got == nil || got.Cache != "plc-boltless" || got.PullToken != "pull" || got.PushToken != "push" {
t.Fatalf("trusted run's cache = %+v, want pull and push", got)
}
if budget != 15*time.Minute {
t.Fatalf("grant asked for a %s budget, want the slot's 15m", budget)
}
if run.Name != "build" || run.PipelineId == "" {
t.Errorf("grant asked for run %+v, want the build workflow's id", run)
}
}
func TestWorkflowRunsWithoutACacheWhenTheGrantFails(t *testing.T) {
eng := newEchoEngine()
runEchoPipelineRecords(t, eng, &models.Pipeline{
RepoDid: "did:plc:boltless",
TrustedSource: true,
GrantNixCache: func(context.Context, models.WorkflowId, time.Duration) (*models.NixCacheGrant, error) {
return nil, errors.New("attic is down")
},
}, audienceRecord("build", "vault"))
if got := eng.nixCacheOf("build"); got != nil {
t.Fatalf("cache = %+v, want none", got)
}
select {
case <-eng.ranWorkflow("build"):
default:
t.Fatal("the workflow never ran its steps")
}
}