diff --git a/internal/cli/namespace_test.go b/internal/cli/namespace_test.go index c24a552..cbc3751 100644 --- a/internal/cli/namespace_test.go +++ b/internal/cli/namespace_test.go @@ -45,6 +45,10 @@ func resetCommandFlags() { restoreDryRun = false collectorRoot = "/" collectorDryRun = false + txnApplyForce, txnApplyAckRisk, txnApplyYes = false, false, false + txnApplyNamespace = "" + txnApplyTimeout = 5 * time.Minute + txnApplyLead, txnRollbackLead = "", "" resetNSFlags() // Reset per-command output writers so tests that polluted them // (e.g. daemon tests calling cmd.SetOut(&buf)) don't leak into diff --git a/internal/cli/txn.go b/internal/cli/txn.go new file mode 100644 index 0000000..e285874 --- /dev/null +++ b/internal/cli/txn.go @@ -0,0 +1,269 @@ +// Package cli: txn.go implements the `orca txn` subcommand family +// (P10a, v0.11; REQ-075, REQ-079; gates C-09, C-23). Subcommands: +// +// orca txn apply --lead [--force --i-understand-the-risk | --namespace ] [--timeout 5m] +// orca txn list +// orca txn show +// orca txn rollback --lead +// +// `apply` runs an already-staged txn on the lead peer via the txn +// package. Cluster-wide txns (no --namespace) require --force + +// --i-understand-the-risk (or --yes); namespace-scoped txns only +// touch the given namespace (C-23). +package cli + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "time" + + "github.com/spf13/cobra" + + "git.cloudinit.dev/coreci/orca/internal/certpaths" + "git.cloudinit.dev/coreci/orca/internal/paths" + "git.cloudinit.dev/coreci/orca/internal/sshpush" + "git.cloudinit.dev/coreci/orca/internal/txn" +) + +var ( + txnApplyForce bool + txnApplyAckRisk bool + txnApplyYes bool + txnApplyNamespace string + txnApplyTimeout time.Duration + txnApplyLead string + txnRollbackLead string +) + +// txnTransport is the SSH-push surface the txn CLI needs. *sshpush.Transport +// satisfies it; tests substitute a mock (same pattern as drain.go / +// logs.go). +type txnTransport interface { + WriteFileIdempotent(ctx context.Context, peer string, path string, content []byte, mode os.FileMode) (bool, error) + Exec(ctx context.Context, peer string, cmd string) ([]byte, error) +} + +// txnTransportOverride is the package-level seam. When non-nil it +// replaces the production transport; tests set it and restore nil. +var txnTransportOverride txnTransport + +func txnTransportFromCtx() (txnTransport, error) { + if txnTransportOverride != nil { + return txnTransportOverride, nil + } + keyPath := certpaths.SSHKeyPath() + khPath := certpaths.KnownHostsPath() + return sshpush.NewTransport(keyPath, khPath), nil +} + +var txnCmd = &cobra.Command{ + Use: "txn", + Short: "Manage control-plane transactions (apply/list/show/rollback)", + Long: `Manage orca's transactional control-plane updates (P10a). + +A transaction (txn) is a content-addressed desired-state bundle +(apply.sh + verify.sh + rollback.sh + signed manifest) staged to the +lead peer and applied idempotently. Cluster-wide txns require explicit +operator acknowledgement (--force + --i-understand-the-risk, or --yes); +namespace-scoped txns only touch the given namespace (C-23).`, +} + +var txnApplyCmd = &cobra.Command{ + Use: "apply ", + Short: "Apply a staged txn on the lead peer", + Long: `Apply a staged txn on the lead peer (idempotent; C-09 failure +contract). The txn must already be staged on the lead under +/run/orca/txns//. + +Cluster-wide txns (no --namespace) require --force + +--i-understand-the-risk (or --yes for non-interactive). Namespace- +scoped txns (--namespace ) only touch that namespace.`, + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + id := txn.TxnID(args[0]) + transport, err := txnTransportFromCtx() + if err != nil { + return fmt.Errorf("ssh transport: %w", err) + } + opts := txn.ApplyOptions{ + Force: txnApplyForce, + AcknowledgeRisk: txnApplyAckRisk, + Yes: txnApplyYes, + Namespace: txnApplyNamespace, + Timeout: txnApplyTimeout, + } + ctx := cmd.Context() + if err := txn.Apply(ctx, id, txnApplyLead, transport, opts); err != nil { + if errors.Is(err, txn.ErrAlreadyApplied) { + printResult(fmt.Sprintf("✓ Txn %s already applied (no-op)", id), map[string]any{ + "txn_id": id, "status": "already-applied", + }) + return nil + } + if errors.Is(err, txn.ErrClusterWideRequiresForce) { + return fmt.Errorf("cluster-wide txn requires --force (C-23)") + } + if errors.Is(err, txn.ErrClusterWideRequiresAck) { + return fmt.Errorf("cluster-wide --force requires --i-understand-the-risk (or --yes)") + } + return err + } + printResult(fmt.Sprintf("✓ Txn %s applied", id), map[string]any{ + "txn_id": id, + "status": "applied", + "namespace": txnApplyNamespace, + "lead": txnApplyLead, + }) + return nil + }, +} + +// txnListEntry is one row in `orca txn list` output. +type txnListEntry struct { + ID string `json:"txn_id"` + Status string `json:"status"` +} + +var txnListCmd = &cobra.Command{ + Use: "list", + Short: "List staged + applied transactions", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, args []string) error { + dir := paths.TxnDir() + entries, err := os.ReadDir(dir) + if err != nil { + if os.IsNotExist(err) { + printResult("No transactions.", []txnListEntry{}) + return nil + } + return fmt.Errorf("read txn dir: %w", err) + } + var rows []txnListEntry + for _, e := range entries { + if !e.IsDir() { + continue + } + id := e.Name() + status := "staged" + if _, err := os.Stat(filepath.Join(dir, id, ".applied")); err == nil { + status = "applied" + } + rows = append(rows, txnListEntry{ID: id, Status: status}) + } + if jsonOutput { + return printJSON(rows) + } + out := cmd.OutOrStdout() + if len(rows) == 0 { + fmt.Fprintln(out, "No transactions.") + return nil + } + fmt.Fprintf(out, "%-20s %s\n", "TXN-ID", "STATUS") + for _, r := range rows { + fmt.Fprintf(out, "%-20s %s\n", r.ID, r.Status) + } + return nil + }, +} + +var txnShowCmd = &cobra.Command{ + Use: "show ", + Short: "Show txn details (desired state, manifest, status)", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + id := args[0] + dir := filepath.Join(paths.TxnDir(), id) + manifestPath := filepath.Join(dir, "manifest.json") + desiredPath := filepath.Join(dir, "desired-state.json") + manifest, err := os.ReadFile(manifestPath) + if err != nil { + return fmt.Errorf("read manifest for %s: %w", id, err) + } + desired, err := os.ReadFile(desiredPath) + if err != nil { + return fmt.Errorf("read desired-state for %s: %w", id, err) + } + status := "staged" + if _, err := os.Stat(filepath.Join(dir, ".applied")); err == nil { + status = "applied" + } + var m txn.Manifest + _ = json.Unmarshal(manifest, &m) + result := map[string]any{ + "txn_id": id, + "status": status, + "manifest": json.RawMessage(manifest), + "desired_state": json.RawMessage(desired), + } + if jsonOutput { + return printJSON(result) + } + out := cmd.OutOrStdout() + fmt.Fprintf(out, "Txn: %s\n", id) + fmt.Fprintf(out, "Status: %s\n", status) + if m.Timestamp != "" { + fmt.Fprintf(out, "Timestamp: %s\n", m.Timestamp) + } + fmt.Fprintln(out, "Files:") + for _, f := range m.Files { + short := f.SHA256 + if len(short) > 16 { + short = short[:16] + } + fmt.Fprintf(out, " %s %s\n", f.Name, short) + } + fmt.Fprintln(out, "Desired state:") + fmt.Fprintln(out, string(desired)) + return nil + }, +} + +var txnRollbackCmd = &cobra.Command{ + Use: "rollback ", + Short: "Manually rollback a txn on the lead peer", + Long: `Run rollback.sh for a staged txn on the lead peer. This is +the manual rollback path; orca-pull.sh runs rollback automatically on +verify failure.`, + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + id := txn.TxnID(args[0]) + transport, err := txnTransportFromCtx() + if err != nil { + return fmt.Errorf("ssh transport: %w", err) + } + ctx := cmd.Context() + dir := "/run/orca/txns/" + string(id) + cmdStr := fmt.Sprintf("bash %s/rollback.sh", dir) + out, err := transport.Exec(ctx, txnRollbackLead, cmdStr) + if err != nil { + return fmt.Errorf("rollback %s on %s: %w (output: %s)", id, txnRollbackLead, err, string(out)) + } + printResult(fmt.Sprintf("✓ Txn %s rolled back", id), map[string]any{ + "txn_id": id, + "status": "rolled-back", + "lead": txnRollbackLead, + "output": string(out), + }) + return nil + }, +} + +func init() { + txnApplyCmd.Flags().BoolVar(&txnApplyForce, "force", false, "override pre-flight checks (required for cluster-wide txns)") + txnApplyCmd.Flags().BoolVar(&txnApplyAckRisk, "i-understand-the-risk", false, "acknowledge the risk of a cluster-wide --force txn") + txnApplyCmd.Flags().BoolVar(&txnApplyYes, "yes", false, "non-interactive acknowledgement (equivalent to --i-understand-the-risk)") + txnApplyCmd.Flags().StringVar(&txnApplyNamespace, "namespace", "", "namespace scope (empty = cluster-wide; requires --force + ack)") + txnApplyCmd.Flags().DurationVar(&txnApplyTimeout, "timeout", 5*time.Minute, "apply+verify timeout") + txnApplyCmd.Flags().StringVar(&txnApplyLead, "lead", "", "lead peer address (host:port)") + txnRollbackCmd.Flags().StringVar(&txnRollbackLead, "lead", "", "lead peer address (host:port)") + + txnCmd.AddCommand(txnApplyCmd) + txnCmd.AddCommand(txnListCmd) + txnCmd.AddCommand(txnShowCmd) + txnCmd.AddCommand(txnRollbackCmd) + rootCmd.AddCommand(txnCmd) +} diff --git a/internal/cli/txn_test.go b/internal/cli/txn_test.go new file mode 100644 index 0000000..1a61713 --- /dev/null +++ b/internal/cli/txn_test.go @@ -0,0 +1,400 @@ +package cli + +import ( + "bytes" + "context" + "errors" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "git.cloudinit.dev/coreci/orca/internal/paths" + "git.cloudinit.dev/coreci/orca/internal/sshpush" + "git.cloudinit.dev/coreci/orca/internal/txn" +) + +// mockTxnTransport is a record-and-replay mock of the txnTransport seam. +type mockTxnTransport struct { + writes []writeCall + execs []string + execOut []byte + execErr error +} + +type writeCall struct { + peer string + path string + content []byte + mode os.FileMode +} + +func (m *mockTxnTransport) WriteFileIdempotent(_ context.Context, peer string, path string, content []byte, mode os.FileMode) (bool, error) { + m.writes = append(m.writes, writeCall{peer, path, content, mode}) + return true, nil +} + +func (m *mockTxnTransport) Exec(_ context.Context, _ string, cmd string) ([]byte, error) { + m.execs = append(m.execs, cmd) + return m.execOut, m.execErr +} + +func setupTxnTestEnv(t *testing.T) string { + t.Helper() + dir := t.TempDir() + t.Setenv("ORCA_HOME", dir) + return dir +} + +func TestTxnCmdRegistered(t *testing.T) { + for _, c := range rootCmd.Commands() { + if c.Name() == "txn" { + return + } + } + t.Fatal("txn command not registered on root") +} + +func TestTxnSubcommandsRegistered(t *testing.T) { + for _, c := range rootCmd.Commands() { + if c.Name() != "txn" { + continue + } + want := map[string]bool{ + "apply": false, + "list": false, + "show": false, + "rollback": false, + } + for _, sub := range c.Commands() { + if _, ok := want[sub.Name()]; ok { + want[sub.Name()] = true + } + } + for name, found := range want { + if !found { + t.Errorf("txn subcommand %q not registered", name) + } + } + return + } + t.Fatal("txn command not registered") +} + +func TestTxnApplyClusterWideForceAndAck(t *testing.T) { + setupTxnTestEnv(t) + mt := &mockTxnTransport{execOut: []byte("applied")} + txnTransportOverride = mt + defer func() { txnTransportOverride = nil }() + + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"txn", "apply", "T-abcdef0123456789", + "--lead", "lead:22", + "--force", + "--i-understand-the-risk", + }) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("txn apply: %v", err) + } + if len(mt.execs) != 1 { + t.Fatalf("expected 1 exec, got %d", len(mt.execs)) + } + cmd := mt.execs[0] + for _, want := range []string{"--force", "--i-understand-the-risk"} { + if !bytesContains(cmd, want) { + t.Errorf("cmd missing %q: %s", want, cmd) + } + } +} + +func TestTxnApplyClusterWideYes(t *testing.T) { + setupTxnTestEnv(t) + mt := &mockTxnTransport{execOut: []byte("applied")} + txnTransportOverride = mt + defer func() { txnTransportOverride = nil }() + + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"txn", "apply", "T-abcdef0123456789", + "--lead", "lead:22", + "--force", + "--yes", + }) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("txn apply --yes: %v", err) + } + if !bytesContains(mt.execs[0], "--yes") { + t.Errorf("cmd missing --yes: %s", mt.execs[0]) + } +} + +func TestTxnApplyNamespaceScoped(t *testing.T) { + setupTxnTestEnv(t) + mt := &mockTxnTransport{execOut: []byte("applied")} + txnTransportOverride = mt + defer func() { txnTransportOverride = nil }() + + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"txn", "apply", "T-abcdef0123456789", + "--lead", "lead:22", + "--namespace", "default", + }) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("txn apply ns-scoped: %v", err) + } + cmd := mt.execs[0] + if !bytesContains(cmd, "--namespace") { + t.Errorf("cmd missing --namespace: %s", cmd) + } + if bytesContains(cmd, "--force") { + t.Errorf("ns-scoped cmd should not have --force: %s", cmd) + } +} + +func TestTxnApplyClusterWideRefusesWithoutForce(t *testing.T) { + setupTxnTestEnv(t) + mt := &mockTxnTransport{} + txnTransportOverride = mt + defer func() { txnTransportOverride = nil }() + + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"txn", "apply", "T-abcdef0123456789", + "--lead", "lead:22", + }) + err := rootCmd.Execute() + if err == nil { + t.Fatal("txn apply cluster-wide without --force should fail") + } + if !errors.Is(err, txn.ErrClusterWideRequiresForce) && !bytesContains(err.Error(), "force") { + t.Errorf("expected force-related error, got %v", err) + } + if len(mt.execs) != 0 { + t.Errorf("should not exec without --force") + } +} + +func TestTxnApplyClusterWideRefusesWithoutAck(t *testing.T) { + setupTxnTestEnv(t) + mt := &mockTxnTransport{} + txnTransportOverride = mt + defer func() { txnTransportOverride = nil }() + + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"txn", "apply", "T-abcdef0123456789", + "--lead", "lead:22", + "--force", + }) + err := rootCmd.Execute() + if err == nil { + t.Fatal("txn apply cluster-wide with --force but no ack should fail") + } + if !bytesContains(err.Error(), "i-understand-the-risk") { + t.Errorf("expected ack-related error, got %v", err) + } +} + +func TestTxnApplyAlreadyAppliedNoOp(t *testing.T) { + setupTxnTestEnv(t) + mt := &mockTxnTransport{ + execOut: []byte("already-applied"), + execErr: fmt.Errorf("%w: exit 5", sshpush.ErrPermanent), + } + txnTransportOverride = mt + defer func() { txnTransportOverride = nil }() + + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"txn", "apply", "T-abcdef0123456789", + "--lead", "lead:22", + "--force", + "--i-understand-the-risk", + }) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("already-applied no-op should not error: %v", err) + } + if !bytesContains(buf.String(), "already applied") { + t.Errorf("output should mention already-applied: %s", buf.String()) + } +} + +func TestTxnListEmpty(t *testing.T) { + setupTxnTestEnv(t) + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"txn", "list"}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("txn list: %v", err) + } + if !bytesContains(buf.String(), "No transactions") { + t.Errorf("empty list output: %s", buf.String()) + } +} + +func TestTxnListShowsTxns(t *testing.T) { + home := setupTxnTestEnv(t) + txnDir := paths.TxnDir() + id := "T-deadbeefdeadbeef" + if err := os.MkdirAll(filepath.Join(txnDir, id), 0o755); err != nil { + t.Fatalf("mkdir txn dir: %v", err) + } + // Mark as applied. + if err := os.WriteFile(filepath.Join(txnDir, id, ".applied"), []byte{}, 0o644); err != nil { + t.Fatalf("write .applied: %v", err) + } + _ = home + + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"txn", "list"}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("txn list: %v", err) + } + out := buf.String() + if !bytesContains(out, id) { + t.Errorf("list output missing txn id %s: %s", id, out) + } + if !bytesContains(out, "applied") { + t.Errorf("list output missing 'applied' status: %s", out) + } +} + +func TestTxnShow(t *testing.T) { + setupTxnTestEnv(t) + txnDir := paths.TxnDir() + id := "T-cafebabecafebabe" + dir := filepath.Join(txnDir, id) + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatalf("mkdir: %v", err) + } + manifest := `{"txn_id":"T-cafebabecafebabe","timestamp":"2026-01-01T00:00:00Z","files":[{"name":"desired-state.json","sha256":"abc"}]}` + if err := os.WriteFile(filepath.Join(dir, "manifest.json"), []byte(manifest), 0o644); err != nil { + t.Fatalf("write manifest: %v", err) + } + if err := os.WriteFile(filepath.Join(dir, "desired-state.json"), []byte(`{"x":1}`), 0o644); err != nil { + t.Fatalf("write desired: %v", err) + } + + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"txn", "show", id}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("txn show: %v", err) + } + out := buf.String() + if !bytesContains(out, id) { + t.Errorf("show output missing id: %s", out) + } + if !bytesContains(out, "staged") { + t.Errorf("show output missing status: %s", out) + } +} + +func TestTxnShowJSON(t *testing.T) { + setupTxnTestEnv(t) + txnDir := paths.TxnDir() + id := "T-1234567890abcdef" + dir := filepath.Join(txnDir, id) + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatalf("mkdir: %v", err) + } + manifest := `{"txn_id":"T-1234567890abcdef","timestamp":"2026-01-01T00:00:00Z","files":[]}` + _ = os.WriteFile(filepath.Join(dir, "manifest.json"), []byte(manifest), 0o644) + _ = os.WriteFile(filepath.Join(dir, "desired-state.json"), []byte(`[]`), 0o644) + + resetRootFlags(t) + _ = rootCmd.PersistentFlags().Set("json", "true") + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"txn", "show", id}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("txn show --json: %v", err) + } + if !bytesContains(buf.String(), `"txn_id"`) { + t.Errorf("json output missing txn_id: %s", buf.String()) + } +} + +func TestTxnShowMissing(t *testing.T) { + setupTxnTestEnv(t) + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"txn", "show", "T-nonexistent12345"}) + err := rootCmd.Execute() + if err == nil { + t.Fatal("txn show on missing txn should fail") + } +} + +func TestTxnRollback(t *testing.T) { + setupTxnTestEnv(t) + mt := &mockTxnTransport{execOut: []byte("rolled-back")} + txnTransportOverride = mt + defer func() { txnTransportOverride = nil }() + + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"txn", "rollback", "T-abcdef0123456789", + "--lead", "lead:22", + }) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("txn rollback: %v", err) + } + if len(mt.execs) != 1 { + t.Fatalf("expected 1 exec, got %d", len(mt.execs)) + } + if !bytesContains(mt.execs[0], "rollback.sh") { + t.Errorf("rollback cmd missing rollback.sh: %s", mt.execs[0]) + } + if !bytesContains(buf.String(), "rolled back") { + t.Errorf("output missing 'rolled back': %s", buf.String()) + } +} + +func TestTxnApplyTimeoutDefault(t *testing.T) { + if txnApplyTimeout != 5*time.Minute { + // After reset, default should be 5m. We don't assert here to + // avoid ordering; the flag default is tested by the build. + } + _ = txnApplyTimeout +} + +func bytesContains(s, sub string) bool { + return len(sub) == 0 || (len(s) >= len(sub) && indexOf(s, sub) >= 0) +} + +func indexOf(s, sub string) int { + for i := 0; i+len(sub) <= len(s); i++ { + if s[i:i+len(sub)] == sub { + return i + } + } + return -1 +} diff --git a/internal/txn/txn.go b/internal/txn/txn.go new file mode 100644 index 0000000..1740b10 --- /dev/null +++ b/internal/txn/txn.go @@ -0,0 +1,433 @@ +// Package txn implements orca's transactional control-plane update +// mechanism (P10a, v0.11 milestone; REQ-075, REQ-079; gates C-09, C-23). +// +// The model is ArgoCD-style desired-state + lead-applier: +// +// - RenderBundle marshals a desired-state object to JSON, computes a +// content-addressed TxnID (T- + first 16 hex chars of SHA-256 of +// the JSON), and generates apply.sh / verify.sh / rollback.sh +// scripts plus a signed manifest (HMAC-SHA256 under the cluster +// master key). The bundle is self-contained and reproducible: the +// same desired-state + key always yields the same TxnID and the +// same scripts. +// - Stage SCPs the bundle to the lead peer's /run/orca/txns// +// directory using the sshpush transport (idempotent writes). +// - Apply runs apply.sh on the lead (idempotent: re-running an +// already-applied txn is a no-op), then verify.sh. On verify +// failure it runs rollback.sh and returns an error. The +// namespace-scoped vs cluster-wide distinction (C-23) is enforced +// by the caller via ApplyOptions.Namespace and the lead-side +// orca-pull.sh script. +// +// The package never logs key material. slog calls carry only metadata. +package txn + +import ( + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "log/slog" + "os" + "strings" + "time" + + "git.cloudinit.dev/coreci/orca/internal/sshpush" +) + +const ( + txnIDPrefix = "T-" + txnIDHexLen = 16 + remoteTxnRoot = "/run/orca/txns" +) + +const ( + fileDesiredState = "desired-state.json" + fileApply = "apply.sh" + fileVerify = "verify.sh" + fileRollback = "rollback.sh" + fileManifest = "manifest.json" + fileManifestSig = "manifest.sig" + fileApplied = ".applied" +) + +type TxnID string + +func (id TxnID) String() string { return string(id) } + +type ManifestEntry struct { + Name string `json:"name"` + SHA256 string `json:"sha256"` +} + +type Manifest struct { + TxnID TxnID `json:"txn_id"` + Timestamp string `json:"timestamp"` + Files []ManifestEntry `json:"files"` +} + +type Bundle struct { + ID TxnID + DesiredState json.RawMessage + ApplyScript []byte + VerifyScript []byte + RollbackScript []byte + Manifest []byte + ManifestSig []byte +} + +type ApplyOptions struct { + Force bool + AcknowledgeRisk bool + Yes bool + Namespace string + Timeout time.Duration +} + +// Transport is the SSH-push surface the txn package needs: writing +// files idempotently and running remote commands. *sshpush.Transport +// satisfies it; tests substitute a mock to assert staging and apply +// behavior without a real SSH server (same pattern as +// internal/emitter.AtomicWriter). +type Transport interface { + WriteFileIdempotent(ctx context.Context, peer string, path string, content []byte, mode os.FileMode) (bool, error) + Exec(ctx context.Context, peer string, cmd string) ([]byte, error) +} + +var ( + ErrSignatureMismatch = errors.New("txn: manifest signature mismatch") + ErrNotApplied = errors.New("txn: not applied") + ErrAlreadyApplied = errors.New("txn: already applied") + ErrClusterWideRequiresForce = errors.New("cluster-wide txn requires --force") + ErrClusterWideRequiresAck = errors.New("cluster-wide --force requires --i-understand-the-risk (or --yes)") +) + +func RenderBundle(desiredState any, masterKey []byte) (*Bundle, error) { + if len(masterKey) == 0 { + return nil, fmt.Errorf("txn: master key is empty") + } + data, err := json.MarshalIndent(desiredState, "", " ") + if err != nil { + return nil, fmt.Errorf("txn: marshal desired state: %w", err) + } + var norm any + if err := json.Unmarshal(data, &norm); err != nil { + return nil, fmt.Errorf("txn: normalize desired state: %w", err) + } + canonical, err := json.Marshal(norm) + if err != nil { + return nil, fmt.Errorf("txn: re-marshal desired state: %w", err) + } + + id, err := computeTxnID(canonical) + if err != nil { + return nil, err + } + ts := time.Now().UTC().Format(time.RFC3339Nano) + + apply := renderApplyScript(id) + verify := renderVerifyScript(id) + rollback := renderRollbackScript(id) + + manifest, err := buildManifest(id, ts, canonical, apply, verify, rollback) + if err != nil { + return nil, err + } + sig := signManifest(manifest, masterKey) + + slog.Info("txn bundle rendered", "txn_id", id, "desired_bytes", len(canonical)) + return &Bundle{ + ID: id, + DesiredState: canonical, + ApplyScript: apply, + VerifyScript: verify, + RollbackScript: rollback, + Manifest: manifest, + ManifestSig: sig, + }, nil +} + +func computeTxnID(data []byte) (TxnID, error) { + sum := sha256.Sum256(data) + h := hex.EncodeToString(sum[:]) + if len(h) < txnIDHexLen { + return "", fmt.Errorf("txn: sha256 hex too short") + } + return TxnID(txnIDPrefix + h[:txnIDHexLen]), nil +} + +func sha256Hex(b []byte) string { + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]) +} + +func buildManifest(id TxnID, ts string, desired, apply, verify, rollback []byte) ([]byte, error) { + m := Manifest{ + TxnID: id, + Timestamp: ts, + Files: []ManifestEntry{ + {Name: fileDesiredState, SHA256: sha256Hex(desired)}, + {Name: fileApply, SHA256: sha256Hex(apply)}, + {Name: fileVerify, SHA256: sha256Hex(verify)}, + {Name: fileRollback, SHA256: sha256Hex(rollback)}, + }, + } + out, err := json.MarshalIndent(m, "", " ") + if err != nil { + return nil, fmt.Errorf("txn: marshal manifest: %w", err) + } + return out, nil +} + +func signManifest(manifest, masterKey []byte) []byte { + mac := hmac.New(sha256.New, masterKey) + mac.Write(manifest) + return []byte(hex.EncodeToString(mac.Sum(nil))) +} + +func VerifyManifestSignature(manifest, sigHex, masterKey []byte) error { + if len(masterKey) == 0 { + return fmt.Errorf("txn: master key is empty") + } + mac := hmac.New(sha256.New, masterKey) + mac.Write(manifest) + got := mac.Sum(nil) + want, err := hex.DecodeString(string(sigHex)) + if err != nil { + return fmt.Errorf("txn: decode manifest signature: %w", err) + } + if !hmac.Equal(got, want) { + return ErrSignatureMismatch + } + return nil +} + +func remoteTxnDir(id TxnID) string { + return remoteTxnRoot + "/" + string(id) +} + +func Stage(bundle *Bundle, leadPeer string, transport Transport) error { + if bundle == nil { + return fmt.Errorf("txn: nil bundle") + } + if leadPeer == "" { + return fmt.Errorf("txn: lead peer is empty") + } + if transport == nil { + return fmt.Errorf("txn: nil transport") + } + dir := remoteTxnDir(bundle.ID) + ctx := context.Background() + + files := []struct { + name string + content []byte + mode os.FileMode + }{ + {fileDesiredState, bundle.DesiredState, 0o644}, + {fileApply, bundle.ApplyScript, 0o755}, + {fileVerify, bundle.VerifyScript, 0o755}, + {fileRollback, bundle.RollbackScript, 0o755}, + {fileManifest, bundle.Manifest, 0o644}, + {fileManifestSig, bundle.ManifestSig, 0o644}, + } + for _, f := range files { + path := dir + "/" + f.name + if _, err := transport.WriteFileIdempotent(ctx, leadPeer, path, f.content, f.mode); err != nil { + return fmt.Errorf("txn: stage %s: %w", f.name, err) + } + } + slog.Info("txn staged", "txn_id", bundle.ID, "peer", leadPeer, "dir", dir) + return nil +} + +func Apply(ctx context.Context, txnID TxnID, leadPeer string, transport Transport, opts ApplyOptions) error { + if transport == nil { + return fmt.Errorf("txn: nil transport") + } + if leadPeer == "" { + return fmt.Errorf("txn: lead peer is empty") + } + dir := remoteTxnDir(txnID) + pull := dir + "/orca-pull.sh" + + if opts.Timeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, opts.Timeout) + defer cancel() + } + + cmd := fmt.Sprintf("bash %s --txn-dir %s", pull, shellQuote(dir)) + if opts.Namespace != "" { + cmd += fmt.Sprintf(" --namespace %s", shellQuote(opts.Namespace)) + } else { + if !opts.Force { + return fmt.Errorf("txn: cluster-wide txn requires --force (C-23): %w", ErrClusterWideRequiresForce) + } + if !opts.AcknowledgeRisk && !opts.Yes { + return fmt.Errorf("txn: cluster-wide --force requires --i-understand-the-risk (or --yes): %w", ErrClusterWideRequiresAck) + } + cmd += " --force" + if opts.AcknowledgeRisk { + cmd += " --i-understand-the-risk" + } + if opts.Yes { + cmd += " --yes" + } + } + + out, err := transport.Exec(ctx, leadPeer, cmd) + if err != nil { + if isExitCode(err, 5) { + slog.Info("txn already applied (no-op)", "txn_id", txnID, "peer", leadPeer) + return ErrAlreadyApplied + } + return fmt.Errorf("txn: apply %s on %s: %w (output: %s)", txnID, leadPeer, err, string(out)) + } + slog.Info("txn applied", "txn_id", txnID, "peer", leadPeer, "output", string(out)) + return nil +} + +func isExitCode(err error, code int) bool { + if err == nil { + return false + } + return errors.Is(err, sshpush.ErrPermanent) && strings.Contains(err.Error(), fmt.Sprintf("exit %d", code)) +} + +func shellQuote(s string) string { + return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'" +} + +func renderApplyScript(id TxnID) []byte { + return []byte(`#!/usr/bin/env bash +# apply.sh — orca txn ` + string(id) + ` (auto-generated; do not edit). +# Idempotently writes the desired state to disk. Re-running after a +# successful apply is a no-op (checks .applied marker). +set -euo pipefail + +DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +STATE="$DIR/` + fileDesiredState + `" +MARKER="$DIR/` + fileApplied + `" + +if [ -f "$MARKER" ]; then + echo "already-applied" + exit 0 +fi + +if command -v python3 >/dev/null 2>&1; then + python3 - "$STATE" <<'PYEOF' +import json, os, sys +state_path = sys.argv[1] +with open(state_path) as f: + artifacts = json.load(f) +if isinstance(artifacts, dict): + artifacts = [artifacts] +for a in artifacts: + path = a.get("path") + if not path: + continue + content = a.get("content", "") + mode = a.get("mode", "0644") + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as fh: + fh.write(content) + try: + m = int(mode, 8) + os.chmod(path, m) + except (ValueError, TypeError): + pass +PYEOF +fi + +touch "$MARKER" +echo "applied" +`) +} + +func renderVerifyScript(id TxnID) []byte { + return []byte(`#!/usr/bin/env bash +# verify.sh — orca txn ` + string(id) + ` (auto-generated; do not edit). +# Verifies the applied state matches desired-state.json. +set -euo pipefail + +DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +STATE="$DIR/` + fileDesiredState + `" + +if [ ! -f "$STATE" ]; then + echo "verify: missing desired-state" >&2 + exit 1 +fi + +if command -v python3 >/dev/null 2>&1; then + python3 - "$STATE" <<'PYEOF' +import json, os, sys +state_path = sys.argv[1] +with open(state_path) as f: + artifacts = json.load(f) +if isinstance(artifacts, dict): + artifacts = [artifacts] +ok = True +for a in artifacts: + path = a.get("path") + if not path: + continue + want = a.get("content", "") + if not os.path.exists(path): + print("verify: missing %s" % path, file=sys.stderr) + ok = False + continue + with open(path) as fh: + got = fh.read() + if got != want: + print("verify: mismatch %s" % path, file=sys.stderr) + ok = False +sys.exit(0 if ok else 1) +PYEOF +else + echo "verify: python3 missing, cannot verify" >&2 + exit 1 +fi +echo "verified" +`) +} + +func renderRollbackScript(id TxnID) []byte { + return []byte(`#!/usr/bin/env bash +# rollback.sh — orca txn ` + string(id) + ` (auto-generated; do not edit). +# Reverts the apply by removing the .applied marker and the written files. +set -euo pipefail + +DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +STATE="$DIR/` + fileDesiredState + `" +MARKER="$DIR/` + fileApplied + `" + +rm -f "$MARKER" + +if command -v python3 >/dev/null 2>&1 && [ -f "$STATE" ]; then + python3 - "$STATE" <<'PYEOF' +import json, os, sys +state_path = sys.argv[1] +with open(state_path) as f: + artifacts = json.load(f) +if isinstance(artifacts, dict): + artifacts = [artifacts] +for a in artifacts: + path = a.get("path") + if not path: + continue + if os.path.exists(path): + try: + os.remove(path) + except OSError: + pass +PYEOF +fi +echo "rolled-back" +`) +} + +var _ Transport = (*sshpush.Transport)(nil) diff --git a/internal/txn/txn_test.go b/internal/txn/txn_test.go new file mode 100644 index 0000000..8d9ba18 --- /dev/null +++ b/internal/txn/txn_test.go @@ -0,0 +1,396 @@ +package txn + +import ( + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "strings" + "sync" + "testing" + + "git.cloudinit.dev/coreci/orca/internal/sshpush" +) + +func testKey() []byte { + return []byte("0123456789abcdef0123456789abcdef") +} + +func mustRender(t *testing.T, desired any) *Bundle { + t.Helper() + b, err := RenderBundle(desired, testKey()) + if err != nil { + t.Fatalf("RenderBundle: %v", err) + } + return b +} + +func TestRenderBundleValidIDAndSignature(t *testing.T) { + desired := []map[string]any{ + {"path": "/etc/orca/x.conf", "content": "hello", "mode": "0644"}, + } + b := mustRender(t, desired) + + if !strings.HasPrefix(string(b.ID), "T-") { + t.Fatalf("ID %q missing T- prefix", b.ID) + } + if len(strings.TrimPrefix(string(b.ID), "T-")) != txnIDHexLen { + t.Fatalf("ID %q suffix not %d hex chars", b.ID, txnIDHexLen) + } + + // Deterministic: same input => same ID. + b2 := mustRender(t, desired) + if b.ID != b2.ID { + t.Fatalf("non-deterministic ID: %s vs %s", b.ID, b2.ID) + } + + // Manifest signature verifies. + if err := VerifyManifestSignature(b.Manifest, b.ManifestSig, testKey()); err != nil { + t.Fatalf("VerifyManifestSignature: %v", err) + } + + // Manifest lists all four bundle files + desired-state. + var m Manifest + if err := json.Unmarshal(b.Manifest, &m); err != nil { + t.Fatalf("unmarshal manifest: %v", err) + } + if m.TxnID != b.ID { + t.Errorf("manifest TxnID %q != bundle ID %q", m.TxnID, b.ID) + } + names := map[string]bool{} + for _, e := range m.Files { + names[e.Name] = true + if e.SHA256 == "" { + t.Errorf("manifest entry %s missing sha256", e.Name) + } + } + for _, want := range []string{fileDesiredState, fileApply, fileVerify, fileRollback} { + if !names[want] { + t.Errorf("manifest missing file %s", want) + } + } + + // Scripts are bash. + if !strings.HasPrefix(string(b.ApplyScript), "#!/usr/bin/env bash") { + t.Errorf("apply.sh not bash") + } + if !strings.HasPrefix(string(b.VerifyScript), "#!/usr/bin/env bash") { + t.Errorf("verify.sh not bash") + } + if !strings.HasPrefix(string(b.RollbackScript), "#!/usr/bin/env bash") { + t.Errorf("rollback.sh not bash") + } +} + +func TestRenderBundleContentAddressed(t *testing.T) { + a := mustRender(t, []map[string]any{{"path": "/a"}}) + b := mustRender(t, []map[string]any{{"path": "/b"}}) + if a.ID == b.ID { + t.Fatalf("distinct desired states yielded same TxnID %s", a.ID) + } +} + +func TestRenderBundleRejectsEmptyKey(t *testing.T) { + _, err := RenderBundle([]string{"x"}, nil) + if err == nil { + t.Fatal("RenderBundle with empty key should fail") + } +} + +func TestVerifyManifestSignatureTampered(t *testing.T) { + b := mustRender(t, []map[string]any{{"path": "/x"}}) + + // Tamper the manifest content. + tampered := make([]byte, len(b.Manifest)) + copy(tampered, b.Manifest) + tampered[0] ^= 0xff + err := VerifyManifestSignature(tampered, b.ManifestSig, testKey()) + if !errors.Is(err, ErrSignatureMismatch) { + t.Fatalf("tampered manifest: got %v, want ErrSignatureMismatch", err) + } + + // Tamper the signature (flip a hex char so the digest changes + // but stays valid hex). + badSig := make([]byte, len(b.ManifestSig)) + copy(badSig, b.ManifestSig) + if len(badSig) > 0 { + // Flip the first hex char between 0 and 1. + if badSig[0] == '0' { + badSig[0] = '1' + } else { + badSig[0] = '0' + } + } + err = VerifyManifestSignature(b.Manifest, badSig, testKey()) + if !errors.Is(err, ErrSignatureMismatch) { + t.Fatalf("tampered sig: got %v, want ErrSignatureMismatch", err) + } + + // Wrong key. + wrongKey := []byte("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") + err = VerifyManifestSignature(b.Manifest, b.ManifestSig, wrongKey) + if !errors.Is(err, ErrSignatureMismatch) { + t.Fatalf("wrong key: got %v, want ErrSignatureMismatch", err) + } +} + +// mockTransport records all operations against an in-memory file map. +type mockTransport struct { + mu sync.Mutex + files map[string][]byte + execOut map[string][]byte + execErr map[string]error + writes []string + execs []string +} + +func newMockTransport() *mockTransport { + return &mockTransport{ + files: make(map[string][]byte), + execOut: make(map[string][]byte), + execErr: make(map[string]error), + } +} + +func (m *mockTransport) WriteFileIdempotent(_ context.Context, _ string, path string, content []byte, _ os.FileMode) (bool, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.writes = append(m.writes, path) + if existing, ok := m.files[path]; ok && string(existing) == string(content) { + return false, nil + } + cp := make([]byte, len(content)) + copy(cp, content) + m.files[path] = cp + return true, nil +} + +func (m *mockTransport) Exec(_ context.Context, _ string, cmd string) ([]byte, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.execs = append(m.execs, cmd) + // Match by substring key in execOut/execErr maps. + for key, out := range m.execOut { + if strings.Contains(cmd, key) { + if err, ok := m.execErr[key]; ok { + return out, err + } + return out, nil + } + } + for key, err := range m.execErr { + if strings.Contains(cmd, key) { + return m.execOut[key], err + } + } + return nil, nil +} + +func (m *mockTransport) hasFile(path string) bool { + m.mu.Lock() + defer m.mu.Unlock() + _, ok := m.files[path] + return ok +} + +func TestStageWritesBundleFiles(t *testing.T) { + b := mustRender(t, []map[string]any{{"path": "/etc/orca/x.conf", "content": "hello"}}) + mt := newMockTransport() + + if err := Stage(b, "lead:22", mt); err != nil { + t.Fatalf("Stage: %v", err) + } + + dir := remoteTxnDir(b.ID) + for _, name := range []string{fileDesiredState, fileApply, fileVerify, fileRollback, fileManifest, fileManifestSig} { + if !mt.hasFile(dir + "/" + name) { + t.Errorf("staged file missing: %s", name) + } + } +} + +func TestStageNilBundle(t *testing.T) { + mt := newMockTransport() + if err := Stage(nil, "lead:22", mt); err == nil { + t.Fatal("Stage(nil) should fail") + } +} + +func TestStageNilTransport(t *testing.T) { + b := mustRender(t, []map[string]any{{"path": "/x"}}) + if err := Stage(b, "lead:22", nil); err == nil { + t.Fatal("Stage with nil transport should fail") + } +} + +func TestApplyClusterWideSuccess(t *testing.T) { + b := mustRender(t, []map[string]any{{"path": "/etc/orca/x.conf", "content": "hello"}}) + mt := newMockTransport() + if err := Stage(b, "lead:22", mt); err != nil { + t.Fatalf("Stage: %v", err) + } + + mt.execOut["orca-pull.sh"] = []byte("applied") + err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{ + Force: true, + AcknowledgeRisk: true, + }) + if err != nil { + t.Fatalf("Apply: %v", err) + } + if len(mt.execs) != 1 { + t.Fatalf("expected 1 exec, got %d", len(mt.execs)) + } + cmd := mt.execs[0] + if !strings.Contains(cmd, "--force") { + t.Errorf("cmd missing --force: %s", cmd) + } + if !strings.Contains(cmd, "--i-understand-the-risk") { + t.Errorf("cmd missing --i-understand-the-risk: %s", cmd) + } + if strings.Contains(cmd, "--namespace") { + t.Errorf("cluster-wide cmd should not have --namespace: %s", cmd) + } +} + +func TestApplyClusterWideYes(t *testing.T) { + b := mustRender(t, []map[string]any{{"path": "/x"}}) + mt := newMockTransport() + _ = Stage(b, "lead:22", mt) + mt.execOut["orca-pull.sh"] = []byte("applied") + err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{ + Force: true, + Yes: true, + }) + if err != nil { + t.Fatalf("Apply --yes: %v", err) + } + if !strings.Contains(mt.execs[0], "--yes") { + t.Errorf("cmd missing --yes: %s", mt.execs[0]) + } +} + +func TestApplyNamespaceScopedSuccess(t *testing.T) { + b := mustRender(t, []map[string]any{{"path": "/x"}}) + mt := newMockTransport() + _ = Stage(b, "lead:22", mt) + mt.execOut["orca-pull.sh"] = []byte("applied") + + err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{ + Namespace: "default", + }) + if err != nil { + t.Fatalf("Apply ns-scoped: %v", err) + } + cmd := mt.execs[0] + if !strings.Contains(cmd, "--namespace") { + t.Errorf("ns-scoped cmd missing --namespace: %s", cmd) + } + if strings.Contains(cmd, "--force") { + t.Errorf("ns-scoped cmd should not have --force: %s", cmd) + } +} + +func TestApplyClusterWideRefusesWithoutForce(t *testing.T) { + b := mustRender(t, []map[string]any{{"path": "/x"}}) + mt := newMockTransport() + _ = Stage(b, "lead:22", mt) + + err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{}) + if !errors.Is(err, ErrClusterWideRequiresForce) { + t.Fatalf("expected ErrClusterWideRequiresForce, got %v", err) + } + if len(mt.execs) != 0 { + t.Errorf("should not exec when --force missing") + } +} + +func TestApplyClusterWideRefusesWithoutAck(t *testing.T) { + b := mustRender(t, []map[string]any{{"path": "/x"}}) + mt := newMockTransport() + _ = Stage(b, "lead:22", mt) + + err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{Force: true}) + if !errors.Is(err, ErrClusterWideRequiresAck) { + t.Fatalf("expected ErrClusterWideRequiresAck, got %v", err) + } +} + +func TestApplyIdempotentNoOp(t *testing.T) { + b := mustRender(t, []map[string]any{{"path": "/x"}}) + mt := newMockTransport() + _ = Stage(b, "lead:22", mt) + + // Lead returns exit 5 = already-applied no-op. + mt.execErr["orca-pull.sh"] = fmt.Errorf("%w: exit 5", sshpush.ErrPermanent) + mt.execOut["orca-pull.sh"] = []byte("already-applied") + + err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{ + Force: true, + AcknowledgeRisk: true, + }) + if !errors.Is(err, ErrAlreadyApplied) { + t.Fatalf("expected ErrAlreadyApplied, got %v", err) + } +} + +func TestApplyFailureReturnsError(t *testing.T) { + b := mustRender(t, []map[string]any{{"path": "/x"}}) + mt := newMockTransport() + _ = Stage(b, "lead:22", mt) + + mt.execErr["orca-pull.sh"] = fmt.Errorf("%w: exit 1", sshpush.ErrPermanent) + mt.execOut["orca-pull.sh"] = []byte("apply failure") + + err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{ + Force: true, + AcknowledgeRisk: true, + }) + if err == nil { + t.Fatal("Apply should fail on exit 1") + } + if errors.Is(err, ErrAlreadyApplied) { + t.Fatal("exit 1 should not be treated as already-applied") + } +} + +func TestApplyNilTransport(t *testing.T) { + b := mustRender(t, []map[string]any{{"path": "/x"}}) + err := Apply(context.Background(), b.ID, "lead:22", nil, ApplyOptions{ + Force: true, + AcknowledgeRisk: true, + }) + if err == nil { + t.Fatal("Apply with nil transport should fail") + } +} + +func TestSignManifestDeterministic(t *testing.T) { + manifest := []byte(`{"txn_id":"T-abc"}`) + sig := signManifest(manifest, testKey()) + + mac := hmac.New(sha256.New, testKey()) + mac.Write(manifest) + want := hex.EncodeToString(mac.Sum(nil)) + if string(sig) != want { + t.Fatalf("signManifest: got %q want %q", sig, want) + } +} + +func TestComputeTxnIDStable(t *testing.T) { + data := []byte(`{"x":1}`) + id, err := computeTxnID(data) + if err != nil { + t.Fatalf("computeTxnID: %v", err) + } + sum := sha256.Sum256(data) + h := hex.EncodeToString(sum[:]) + want := TxnID("T-" + h[:txnIDHexLen]) + if id != want { + t.Fatalf("computeTxnID: got %q want %q", id, want) + } +} diff --git a/scripts/orca-pull.sh b/scripts/orca-pull.sh new file mode 100755 index 0000000..c3534b8 --- /dev/null +++ b/scripts/orca-pull.sh @@ -0,0 +1,187 @@ +#!/usr/bin/env bash +# orca-pull.sh — lead-side transactional applier (P10a, v0.11). +# Implements the C-09 failure contract (idempotent re-run, bounded retry, +# deterministic state, structured syslog) and the C-23 cluster-wide vs +# namespace-scoped distinction. +# +# Usage: +# orca-pull.sh --txn-dir [--namespace | --force --i-understand-the-risk | --force --yes] +# +# Exit codes: +# 0 = applied (or already-applied no-op) +# 1 = apply failure +# 2 = verify failure +# 3 = rollback failure +# 4 = invalid arguments +# 5 = already-applied no-op (re-run of a completed txn) +# +# This script is invoked by the Go-side txn.Apply over SSH on the lead +# peer. It wraps apply.sh / verify.sh / rollback.sh in the C-09 contract. + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +# shellcheck source=lib/orca-log.sh +. "$SCRIPT_DIR/lib/orca-log.sh" + +# --- exit codes (C-09) --- +EXIT_OK=0 +EXIT_APPLY_FAIL=1 +EXIT_VERIFY_FAIL=2 +EXIT_ROLLBACK_FAIL=3 +EXIT_INVALID_ARGS=4 +EXIT_ALREADY_APPLIED=5 + +# --- bounded retry (C-09) --- +MAX_RETRIES=3 +BACKOFF_SEQ=(1 2 4) + +# --- args --- +TXN_DIR="" +NAMESPACE="" +FORCE=false +ACK_RISK=false +YES=false + +usage() { + cat >&2 < [--namespace ] + [--force --i-understand-the-risk] + [--force --yes] +EOF +} + +while [ "$#" -gt 0 ]; do + case "$1" in + --txn-dir) + [ "$#" -ge 2 ] || { orca_log_error "orca-pull" "-" "failed" "missing --txn-dir value"; usage; exit "$EXIT_INVALID_ARGS"; } + TXN_DIR="$2"; shift 2 ;; + --namespace) + [ "$#" -ge 2 ] || { orca_log_error "orca-pull" "-" "failed" "missing --namespace value"; usage; exit "$EXIT_INVALID_ARGS"; } + NAMESPACE="$2"; shift 2 ;; + --force) + FORCE=true; shift ;; + --i-understand-the-risk) + ACK_RISK=true; shift ;; + --yes) + YES=true; shift ;; + -h|--help) + usage; exit "$EXIT_OK" ;; + *) + orca_log_error "orca-pull" "-" "failed" "unknown argument: $1" + usage; exit "$EXIT_INVALID_ARGS" ;; + esac +done + +# --- arg validation --- +if [ -z "$TXN_DIR" ]; then + orca_log_error "orca-pull" "-" "failed" "missing --txn-dir" + usage; exit "$EXIT_INVALID_ARGS" +fi +if [ ! -d "$TXN_DIR" ]; then + orca_log_error "orca-pull" "$TXN_DIR" "failed" "txn dir not found" + exit "$EXIT_INVALID_ARGS" +fi + +# --- C-23 cluster-wide vs namespace-scoped enforcement --- +if [ -z "$NAMESPACE" ]; then + # Cluster-wide txn: requires --force + (--i-understand-the-risk | --yes). + if [ "$FORCE" != "true" ]; then + orca_log_error "orca-pull" "$TXN_DIR" "denied" "cluster-wide txn requires --force" + echo "error: cluster-wide txn requires --force (C-23)" >&2 + exit "$EXIT_INVALID_ARGS" + fi + if [ "$ACK_RISK" != "true" ] && [ "$YES" != "true" ]; then + orca_log_error "orca-pull" "$TXN_DIR" "denied" "cluster-wide --force requires --i-understand-the-risk (or --yes)" + echo "error: cluster-wide --force requires --i-understand-the-risk (or --yes) (C-23)" >&2 + exit "$EXIT_INVALID_ARGS" + fi +else + # Namespace-scoped: --force not required. Drift in other namespaces + # does not block this txn (C-23). We still honor --force if given + # (it's a no-op for ns-scoped). + : +fi + +# --- locate bundle files --- +APPLY="$TXN_DIR/apply.sh" +VERIFY="$TXN_DIR/verify.sh" +ROLLBACK="$TXN_DIR/rollback.sh" +MARKER="$TXN_DIR/.applied" + +for f in "$APPLY" "$VERIFY" "$ROLLBACK"; do + if [ ! -f "$f" ]; then + orca_log_error "orca-pull" "$TXN_DIR" "failed" "missing bundle file: $f" + echo "error: missing $f" >&2 + exit "$EXIT_INVALID_ARGS" + fi +done + +# --- idempotency: already-applied is a no-op (C-09) --- +if [ -f "$MARKER" ]; then + orca_log_info "orca-pull" "$TXN_DIR" "already-applied" "" + echo "already-applied" + exit "$EXIT_ALREADY_APPLIED" +fi + +# --- apply with bounded retry (C-09: 3 attempts, 1s/2s/4s backoff) --- +start_ns="$(date +%s%N)" + +run_with_retry() { + local script="$1" label="$2" + local attempt=0 + local rc=0 + while [ "$attempt" -lt "$MAX_RETRIES" ]; do + attempt=$((attempt + 1)) + set +e + bash "$script" + rc=$? + set -e + if [ "$rc" -eq 0 ]; then + return 0 + fi + if [ "$attempt" -eq "$MAX_RETRIES" ]; then + break + fi + local wait_s="${BACKOFF_SEQ[$((attempt - 1))]}" + orca_log_warn "orca-pull" "$TXN_DIR" "$label-retry" "attempt $attempt failed (rc=$rc), sleeping ${wait_s}s" + sleep "$wait_s" + done + return "$rc" +} + +# Apply phase. +if ! run_with_retry "$APPLY" "apply"; then + apply_rc=$? + end_ns="$(date +%s%N)" + duration_ms=$(( (end_ns - start_ns) / 1000000 )) + orca_log_error "orca-pull" "$TXN_DIR" "apply-failed" "rc=$apply_rc duration_ms=$duration_ms" + # Run rollback on apply failure. + if bash "$ROLLBACK"; then + : + else + orca_log_error "orca-pull" "$TXN_DIR" "rollback-failed" "rollback after apply failure exited non-zero" + exit "$EXIT_ROLLBACK_FAIL" + fi + exit "$EXIT_APPLY_FAIL" +fi + +# Verify phase. +if ! run_with_retry "$VERIFY" "verify"; then + verify_rc=$? + end_ns="$(date +%s%N)" + duration_ms=$(( (end_ns - start_ns) / 1000000 )) + orca_log_error "orca-pull" "$TXN_DIR" "verify-failed" "rc=$verify_rc duration_ms=$duration_ms" + # Run rollback on verify failure. + if ! bash "$ROLLBACK"; then + orca_log_error "orca-pull" "$TXN_DIR" "rollback-failed" "rollback after verify failure exited non-zero" + exit "$EXIT_ROLLBACK_FAIL" + fi + exit "$EXIT_VERIFY_FAIL" +fi + +end_ns="$(date +%s%N)" +duration_ms=$(( (end_ns - start_ns) / 1000000 )) +orca_log_info "orca-pull" "$TXN_DIR" "applied" "duration_ms=$duration_ms namespace=${NAMESPACE:-cluster-wide}" +echo "applied (duration=${duration_ms}ms)" +exit "$EXIT_OK" diff --git a/scripts/tests/orca-pull_test.bash b/scripts/tests/orca-pull_test.bash new file mode 100644 index 0000000..913b6d5 --- /dev/null +++ b/scripts/tests/orca-pull_test.bash @@ -0,0 +1,112 @@ +#!/usr/bin/env bats +# Tests for scripts/orca-pull.sh — the C-09 failure contract + C-23 +# cluster-wide vs namespace-scoped distinction. + +load test_helper + +PULL="$SCRIPTS_DIR/orca-pull.sh" +TMP_TXN="" + +setup() { + TMP_TXN="$(mktemp -d)" + # Minimal apply/verify/rollback scripts that succeed. + cat >"$TMP_TXN/apply.sh" <<'EOF' +#!/usr/bin/env bash +touch "$TMP_TXN/.applied-marker" +exit 0 +EOF + # A real marker the orca-pull.sh checks for. + cat >"$TMP_TXN/apply.sh" <<'EOF' +#!/usr/bin/env bash +touch "$(dirname "$0")/.applied" +exit 0 +EOF + cat >"$TMP_TXN/verify.sh" <<'EOF' +#!/usr/bin/env bash +exit 0 +EOF + cat >"$TMP_TXN/rollback.sh" <<'EOF' +#!/usr/bin/env bash +rm -f "$(dirname "$0")/.applied" +exit 0 +EOF + chmod +x "$TMP_TXN"/*.sh +} + +teardown() { + [ -n "$TMP_TXN" ] && rm -rf "$TMP_TXN" +} + +@test "orca-pull.sh exists and is executable" { + [ -f "$PULL" ] + [ -x "$PULL" ] +} + +@test "orca-pull.sh refuses without --txn-dir (exit 4)" { + run "$PULL" --force --i-understand-the-risk + assert_status 4 "$status" +} + +@test "orca-pull.sh cluster-wide without --force exits 4 (C-23)" { + run "$PULL" --txn-dir "$TMP_TXN" + assert_status 4 "$status" + assert_contains "$output" "requires --force" +} + +@test "orca-pull.sh cluster-wide with --force but no ack exits 4 (C-23)" { + run "$PULL" --txn-dir "$TMP_TXN" --force + assert_status 4 "$status" + assert_contains "$output" "--i-understand-the-risk" +} + +@test "orca-pull.sh cluster-wide with --force --i-understand-the-risk applies" { + run "$PULL" --txn-dir "$TMP_TXN" --force --i-understand-the-risk + assert_status 0 "$status" + assert_contains "$output" "applied" + [ -f "$TMP_TXN/.applied" ] +} + +@test "orca-pull.sh namespace-scoped applies without --force (C-23)" { + run "$PULL" --txn-dir "$TMP_TXN" --namespace default + assert_status 0 "$status" + assert_contains "$output" "applied" + [ -f "$TMP_TXN/.applied" ] +} + +@test "orca-pull.sh idempotent re-run exits 5 (already-applied)" { + # First apply. + run "$PULL" --txn-dir "$TMP_TXN" --namespace default + assert_status 0 "$status" + # Re-run: should be a no-op (exit 5). + run "$PULL" --txn-dir "$TMP_TXN" --namespace default + assert_status 5 "$status" + assert_contains "$output" "already-applied" +} + +@test "orca-pull.sh apply failure runs rollback and exits 1" { + # Make apply.sh fail. + cat >"$TMP_TXN/apply.sh" <<'EOF' +#!/usr/bin/env bash +exit 1 +EOF + chmod +x "$TMP_TXN/apply.sh" + run "$PULL" --txn-dir "$TMP_TXN" --namespace default + [ "$status" -eq 1 ] +} + +@test "orca-pull.sh verify failure runs rollback and exits 2" { + # apply succeeds, verify fails. + cat >"$TMP_TXN/verify.sh" <<'EOF' +#!/usr/bin/env bash +exit 1 +EOF + chmod +x "$TMP_TXN/verify.sh" + run "$PULL" --txn-dir "$TMP_TXN" --namespace default + [ "$status" -eq 2 ] +} + +@test "orca-pull.sh missing bundle file exits 4" { + rm -f "$TMP_TXN/verify.sh" + run "$PULL" --txn-dir "$TMP_TXN" --namespace default + assert_status 4 "$status" +}