From 943bf0945f26333ae157b334a97a16c373fc06ed Mon Sep 17 00:00:00 2001 From: Giovanni Ferri Date: Wed, 9 Sep 2026 21:22:25 +0100 Subject: [PATCH 1/2] feat(agent): make runner handoff non-persisting --- api/v1alpha1/impvm_types.go | 15 ++++ api/v1alpha1/zz_generated.deepcopy.go | 5 ++ config/crd/bases/imp.dev_impvms.yaml | 16 ++++ internal/agent/runnerlaunch/launcher.go | 88 +++++++++++++++----- internal/agent/runnerlaunch/launcher_test.go | 40 +++++++++ 5 files changed, 144 insertions(+), 20 deletions(-) diff --git a/api/v1alpha1/impvm_types.go b/api/v1alpha1/impvm_types.go index 785f6c6..23f4516 100644 --- a/api/v1alpha1/impvm_types.go +++ b/api/v1alpha1/impvm_types.go @@ -224,6 +224,21 @@ type ImpVMStatus struct { // +optional CiliumEndpointID int64 `json:"ciliumEndpointID,omitempty"` + // RunnerHandoffAccepted records that the one-time runner payload was handed + // to the guest Exec boundary. It is also the durable ambiguity marker: an + // agent restart must not replay a payload after this is set. + // +optional + RunnerHandoffAccepted bool `json:"runnerHandoffAccepted,omitempty"` + + // RunnerExitCode is the non-secret exit result returned by the guest runner. + // +optional + RunnerExitCode *int32 `json:"runnerExitCode,omitempty"` + + // RunnerFailureReason is a bounded, non-secret handoff failure summary. + // +optional + // +kubebuilder:validation:MaxLength=256 + RunnerFailureReason string `json:"runnerFailureReason,omitempty"` + // Conditions follow the standard k8s condition convention. // +optional // +listType=map diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 5d8cdf7..98df6cf 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -1272,6 +1272,11 @@ func (in *ImpVMStatus) DeepCopyInto(out *ImpVMStatus) { in, out := &in.ExhaustedAt, &out.ExhaustedAt *out = (*in).DeepCopy() } + if in.RunnerExitCode != nil { + in, out := &in.RunnerExitCode, &out.RunnerExitCode + *out = new(int32) + **out = **in + } if in.Conditions != nil { in, out := &in.Conditions, &out.Conditions *out = make([]v1.Condition, len(*in)) diff --git a/config/crd/bases/imp.dev_impvms.yaml b/config/crd/bases/imp.dev_impvms.yaml index 22abde2..a81ee1c 100644 --- a/config/crd/bases/imp.dev_impvms.yaml +++ b/config/crd/bases/imp.dev_impvms.yaml @@ -789,6 +789,22 @@ spec: has been restarted. format: int32 type: integer + runnerExitCode: + description: RunnerExitCode is the non-secret exit result returned + by the guest runner. + format: int32 + type: integer + runnerFailureReason: + description: RunnerFailureReason is a bounded, non-secret handoff + failure summary. + maxLength: 256 + type: string + runnerHandoffAccepted: + description: |- + RunnerHandoffAccepted records that the one-time runner payload was handed + to the guest Exec boundary. It is also the durable ambiguity marker: an + agent restart must not replay a payload after this is set. + type: boolean runningAt: description: RunningAt is the time the VM first reached Running phase. format: date-time diff --git a/internal/agent/runnerlaunch/launcher.go b/internal/agent/runnerlaunch/launcher.go index 1bc3bc9..a9ae00e 100644 --- a/internal/agent/runnerlaunch/launcher.go +++ b/internal/agent/runnerlaunch/launcher.go @@ -7,9 +7,8 @@ // referenced by ImpVM.spec.runnerConfigSecret. After the VM reaches Running, // the agent dials the guest agent over VSOCK, sends an Exec for // /usr/local/bin/runner with IMP_GITHUB_JITCONFIG in the command environment, -// and deletes the Secret — the payload is one-time and must not outlive -// handoff. If the Secret is already gone (agent restart after a successful -// handoff) the launcher is a no-op. +// and deletes the Secret only after the runner exits. The payload is one-time +// and must not be replayed after an ambiguous handoff. package runnerlaunch import ( @@ -40,6 +39,15 @@ var ( ) const maxRunnerStderrLogBytes = 1024 +const maxRunnerFailureReasonBytes = 256 + +func boundedFailureReason(reason, payload string) string { + reason = sanitizeRunnerStderr(reason, payload) + if len(reason) <= maxRunnerFailureReasonBytes { + return reason + } + return reason[:maxRunnerFailureReasonBytes-3] + "..." +} // sanitizeRunnerStderr keeps enough guest diagnostics to identify startup // failures without copying one-time credentials into controller logs. @@ -127,14 +135,23 @@ func (l *Launcher) Run(ctx context.Context, vm *impv1alpha1.ImpVM) Result { if secretName == "" { return Result{} } - launchKey := vm.Namespace + "/" + vm.Name + launchKey := vm.Namespace + "/" + vm.Name + "/" + string(vm.UID) if _, loaded := launches.LoadOrStore(launchKey, struct{}{}); loaded { return Result{} } defer launches.Delete(launchKey) - // Read the payload. Missing Secret after an agent restart means the - // handoff already happened; treat as done. + // The accepted marker is written before Exec because the unary RPC has no + // acknowledgement boundary: it returns only when the guest process exits. + // A restart after this write is therefore ambiguous and must not replay. + if vm.Status.RunnerExitCode != nil || vm.Status.RunnerHandoffAccepted { + if vm.Status.RunnerExitCode != nil { + return Result{HandoffDone: true, ExitCode: *vm.Status.RunnerExitCode} + } + return Result{Err: fmt.Errorf("runner handoff for VM UID %s is ambiguous; refusing replay", vm.UID)} + } + + // Read the payload through the uncached APIReader by exact namespace/name. var secret corev1.Secret reader := l.Reader if reader == nil { @@ -144,11 +161,9 @@ func (l *Launcher) Run(ctx context.Context, vm *impv1alpha1.ImpVM) Result { return Result{Err: fmt.Errorf("runner launcher requires Kubernetes reader and client")} } err := reader.Get(ctx, client.ObjectKey{Namespace: vm.Namespace, Name: secretName}, &secret) - switch { - case apierrors.IsNotFound(err): - log.Info("runner config Secret already deleted; handoff previously completed", "secret", secretName) - return Result{HandoffDone: true} - case err != nil: + if err != nil { + // NotFound is intentionally not interpreted as a completed handoff: it + // may be a cleanup race or an agent restart in an unknown state. return Result{Err: fmt.Errorf("get runner config Secret %s: %w", secretName, err)} } config, err := runner.JITConfigFromSecretValue( @@ -157,9 +172,14 @@ func (l *Launcher) Run(ctx context.Context, vm *impv1alpha1.ImpVM) Result { secret.Annotations[runner.JITConfigRunnerAnnotation], ) if err != nil { + l.recordFailure(ctx, vm, err.Error(), "") return Result{Err: fmt.Errorf("decode runner config Secret %s: %w", secretName, err)} } + if err := l.recordAccepted(ctx, vm); err != nil { + return Result{Err: fmt.Errorf("record runner handoff acceptance: %w", err)} + } + // Dial the guest agent, retrying while the guest finishes booting. conn, err := l.dialWithRetry(ctx) if err != nil { @@ -168,7 +188,7 @@ func (l *Launcher) Run(ctx context.Context, vm *impv1alpha1.ImpVM) Result { defer conn.Close() //nolint:errcheck guest := pb.NewGuestAgentClient(conn) - // Hand off. env keeps the payload out of argv (not visible in guest ps). + // Hand off. Env keeps the payload out of argv (not visible in guest ps). hCtx, cancel := context.WithTimeout(ctx, handoffTimeout) defer cancel() resp, err := guest.Exec(hCtx, &pb.ExecRequest{ @@ -180,22 +200,27 @@ func (l *Launcher) Run(ctx context.Context, vm *impv1alpha1.ImpVM) Result { TimeoutSeconds: int32(handoffTimeout / time.Second), }) if err != nil { + l.recordFailure(ctx, vm, err.Error(), config.EncodedConfig) return Result{Err: fmt.Errorf("launch runner in guest: %w", err)} } - // One-time payload: delete immediately after successful delivery. - delErr := l.Client.Delete(ctx, &secret) - if delErr != nil && !apierrors.IsNotFound(delErr) { - // Non-fatal: expiry and pool cleanup will collect it, but surface it. - log.Error("failed to delete runner config Secret after handoff", - "secret", secretName, "err", delErr) - } - if resp.ExitCode != 0 { if stderr := runnerFailureStderr(resp.ExitCode, resp.Stderr, config.EncodedConfig); stderr != "" { log.Warn("runner emitted stderr", "stderr", stderr) } } + if err := l.recordOutcome(ctx, vm, resp.ExitCode, runnerFailureStderr(resp.ExitCode, resp.Stderr, config.EncodedConfig)); err != nil { + log.Error("failed to persist runner handoff outcome", "err", err) + } + + // The Secret remains until unary Exec returns, which is the runner-exit + // boundary. Delete-NotFound is an idempotent cleanup success. + delErr := l.Client.Delete(ctx, &secret) + if delErr != nil && !apierrors.IsNotFound(delErr) { + log.Error("failed to delete runner config Secret after runner exit", + "secret", secretName, "err", delErr) + } + log.Info("runner exited", "exit_code", resp.ExitCode, "secret", secretName) if resp.ExitCode != 0 { return Result{HandoffDone: true, ExitCode: resp.ExitCode, @@ -204,6 +229,29 @@ func (l *Launcher) Run(ctx context.Context, vm *impv1alpha1.ImpVM) Result { return Result{HandoffDone: true, ExitCode: resp.ExitCode} } +func (l *Launcher) recordAccepted(ctx context.Context, vm *impv1alpha1.ImpVM) error { + base := vm.DeepCopy() + vm.Status.RunnerHandoffAccepted = true + return l.Client.Status().Patch(ctx, vm, client.MergeFrom(base)) +} + +func (l *Launcher) recordFailure(ctx context.Context, vm *impv1alpha1.ImpVM, reason, payload string) { + base := vm.DeepCopy() + vm.Status.RunnerFailureReason = boundedFailureReason(reason, payload) + if err := l.Client.Status().Patch(ctx, vm, client.MergeFrom(base)); err != nil { + if l.Log != nil { + l.Log.Error("failed to persist runner handoff failure", "err", err) + } + } +} + +func (l *Launcher) recordOutcome(ctx context.Context, vm *impv1alpha1.ImpVM, exitCode int32, reason string) error { + base := vm.DeepCopy() + vm.Status.RunnerExitCode = &exitCode + vm.Status.RunnerFailureReason = boundedFailureReason(reason, "") + return l.Client.Status().Patch(ctx, vm, client.MergeFrom(base)) +} + // dialWithRetry waits for the lazy gRPC client to establish a real HTTP/2 // connection. Creating a grpc.ClientConn alone does not call the VSOCK dialer. func (l *Launcher) dialWithRetry(ctx context.Context) (*grpc.ClientConn, error) { diff --git a/internal/agent/runnerlaunch/launcher_test.go b/internal/agent/runnerlaunch/launcher_test.go index 4cbe389..7a29059 100644 --- a/internal/agent/runnerlaunch/launcher_test.go +++ b/internal/agent/runnerlaunch/launcher_test.go @@ -9,6 +9,13 @@ import ( "strings" "testing" "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + impv1alpha1 "github.com/syscode-labs/imp/api/v1alpha1" ) func TestDialWithRetryWaitsForRealConnection(t *testing.T) { @@ -63,3 +70,36 @@ func TestSanitizeRunnerStderrSuccessDoesNotLog(t *testing.T) { t.Fatalf("empty failure stderr = %q, want empty", got) } } + +func TestRunnerOutcomeIsNonSecretAndRestartSafe(t *testing.T) { + scheme := runtime.NewScheme() + if err := impv1alpha1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + vm := &impv1alpha1.ImpVM{ + ObjectMeta: metav1.ObjectMeta{Name: "vm", Namespace: "ns", UID: "vm-uid"}, + } + c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(vm).Build() + launcher := &Launcher{Client: c} + if err := launcher.recordAccepted(context.Background(), vm); err != nil { + t.Fatalf("recordAccepted: %v", err) + } + if !vm.Status.RunnerHandoffAccepted { + t.Fatal("handoff acceptance was not recorded") + } + launcher.recordFailure(context.Background(), vm, "runner failed token=secret-value", "secret-value") + if strings.Contains(vm.Status.RunnerFailureReason, "secret-value") || strings.Contains(vm.Status.RunnerFailureReason, "token=") { + t.Fatalf("failure reason leaked payload: %q", vm.Status.RunnerFailureReason) + } + code := int32(17) + if err := launcher.recordOutcome(context.Background(), vm, code, "runner exited"); err != nil { + t.Fatalf("recordOutcome: %v", err) + } + stored := &impv1alpha1.ImpVM{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(vm), stored); err != nil { + t.Fatalf("get stored VM: %v", err) + } + if stored.Status.RunnerExitCode == nil || *stored.Status.RunnerExitCode != code { + t.Fatalf("stored exit code = %v, want %d", stored.Status.RunnerExitCode, code) + } +} From 45e8b7ce2a950458d8212ab61d4af3187c6791ff Mon Sep 17 00:00:00 2001 From: Giovanni Ferri Date: Wed, 9 Sep 2026 23:20:24 +0100 Subject: [PATCH 2/2] fix(agent): harden non-persisting runner handoff --- internal/agent/runnerlaunch/launcher.go | 8 ++++++- internal/agent/runnerlaunch/launcher_test.go | 22 +++++++++++++++++++- 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/internal/agent/runnerlaunch/launcher.go b/internal/agent/runnerlaunch/launcher.go index a9ae00e..ca02164 100644 --- a/internal/agent/runnerlaunch/launcher.go +++ b/internal/agent/runnerlaunch/launcher.go @@ -46,7 +46,11 @@ func boundedFailureReason(reason, payload string) string { if len(reason) <= maxRunnerFailureReasonBytes { return reason } - return reason[:maxRunnerFailureReasonBytes-3] + "..." + limit := maxRunnerFailureReasonBytes - 3 + for limit > 0 && (reason[limit]&0xc0) == 0x80 { + limit-- + } + return reason[:limit] + "..." } // sanitizeRunnerStderr keeps enough guest diagnostics to identify startup @@ -162,6 +166,7 @@ func (l *Launcher) Run(ctx context.Context, vm *impv1alpha1.ImpVM) Result { } err := reader.Get(ctx, client.ObjectKey{Namespace: vm.Namespace, Name: secretName}, &secret) if err != nil { + l.recordFailure(ctx, vm, err.Error(), "") // NotFound is intentionally not interpreted as a completed handoff: it // may be a cleanup race or an agent restart in an unknown state. return Result{Err: fmt.Errorf("get runner config Secret %s: %w", secretName, err)} @@ -183,6 +188,7 @@ func (l *Launcher) Run(ctx context.Context, vm *impv1alpha1.ImpVM) Result { // Dial the guest agent, retrying while the guest finishes booting. conn, err := l.dialWithRetry(ctx) if err != nil { + l.recordFailure(ctx, vm, err.Error(), config.EncodedConfig) return Result{Err: fmt.Errorf("dial guest agent: %w", err)} } defer conn.Close() //nolint:errcheck diff --git a/internal/agent/runnerlaunch/launcher_test.go b/internal/agent/runnerlaunch/launcher_test.go index 7a29059..74f34c1 100644 --- a/internal/agent/runnerlaunch/launcher_test.go +++ b/internal/agent/runnerlaunch/launcher_test.go @@ -79,7 +79,7 @@ func TestRunnerOutcomeIsNonSecretAndRestartSafe(t *testing.T) { vm := &impv1alpha1.ImpVM{ ObjectMeta: metav1.ObjectMeta{Name: "vm", Namespace: "ns", UID: "vm-uid"}, } - c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(vm).Build() + c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(vm).WithStatusSubresource(vm).Build() launcher := &Launcher{Client: c} if err := launcher.recordAccepted(context.Background(), vm); err != nil { t.Fatalf("recordAccepted: %v", err) @@ -99,7 +99,27 @@ func TestRunnerOutcomeIsNonSecretAndRestartSafe(t *testing.T) { if err := c.Get(context.Background(), client.ObjectKeyFromObject(vm), stored); err != nil { t.Fatalf("get stored VM: %v", err) } + if !stored.Status.RunnerHandoffAccepted { + t.Fatal("stored VM lost the durable handoff marker") + } if stored.Status.RunnerExitCode == nil || *stored.Status.RunnerExitCode != code { t.Fatalf("stored exit code = %v, want %d", stored.Status.RunnerExitCode, code) } } + +func TestRunnerHandoffMarkerRefusesReplay(t *testing.T) { + scheme := runtime.NewScheme() + if err := impv1alpha1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + vm := &impv1alpha1.ImpVM{ + ObjectMeta: metav1.ObjectMeta{Name: "vm", Namespace: "ns", UID: "vm-uid"}, + Spec: impv1alpha1.ImpVMSpec{RunnerConfigSecret: "jitconfig"}, + Status: impv1alpha1.ImpVMStatus{RunnerHandoffAccepted: true}, + } + c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(vm).WithStatusSubresource(vm).Build() + result := (&Launcher{Client: c}).Run(context.Background(), vm) + if result.Err == nil || !strings.Contains(result.Err.Error(), "refusing replay") { + t.Fatalf("Run() error = %v, want durable replay refusal", result.Err) + } +}