diff --git a/README.md b/README.md
index 67fe64d..b90a758 100644
--- a/README.md
+++ b/README.md
@@ -84,6 +84,17 @@ the task fails before the script runs. This is workdir containment, not a full
mount namespace: scripts can still read or write other host paths allowed to
the agent user.
+NetGuard apply leases explicitly marked `durable_result` by the server are
+journaled before execution and upload. This scope matches NetGuard's atomic
+server-side result/approval/binding transition; heterogeneous legacy tasks keep
+their historical one-shot delivery semantics. The outbox uses a private,
+server-and-node-specific subdirectory under `LATTICE_LOG_STATE_DIR`; manual runs
+without that setting use the current user's cache directory. Override the base
+directory with `LATTICE_TASK_OUTBOX_DIR` or `-task-outbox-dir`. If a marked lease
+cannot be written, it does not run. After a restart, completed marked results
+are retried first and an interrupted marked task is reported as an unknown
+outcome rather than executed a second time.
+
For least-privilege Linux systemd installs, set `LATTICE_AGENT_RUN_USER` before
running `scripts/install.sh`:
@@ -302,6 +313,10 @@ missing checksum manifest aborts the install before the binary is written.
- `LATTICE_AGENT_ALLOW_EXEC=1` enables bounded task execution.
- `LATTICE_AGENT_ALLOW_ROOT_EXEC=1` permits task execution while the agent runs
as root.
+- `LATTICE_TASK_OUTBOX_DIR` overrides the durable NetGuard result-journal base. The
+ installer creates a private `task-outbox` leaf beneath that base and
+ preserves it across reconfiguration; otherwise journals share
+ `LATTICE_LOG_STATE_DIR`.
- `LATTICE_NO_EXEC=1` is the hard kill switch and overrides execution/terminal
enablement.
- `LATTICE_AGENT_RUN_USER` / `LATTICE_AGENT_RUN_GROUP` configure an optional
@@ -376,6 +391,9 @@ node id.
scripts default to owner-only access.
- Leased tasks carry a server-issued `lease_id`; the agent returns it with the
result and exposes it to the task as `LATTICE_TASK_LEASE_ID` for traceability.
+- A lease must be durably journaled before execution. Completed or
+ unknown-outcome results remain in the outbox until the server acknowledges
+ them, and are flushed before the agent fetches any new task.
- Leased task payloads contain only execution fields; control-plane actor/token
metadata is not sent to agents.
diff --git a/cmd/lattice-agent/guard_reality_test.go b/cmd/lattice-agent/guard_reality_test.go
index 53b3836..8e57d6e 100644
--- a/cmd/lattice-agent/guard_reality_test.go
+++ b/cmd/lattice-agent/guard_reality_test.go
@@ -1,11 +1,13 @@
package main
import (
+ "bytes"
"context"
"encoding/json"
"errors"
"net/http"
"reflect"
+ "regexp"
"testing"
"time"
@@ -13,6 +15,61 @@ import (
"github.com/LatticeNet/lattice-sdk/model"
)
+func TestWriteGuardManagedSHAOnlyOutputsCanonicalHashOnSuccess(t *testing.T) {
+ valid := "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
+ tests := []struct {
+ name string
+ collect func(context.Context, guardreality.Source) (string, error)
+ want string
+ wantErr bool
+ }{
+ {
+ name: "success",
+ collect: func(context.Context, guardreality.Source) (string, error) {
+ return valid, nil
+ },
+ want: valid + "\n",
+ },
+ {
+ name: "collection failure",
+ collect: func(context.Context, guardreality.Source) (string, error) {
+ return "", errors.New("nft unavailable")
+ },
+ wantErr: true,
+ },
+ {
+ name: "invalid collector value",
+ collect: func(context.Context, guardreality.Source) (string, error) {
+ return "NOT-A-SHA", nil
+ },
+ wantErr: true,
+ },
+ }
+
+ for _, tc := range tests {
+ t.Run(tc.name, func(t *testing.T) {
+ var out bytes.Buffer
+ err := writeGuardManagedSHA(context.Background(), &out, tc.collect)
+ if (err != nil) != tc.wantErr {
+ t.Fatalf("error = %v, wantErr=%v", err, tc.wantErr)
+ }
+ if out.String() != tc.want {
+ t.Fatalf("stdout = %q, want %q", out.String(), tc.want)
+ }
+ if out.Len() > 0 && !regexp.MustCompile(`^[0-9a-f]{64}\n$`).Match(out.Bytes()) {
+ t.Fatalf("successful stdout is not one lowercase SHA-256: %q", out.String())
+ }
+ })
+ }
+}
+
+func TestReportedCapabilitiesAdvertiseGuardManagedSHA(t *testing.T) {
+ got := reportedCapabilities()
+ if !reflect.DeepEqual(got, []string{guardManagedSHACapability}) {
+ t.Fatalf("reported capabilities = %#v", got)
+ }
+}
+
func TestReportGuardReality(t *testing.T) {
originalClient := httpClient
t.Cleanup(func() { httpClient = originalClient })
diff --git a/cmd/lattice-agent/main.go b/cmd/lattice-agent/main.go
index 90646ee..635f676 100644
--- a/cmd/lattice-agent/main.go
+++ b/cmd/lattice-agent/main.go
@@ -3,6 +3,7 @@ package main
import (
"bytes"
"context"
+ "crypto/sha256"
"encoding/json"
"errors"
"flag"
@@ -15,6 +16,7 @@ import (
"net/url"
"os"
"os/exec"
+ "path/filepath"
"regexp"
"sort"
"strconv"
@@ -31,6 +33,7 @@ import (
"github.com/LatticeNet/lattice-node-agent/internal/singboxdiscover"
"github.com/LatticeNet/lattice-node-agent/internal/sshwatch"
"github.com/LatticeNet/lattice-node-agent/internal/taskexec"
+ "github.com/LatticeNet/lattice-node-agent/internal/taskoutbox"
"github.com/LatticeNet/lattice-sdk/model"
)
@@ -63,8 +66,14 @@ const (
defaultDebugMaxBatchLines = 100
debugSinkMaxLines = 1000
guardRealityReportTimeout = 10 * time.Second
+ guardManagedSHACapability = "netguard-managed-sha-v1"
+ agentCapabilitiesHeader = "X-Lattice-Agent-Capabilities"
)
+func reportedCapabilities() []string {
+ return []string{guardManagedSHACapability}
+}
+
type agentConfig struct {
Server string
NodeID string
@@ -144,6 +153,7 @@ type agentConfig struct {
SingBoxBin string
SingBoxMeta string
LogStateDir string
+ TaskOutboxDir string
}
type agentRuntimePayload struct {
@@ -178,6 +188,7 @@ func main() {
var cfg agentConfig
var printVersion bool
var printCompat bool
+ var printGuardManagedSHA bool
flag.StringVar(&cfg.Server, "server", env("LATTICE_SERVER", "http://127.0.0.1:8088"), "server base URL")
flag.StringVar(&cfg.NodeID, "node-id", os.Getenv("LATTICE_NODE_ID"), "node id")
flag.StringVar(&cfg.Token, "token", os.Getenv("LATTICE_NODE_TOKEN"), "node enrollment token")
@@ -235,8 +246,10 @@ func main() {
flag.StringVar(&cfg.SingBoxBin, "singbox-bin", env("LATTICE_SINGBOX_BIN", "sb"), "sb management binary for -singbox-discover (default \"sb\" resolved on PATH)")
flag.StringVar(&cfg.SingBoxMeta, "singbox-meta", env("LATTICE_SINGBOX_META", ""), "design-15 sing-box sidecar metadata path for -singbox-discover (default /etc/sing-box/lattice-metadata.json)")
flag.StringVar(&cfg.LogStateDir, "log-state-dir", os.Getenv("LATTICE_LOG_STATE_DIR"), "directory for log-tail checkpoints (empty disables checkpoint persistence; sources still tail from end)")
+ flag.StringVar(&cfg.TaskOutboxDir, "task-outbox-dir", os.Getenv("LATTICE_TASK_OUTBOX_DIR"), "base directory for durable task-result journals (default: log state dir, or the user cache directory for manual runs)")
flag.BoolVar(&printVersion, "version", false, "print lattice-agent version and exit")
flag.BoolVar(&printCompat, "compat-json", false, "print embedded server/dashboard compatibility metadata and exit")
+ flag.BoolVar(&printGuardManagedSHA, "guard-managed-sha", false, "print the canonical SHA-256 of the managed lattice_guard nft table and exit")
flag.Parse()
if printVersion {
fmt.Println(version)
@@ -248,6 +261,12 @@ func main() {
}
return
}
+ if printGuardManagedSHA {
+ if err := writeGuardManagedSHA(context.Background(), os.Stdout, guardreality.CollectManagedTableSHA); err != nil {
+ log.Fatalf("guard managed SHA collection failed: %v", err)
+ }
+ return
+ }
cfg.Debug = cfg.LocalDebug
cfg.DebugMaxLineBytes = defaultDebugMaxLineBytes
cfg.DebugMaxBatchLines = defaultDebugMaxBatchLines
@@ -303,6 +322,22 @@ func main() {
if cfg.NodeID == "" || cfg.Token == "" {
log.Fatal("node-id and token are required")
}
+ outboxDir, err := taskResultOutboxDir(cfg)
+ if err != nil {
+ log.Fatalf("task result outbox path failed: %v", err)
+ }
+ taskResults, err := taskoutbox.Open(outboxDir)
+ if err != nil {
+ log.Fatalf("task result outbox initialization failed: %v", err)
+ }
+ defer taskResults.Close()
+ agentBinary, err := os.Executable()
+ if err != nil {
+ log.Fatalf("resolve lattice-agent executable failed: %v", err)
+ }
+ if !filepath.IsAbs(agentBinary) {
+ log.Fatalf("resolved lattice-agent executable is not absolute: %q", agentBinary)
+ }
if cfg.AllowTerminal && os.Geteuid() == 0 && !cfg.AllowRoot {
log.Printf("warning: terminal sessions disabled because agent is running as root without -allow-root-exec")
cfg.AllowTerminal = false
@@ -318,6 +353,7 @@ func main() {
if err := postAgentJSON(cfg, "/api/agent/hello", map[string]any{
"version": version,
"compatibility": compatibilityPayload(),
+ "capabilities": reportedCapabilities(),
"public_ip": cfg.PublicIP,
"public_ipv6": cfg.PublicIPv6,
"internal_ip": cfg.InternalIP,
@@ -343,7 +379,10 @@ func main() {
go runTerminalLoop(context.Background(), cfg)
}
- runner := taskexec.Runner{AllowExec: cfg.AllowExec, AllowRoot: cfg.AllowRoot, Cgroup: cfg.taskCgroupConfig(), WorkdirRoot: cfg.TaskWorkRoot}
+ runner := taskexec.Runner{
+ AllowExec: cfg.AllowExec, AllowRoot: cfg.AllowRoot, Cgroup: cfg.taskCgroupConfig(),
+ WorkdirRoot: cfg.TaskWorkRoot, AgentBinary: agentBinary,
+ }
monitors := newMonitorManager(cfg)
logTailers := newLogTailManager(cfg)
ticker := time.NewTicker(cfg.Interval)
@@ -366,7 +405,7 @@ func main() {
if err := reportSingBoxInventory(cfg); err != nil {
log.Printf("singbox discover error: %v", err)
}
- if err := runTasks(cfg, runner); err != nil {
+ if err := runTasks(cfg, runner, taskResults); err != nil {
log.Printf("task poll error: %v", err)
}
if assigned, err := fetchMonitors(cfg); err != nil {
@@ -392,6 +431,20 @@ func main() {
}
}
+var guardManagedSHARe = regexp.MustCompile(`^[0-9a-f]{64}$`)
+
+func writeGuardManagedSHA(ctx context.Context, out io.Writer, collect func(context.Context, guardreality.Source) (string, error)) error {
+ sha, err := collect(ctx, guardreality.Source{})
+ if err != nil {
+ return err
+ }
+ if !guardManagedSHARe.MatchString(sha) {
+ return fmt.Errorf("collector returned invalid managed table SHA")
+ }
+ _, err = fmt.Fprintln(out, sha)
+ return err
+}
+
// monitorManager keeps one goroutine per assigned monitor, each probing on its
// own interval. reconcile is called every poll to start new monitors, stop
// removed ones, and restart any whose definition changed.
@@ -624,6 +677,7 @@ func reportMetrics(cfg agentConfig) error {
return postAgentJSON(cfg, "/api/agent/metrics", map[string]any{
"version": version,
"compatibility": compatibilityPayload(),
+ "capabilities": reportedCapabilities(),
"agent_runtime": agentRuntimePayload{
AllowExec: cfg.AllowExec,
AllowRootExec: cfg.AllowRoot,
@@ -1046,12 +1100,37 @@ func resolveProxyUsageSecret(cfg *agentConfig) error {
return nil
}
-func runTasks(cfg agentConfig, runner taskexec.Runner) error {
+type taskRunner interface {
+ Run(model.Task) model.TaskResult
+}
+
+type taskResultOutbox interface {
+ Begin(model.Task) (bool, error)
+ Complete(model.TaskResult) (bool, error)
+ ConfirmDurability() error
+ RecoverInterrupted(string) error
+ Pending() ([]taskoutbox.Entry, error)
+ Remove(taskoutbox.Entry) error
+}
+
+type leasedAgentTask struct {
+ model.Task
+ DurableResult bool `json:"durable_result"`
+}
+
+func runTasks(cfg agentConfig, runner taskRunner, outbox taskResultOutbox) error {
+ if err := outbox.RecoverInterrupted(cfg.NodeID); err != nil {
+ return fmt.Errorf("recover interrupted task results: %w", err)
+ }
+ if err := flushTaskResults(cfg, outbox); err != nil {
+ return err
+ }
req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("%s/api/agent/tasks?node_id=%s", cfg.Server, cfg.NodeID), nil)
if err != nil {
return err
}
req.Header.Set("Authorization", "Bearer "+cfg.Token)
+ req.Header.Set(agentCapabilitiesHeader, strings.Join(reportedCapabilities(), ","))
resp, err := httpClient.Do(req)
if err != nil {
return err
@@ -1060,25 +1139,120 @@ func runTasks(cfg agentConfig, runner taskexec.Runner) error {
if resp.StatusCode != http.StatusOK {
return agentHTTPError(resp, "fetch tasks")
}
- var tasks []model.Task
+ var tasks []leasedAgentTask
if err := json.NewDecoder(resp.Body).Decode(&tasks); err != nil {
return err
}
debugf(cfg, "tasks fetched: count=%d", len(tasks))
- for _, task := range tasks {
+ for _, leased := range tasks {
+ task := leased.Task
+ if !leased.DurableResult {
+ debugf(cfg, "task start without durable-result protocol: id=%s interpreter=%s timeout=%ds", task.ID, task.Interpreter, task.TimeoutSec)
+ result := runner.Run(task)
+ result.NodeID = cfg.NodeID
+ if result.FinishedAt.IsZero() {
+ result.FinishedAt = time.Now().UTC()
+ }
+ if err := postAgentJSON(cfg, "/api/agent/task-result", map[string]any{"result": result}, nil); err != nil {
+ return fmt.Errorf("post task result %s: %w", task.ID, err)
+ }
+ continue
+ }
+ committed, err := outbox.Begin(task)
+ if err != nil {
+ journalErr := fmt.Errorf("journal task lease %s before execution: %w", task.ID, err)
+ if committed {
+ // The lease journal is visible but its directory durability is
+ // uncertain. Do not post a different direct result: the next poll
+ // will recover this exact journal as an unknown outcome.
+ return journalErr
+ }
+ // No host code ran, so it is safe to make one best-effort terminal
+ // report even though durable retry storage is unavailable. This avoids
+ // silently stranding the server lease while preserving the fail-closed
+ // rule that an unjournaled task is never executed.
+ failure := model.TaskResult{
+ TaskID: task.ID, LeaseID: task.LeaseID, NodeID: cfg.NodeID, ExitCode: -1,
+ Error: "task was not executed because its durable lease journal could not be written",
+ FinishedAt: time.Now().UTC(),
+ }
+ if postErr := postAgentJSON(cfg, "/api/agent/task-result", map[string]any{"result": failure}, nil); postErr != nil {
+ return fmt.Errorf("%v; report unexecuted task: %w", journalErr, postErr)
+ }
+ return journalErr
+ }
+ if !committed {
+ // The server intentionally redelivers an unacknowledged lease when a
+ // prior task-poll response may have been lost. An exact existing journal
+ // is the execution authority, so the duplicate response is a no-op.
+ debugf(cfg, "task lease already journaled: id=%s", task.ID)
+ continue
+ }
debugf(cfg, "task start: id=%s interpreter=%s timeout=%ds", task.ID, task.Interpreter, task.TimeoutSec)
result := runner.Run(task)
result.NodeID = cfg.NodeID
debugf(cfg, "task complete: id=%s exit_code=%d error=%t", task.ID, result.ExitCode, result.Error != "")
+ completed, completeErr := outbox.Complete(result)
+ if completeErr != nil {
+ journalErr := fmt.Errorf("journal task result %s before upload: %w", task.ID, completeErr)
+ if completed {
+ // Rename published the exact result even though directory durability
+ // was uncertain. Confirm the local transition before allowing the
+ // server to commit the result; otherwise a crash could expose the old
+ // leased journal and synthesize a conflicting unknown-outcome result.
+ if confirmErr := outbox.ConfirmDurability(); confirmErr != nil {
+ return fmt.Errorf("%v; confirm published task result: %w", journalErr, confirmErr)
+ }
+ if flushErr := flushTaskResults(cfg, outbox); flushErr != nil {
+ return fmt.Errorf("%v; upload confirmed task result: %w", journalErr, flushErr)
+ }
+ }
+ return journalErr
+ }
+ if err := flushTaskResults(cfg, outbox); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func flushTaskResults(cfg agentConfig, outbox taskResultOutbox) error {
+ pending, err := outbox.Pending()
+ if err != nil {
+ return fmt.Errorf("read pending task results: %w", err)
+ }
+ for _, entry := range pending {
+ if entry.Result == nil {
+ return fmt.Errorf("pending task result %s is empty", entry.Task.ID)
+ }
if err := postAgentJSON(cfg, "/api/agent/task-result", map[string]any{
- "result": result,
+ "result": *entry.Result,
}, nil); err != nil {
- return err
+ return fmt.Errorf("flush durable task result %s: %w", entry.Task.ID, err)
+ }
+ if err := outbox.Remove(entry); err != nil {
+ return fmt.Errorf("remove acknowledged task result %s: %w", entry.Task.ID, err)
}
}
return nil
}
+func taskResultOutboxDir(cfg agentConfig) (string, error) {
+ base := strings.TrimSpace(cfg.TaskOutboxDir)
+ if base == "" {
+ base = strings.TrimSpace(cfg.LogStateDir)
+ }
+ if base == "" {
+ cacheDir, err := os.UserCacheDir()
+ if err != nil {
+ return "", err
+ }
+ base = filepath.Join(cacheDir, "lattice-agent")
+ }
+ nodeHash := sha256.Sum256([]byte(cfg.Server + "\x00" + cfg.NodeID))
+ return filepath.Join(base, "task-outbox", fmt.Sprintf("%x", nodeHash[:])), nil
+}
+
func postAgentJSON(cfg agentConfig, path string, payload map[string]any, out any) error {
return postAgentJSONContext(context.Background(), cfg, path, payload, out)
}
diff --git a/cmd/lattice-agent/main_test.go b/cmd/lattice-agent/main_test.go
index 249f0fc..c8dcbfc 100644
--- a/cmd/lattice-agent/main_test.go
+++ b/cmd/lattice-agent/main_test.go
@@ -14,6 +14,7 @@ import (
"testing"
"time"
+ "github.com/LatticeNet/lattice-node-agent/internal/taskoutbox"
"github.com/LatticeNet/lattice-sdk/model"
)
@@ -23,6 +24,19 @@ func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) {
return f(r)
}
+type countingTaskRunner struct {
+ calls int
+ result model.TaskResult
+}
+
+func (r *countingTaskRunner) Run(task model.Task) model.TaskResult {
+ r.calls++
+ result := r.result
+ result.TaskID = task.ID
+ result.LeaseID = task.LeaseID
+ return result
+}
+
func TestVersionMatchesCurrentRelease(t *testing.T) {
if version != "0.3.3" {
t.Fatalf("version = %q, want 0.3.3", version)
@@ -225,6 +239,435 @@ func TestPostJSONReturnsStructuredServerDiagnostics(t *testing.T) {
requireErrorContains(t, err, "request_id=req-task")
}
+func TestRunTasksRetainsResultAcrossTransientServerFailure(t *testing.T) {
+ for _, firstStatus := range []int{http.StatusInternalServerError, http.StatusConflict} {
+ t.Run(http.StatusText(firstStatus), func(t *testing.T) {
+ oldClient := httpClient
+ defer func() { httpClient = oldClient }()
+
+ store, err := taskoutbox.Open(t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ task := model.Task{ID: "task-a", LeaseID: "lease-a", Interpreter: "sh", Script: "echo ok", TimeoutSec: 10, OutputLimit: 1024}
+ runner := &countingTaskRunner{result: model.TaskResult{ExitCode: 0, Stdout: "ok", StartedAt: time.Now().UTC(), FinishedAt: time.Now().UTC()}}
+ postCalls := 0
+ fetchCalls := 0
+ var posted []model.TaskResult
+ httpClient = &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
+ switch r.URL.Path {
+ case "/api/agent/tasks":
+ if r.Header.Get(agentCapabilitiesHeader) != guardManagedSHACapability {
+ return testResponse(http.StatusBadRequest, "missing lease-time capability"), nil
+ }
+ fetchCalls++
+ if fetchCalls == 1 {
+ data, _ := json.Marshal([]leasedAgentTask{{Task: task, DurableResult: true}})
+ return testResponse(http.StatusOK, string(data)), nil
+ }
+ return testResponse(http.StatusOK, `[]`), nil
+ case "/api/agent/task-result":
+ postCalls++
+ var body struct {
+ Result model.TaskResult `json:"result"`
+ }
+ if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+ t.Fatal(err)
+ }
+ posted = append(posted, body.Result)
+ if postCalls == 1 {
+ return testResponse(firstStatus, `{"error":{"code":"retry","message":"retry same lease"}}`), nil
+ }
+ return testResponse(http.StatusOK, `{"ok":true}`), nil
+ default:
+ return testResponse(http.StatusNotFound, ""), nil
+ }
+ })}
+ cfg := agentConfig{Server: "http://lattice.test", NodeID: "node-a", Token: "secret"}
+
+ if err := runTasks(cfg, runner, store); err == nil {
+ t.Fatal("first result upload should fail")
+ }
+ if pending, err := store.Pending(); err != nil || len(pending) != 1 {
+ t.Fatalf("pending after failed upload = %+v, err=%v", pending, err)
+ }
+ if err := runTasks(cfg, runner, store); err != nil {
+ t.Fatal(err)
+ }
+ if runner.calls != 1 {
+ t.Fatalf("runner calls = %d, want 1", runner.calls)
+ }
+ if postCalls != 2 || len(posted) != 2 || !reflect.DeepEqual(posted[0], posted[1]) {
+ t.Fatalf("result retry changed: calls=%d posted=%+v", postCalls, posted)
+ }
+ if pending, err := store.Pending(); err != nil || len(pending) != 0 {
+ t.Fatalf("pending after acknowledgement = %+v, err=%v", pending, err)
+ }
+ })
+ }
+}
+
+func TestRunTasksRestartFlushesCompletedResultBeforeFetch(t *testing.T) {
+ oldClient := httpClient
+ defer func() { httpClient = oldClient }()
+ dir := t.TempDir()
+ store, err := taskoutbox.Open(dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ task := model.Task{ID: "task-a", LeaseID: "lease-a", Interpreter: "sh", Script: "echo ok"}
+ if _, err := store.Begin(task); err != nil {
+ t.Fatal(err)
+ }
+ result := model.TaskResult{TaskID: task.ID, LeaseID: task.LeaseID, NodeID: "node-a", ExitCode: 0, Stdout: "persisted"}
+ if _, err := store.Complete(result); err != nil {
+ t.Fatal(err)
+ }
+ if err := store.Close(); err != nil {
+ t.Fatal(err)
+ }
+ restarted, err := taskoutbox.Open(dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer restarted.Close()
+ runner := &countingTaskRunner{}
+ var order []string
+ httpClient = &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
+ order = append(order, r.URL.Path)
+ if r.URL.Path == "/api/agent/task-result" {
+ return testResponse(http.StatusOK, `{"ok":true}`), nil
+ }
+ return testResponse(http.StatusOK, `[]`), nil
+ })}
+ if err := runTasks(agentConfig{Server: "http://lattice.test", NodeID: "node-a", Token: "secret"}, runner, restarted); err != nil {
+ t.Fatal(err)
+ }
+ if runner.calls != 0 {
+ t.Fatalf("completed task was re-executed: calls=%d", runner.calls)
+ }
+ wantOrder := []string{"/api/agent/task-result", "/api/agent/tasks"}
+ if !reflect.DeepEqual(order, wantOrder) {
+ t.Fatalf("request order = %v, want %v", order, wantOrder)
+ }
+}
+
+func TestRunTasksRestartConvertsInterruptedLeaseToUnknownOutcome(t *testing.T) {
+ oldClient := httpClient
+ defer func() { httpClient = oldClient }()
+ dir := t.TempDir()
+ store, err := taskoutbox.Open(dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ if _, err := store.Begin(model.Task{ID: "task-a", LeaseID: "lease-a", Interpreter: "sh", Script: "mutate host"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := store.Close(); err != nil {
+ t.Fatal(err)
+ }
+ restarted, err := taskoutbox.Open(dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer restarted.Close()
+ runner := &countingTaskRunner{}
+ var posted model.TaskResult
+ httpClient = &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
+ if r.URL.Path == "/api/agent/task-result" {
+ var body struct {
+ Result model.TaskResult `json:"result"`
+ }
+ if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+ t.Fatal(err)
+ }
+ posted = body.Result
+ return testResponse(http.StatusOK, `{"ok":true}`), nil
+ }
+ return testResponse(http.StatusOK, `[]`), nil
+ })}
+ if err := runTasks(agentConfig{Server: "http://lattice.test", NodeID: "node-a", Token: "secret"}, runner, restarted); err != nil {
+ t.Fatal(err)
+ }
+ if runner.calls != 0 {
+ t.Fatalf("interrupted task was re-executed: calls=%d", runner.calls)
+ }
+ if posted.ExitCode != -1 || !strings.Contains(posted.Error, "outcome is unknown") || !strings.Contains(posted.Error, "not re-executed") {
+ t.Fatalf("interrupted result is not honest: %+v", posted)
+ }
+}
+
+type beginFailingOutbox struct {
+ committed bool
+ err error
+}
+
+func (o beginFailingOutbox) Begin(model.Task) (bool, error) { return o.committed, o.err }
+func (o beginFailingOutbox) Complete(model.TaskResult) (bool, error) { return true, nil }
+func (o beginFailingOutbox) ConfirmDurability() error { return nil }
+func (o beginFailingOutbox) RecoverInterrupted(string) error { return nil }
+func (o beginFailingOutbox) Pending() ([]taskoutbox.Entry, error) { return nil, nil }
+func (o beginFailingOutbox) Remove(taskoutbox.Entry) error { return nil }
+
+func TestRunTasksJournalFailurePreventsExecution(t *testing.T) {
+ oldClient := httpClient
+ defer func() { httpClient = oldClient }()
+ task := model.Task{ID: "task-a", LeaseID: "lease-a", Interpreter: "sh", Script: "must not run"}
+ data, _ := json.Marshal([]leasedAgentTask{{Task: task, DurableResult: true}})
+ var reported model.TaskResult
+ httpClient = &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
+ if r.URL.Path == "/api/agent/tasks" {
+ return testResponse(http.StatusOK, string(data)), nil
+ }
+ var body struct {
+ Result model.TaskResult `json:"result"`
+ }
+ if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+ t.Fatal(err)
+ }
+ reported = body.Result
+ return testResponse(http.StatusOK, `{"ok":true}`), nil
+ })}
+ runner := &countingTaskRunner{}
+ err := runTasks(
+ agentConfig{Server: "http://lattice.test", NodeID: "node-a", Token: "secret"},
+ runner,
+ beginFailingOutbox{err: errors.New("disk full")},
+ )
+ if err == nil || !strings.Contains(err.Error(), "disk full") {
+ t.Fatalf("runTasks error = %v, want disk failure", err)
+ }
+ if runner.calls != 0 {
+ t.Fatalf("task ran despite journal failure: calls=%d", runner.calls)
+ }
+ if reported.TaskID != task.ID || reported.LeaseID != task.LeaseID || reported.ExitCode != -1 ||
+ !strings.Contains(reported.Error, "not executed") {
+ t.Fatalf("journal failure did not report an honest terminal result: %+v", reported)
+ }
+}
+
+func TestRunTasksPublishedJournalFailureDoesNotPostConflictingDirectResult(t *testing.T) {
+ oldClient := httpClient
+ defer func() { httpClient = oldClient }()
+ task := model.Task{ID: "task-a", LeaseID: "lease-a", Interpreter: "sh", Script: "must not run"}
+ data, _ := json.Marshal([]leasedAgentTask{{Task: task, DurableResult: true}})
+ postCalls := 0
+ httpClient = &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
+ if r.URL.Path == "/api/agent/tasks" {
+ return testResponse(http.StatusOK, string(data)), nil
+ }
+ postCalls++
+ return testResponse(http.StatusOK, `{"ok":true}`), nil
+ })}
+ runner := &countingTaskRunner{}
+ err := runTasks(
+ agentConfig{Server: "http://lattice.test", NodeID: "node-a", Token: "secret"},
+ runner,
+ beginFailingOutbox{committed: true, err: errors.New("directory sync failed")},
+ )
+ if err == nil || !strings.Contains(err.Error(), "directory sync failed") {
+ t.Fatalf("runTasks error = %v, want directory sync failure", err)
+ }
+ if runner.calls != 0 || postCalls != 0 {
+ t.Fatalf("published journal ambiguity ran or directly reported task: runner=%d posts=%d", runner.calls, postCalls)
+ }
+}
+
+func TestRunTasksExactRedeliveryDoesNotExecuteExistingJournal(t *testing.T) {
+ oldClient := httpClient
+ defer func() { httpClient = oldClient }()
+ task := model.Task{ID: "task-a", LeaseID: "lease-a", Interpreter: "sh", Script: "must run once"}
+ data, _ := json.Marshal([]leasedAgentTask{{Task: task, DurableResult: true}})
+ httpClient = &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
+ if r.URL.Path != "/api/agent/tasks" {
+ t.Fatalf("unexpected request for already-journaled lease: %s", r.URL.Path)
+ }
+ return testResponse(http.StatusOK, string(data)), nil
+ })}
+ runner := &countingTaskRunner{}
+ if err := runTasks(
+ agentConfig{Server: "http://lattice.test", NodeID: "node-a", Token: "secret"},
+ runner,
+ beginFailingOutbox{committed: false},
+ ); err != nil {
+ t.Fatal(err)
+ }
+ if runner.calls != 0 {
+ t.Fatalf("exact redelivery executed an existing journal: calls=%d", runner.calls)
+ }
+}
+
+func TestRunTasksKeepsGenericTasksOutsideDurableNetGuardProtocol(t *testing.T) {
+ oldClient := httpClient
+ defer func() { httpClient = oldClient }()
+ task := model.Task{ID: "task-generic", LeaseID: "lease-generic", Interpreter: "sh", Script: "echo generic"}
+ data, _ := json.Marshal([]leasedAgentTask{{Task: task}})
+ posts := 0
+ httpClient = &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
+ switch r.URL.Path {
+ case "/api/agent/tasks":
+ return testResponse(http.StatusOK, string(data)), nil
+ case "/api/agent/task-result":
+ posts++
+ return testResponse(http.StatusOK, `{"ok":true}`), nil
+ default:
+ return testResponse(http.StatusNotFound, ""), nil
+ }
+ })}
+ runner := &countingTaskRunner{result: model.TaskResult{TaskID: task.ID, LeaseID: task.LeaseID}}
+ if err := runTasks(
+ agentConfig{Server: "http://lattice.test", NodeID: "node-a", Token: "secret"},
+ runner,
+ beginFailingOutbox{err: errors.New("generic task must not journal")},
+ ); err != nil {
+ t.Fatal(err)
+ }
+ if runner.calls != 1 || posts != 1 {
+ t.Fatalf("generic task delivery = runner %d posts %d, want one direct execution/result", runner.calls, posts)
+ }
+}
+
+type completePublishingOutbox struct {
+ task model.Task
+ result *model.TaskResult
+ confirmErr error
+ confirmed int
+ removed int
+}
+
+func (o *completePublishingOutbox) Begin(task model.Task) (bool, error) {
+ o.task = task
+ return true, nil
+}
+
+func (o *completePublishingOutbox) Complete(result model.TaskResult) (bool, error) {
+ o.result = &result
+ return true, errors.New("directory sync failed")
+}
+
+func (o *completePublishingOutbox) ConfirmDurability() error {
+ o.confirmed++
+ return o.confirmErr
+}
+
+func (o *completePublishingOutbox) RecoverInterrupted(string) error { return nil }
+
+func (o *completePublishingOutbox) Pending() ([]taskoutbox.Entry, error) {
+ if o.result == nil {
+ return nil, nil
+ }
+ return []taskoutbox.Entry{{Task: o.task, Result: o.result}}, nil
+}
+
+func (o *completePublishingOutbox) Remove(taskoutbox.Entry) error {
+ o.removed++
+ o.result = nil
+ return nil
+}
+
+func TestRunTasksConfirmsAndUploadsResultPublishedBeforeDirectorySyncFailure(t *testing.T) {
+ oldClient := httpClient
+ defer func() { httpClient = oldClient }()
+ task := model.Task{ID: "task-a", LeaseID: "lease-a", Interpreter: "sh", Script: "echo once"}
+ data, _ := json.Marshal([]leasedAgentTask{{Task: task, DurableResult: true}})
+ posts := 0
+ var posted model.TaskResult
+ outbox := &completePublishingOutbox{}
+ httpClient = &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
+ switch r.URL.Path {
+ case "/api/agent/tasks":
+ return testResponse(http.StatusOK, string(data)), nil
+ case "/api/agent/task-result":
+ if outbox.confirmed != 1 {
+ t.Fatalf("result posted before local durability confirmation: confirms=%d", outbox.confirmed)
+ }
+ posts++
+ var body struct {
+ Result model.TaskResult `json:"result"`
+ }
+ if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+ t.Fatal(err)
+ }
+ posted = body.Result
+ return testResponse(http.StatusOK, `{"ok":true}`), nil
+ default:
+ return testResponse(http.StatusNotFound, ""), nil
+ }
+ })}
+ runner := &countingTaskRunner{result: model.TaskResult{TaskID: task.ID, LeaseID: task.LeaseID, ExitCode: 0, Stdout: "exact"}}
+ err := runTasks(agentConfig{Server: "http://lattice.test", NodeID: "node-a", Token: "secret"}, runner, outbox)
+ if err == nil || !strings.Contains(err.Error(), "directory sync failed") {
+ t.Fatalf("runTasks error = %v, want published-result sync warning", err)
+ }
+ if runner.calls != 1 || posts != 1 || outbox.confirmed != 1 || outbox.removed != 1 {
+ t.Fatalf("published result was not confirmed and uploaded exactly once: runner=%d confirms=%d posts=%d removed=%d", runner.calls, outbox.confirmed, posts, outbox.removed)
+ }
+ if posted.TaskID != task.ID || posted.LeaseID != task.LeaseID || posted.NodeID != "node-a" || posted.Stdout != "exact" {
+ t.Fatalf("immediate upload changed result: %+v", posted)
+ }
+}
+
+func TestRunTasksDoesNotUploadUnconfirmedPublishedResult(t *testing.T) {
+ oldClient := httpClient
+ defer func() { httpClient = oldClient }()
+ task := model.Task{ID: "task-a", LeaseID: "lease-a", Interpreter: "sh", Script: "echo once"}
+ data, _ := json.Marshal([]leasedAgentTask{{Task: task, DurableResult: true}})
+ posts := 0
+ httpClient = &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
+ switch r.URL.Path {
+ case "/api/agent/tasks":
+ return testResponse(http.StatusOK, string(data)), nil
+ case "/api/agent/task-result":
+ posts++
+ return testResponse(http.StatusOK, `{"ok":true}`), nil
+ default:
+ return testResponse(http.StatusNotFound, ""), nil
+ }
+ })}
+ runner := &countingTaskRunner{result: model.TaskResult{TaskID: task.ID, LeaseID: task.LeaseID, ExitCode: 0}}
+ outbox := &completePublishingOutbox{confirmErr: errors.New("directory still unavailable")}
+ err := runTasks(agentConfig{Server: "http://lattice.test", NodeID: "node-a", Token: "secret"}, runner, outbox)
+ if err == nil || !strings.Contains(err.Error(), "confirm published task result") {
+ t.Fatalf("runTasks error = %v, want durability confirmation error", err)
+ }
+ if runner.calls != 1 || outbox.confirmed != 1 || posts != 0 || outbox.removed != 0 {
+ t.Fatalf("unconfirmed result escaped locally: runner=%d confirms=%d posts=%d removed=%d", runner.calls, outbox.confirmed, posts, outbox.removed)
+ }
+}
+
+func TestTaskResultOutboxDirPrefersConfiguredStateAndIsolatedIdentity(t *testing.T) {
+ base := t.TempDir()
+ one, err := taskResultOutboxDir(agentConfig{
+ Server: "https://one.example",
+ NodeID: "node-a",
+ LogStateDir: base,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ two, err := taskResultOutboxDir(agentConfig{
+ Server: "https://two.example",
+ NodeID: "node-a",
+ LogStateDir: base,
+ TaskOutboxDir: filepath.Join(base, "override"),
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !strings.HasPrefix(one, filepath.Join(base, "task-outbox")+string(os.PathSeparator)) {
+ t.Fatalf("LogStateDir was not used: %s", one)
+ }
+ if !strings.HasPrefix(two, filepath.Join(base, "override", "task-outbox")+string(os.PathSeparator)) {
+ t.Fatalf("TaskOutboxDir override was not used: %s", two)
+ }
+ if one == two || filepath.Base(one) == filepath.Base(two) {
+ t.Fatalf("server/node identities were not isolated: one=%s two=%s", one, two)
+ }
+}
+
func TestShipLogBatchReturnsStatusAndStructuredDiagnostics(t *testing.T) {
oldClient := httpClient
defer func() { httpClient = oldClient }()
diff --git a/internal/guardreality/collect.go b/internal/guardreality/collect.go
index 6afabee..ac15e89 100644
--- a/internal/guardreality/collect.go
+++ b/internal/guardreality/collect.go
@@ -43,6 +43,65 @@ type Source struct {
Runner Runner
}
+// CollectManagedTableSHA reads the current nftables ruleset and returns the
+// canonical hash of the managed inet lattice_guard table. It fails closed when
+// the command, JSON parsing, or managed-table lookup fails.
+func CollectManagedTableSHA(ctx context.Context, source Source) (string, error) {
+ if ctx == nil {
+ ctx = context.Background()
+ }
+ timeout := source.Timeout
+ if timeout <= 0 {
+ timeout = defaultTimeout
+ }
+ run := source.Runner
+ if run == nil {
+ run = runBoundedCommand
+ }
+ nftBinary := firstNonEmpty(source.NFTBinary, "nft")
+
+ rulesetOut, err := runStep(ctx, timeout, run, nftBinary, "-j", "list", "ruleset")
+ if err != nil {
+ return "", err
+ }
+ managedSHA, _, err := ParseNFTRuleset(rulesetOut)
+ if err != nil {
+ return "", fmt.Errorf("parse nft ruleset: %w", err)
+ }
+ if managedSHA == "" {
+ return "", fmt.Errorf("managed nft table %s %s not found", managedFamily, managedTable)
+ }
+ hasContent, err := managedTableHasContent(rulesetOut)
+ if err != nil {
+ return "", fmt.Errorf("inspect managed nft table: %w", err)
+ }
+ if !hasContent {
+ return "", fmt.Errorf("managed nft table %s %s is empty", managedFamily, managedTable)
+ }
+ return managedSHA, nil
+}
+
+func managedTableHasContent(raw []byte) (bool, error) {
+ var payload struct {
+ NFTables []map[string]any `json:"nftables"`
+ }
+ if err := json.Unmarshal(bytes.TrimSpace(raw), &payload); err != nil {
+ return false, err
+ }
+ for _, entry := range payload.NFTables {
+ for kind, rawBody := range entry {
+ if kind == "table" {
+ continue
+ }
+ body, ok := rawBody.(map[string]any)
+ if ok && nftObjectBelongsToManaged(kind, body) {
+ return true, nil
+ }
+ }
+ }
+ return false, nil
+}
+
// Collect runs the read-only guard reality commands and normalizes their output
// into the shared SDK model. The caller-supplied node id wins over anything a
// command could report.
diff --git a/internal/guardreality/collect_test.go b/internal/guardreality/collect_test.go
index 81fe903..db09131 100644
--- a/internal/guardreality/collect_test.go
+++ b/internal/guardreality/collect_test.go
@@ -82,6 +82,63 @@ func TestCollectPropagatesCommandFailure(t *testing.T) {
}
}
+func TestCollectManagedTableSHA(t *testing.T) {
+ collect := func(ruleset string, runnerErr error) (string, []string, error) {
+ calls := []string{}
+ got, err := CollectManagedTableSHA(context.Background(), Source{
+ NFTBinary: "custom-nft",
+ Runner: func(_ context.Context, name string, args ...string) ([]byte, error) {
+ calls = append(calls, name+" "+strings.Join(args, " "))
+ if runnerErr != nil {
+ return nil, runnerErr
+ }
+ return []byte(ruleset), nil
+ },
+ })
+ return got, calls, err
+ }
+
+ first, calls, err := collect(nftFixture("11", "22"), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !reflect.DeepEqual(calls, []string{"custom-nft -j list ruleset"}) {
+ t.Fatalf("commands = %#v, want one bounded ruleset read", calls)
+ }
+ second, _, err := collect(nftFixture("99", "100"), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if first == "" || first != second {
+ t.Fatalf("managed hash should be stable across handle churn: %q vs %q", first, second)
+ }
+ changed, _, err := collect(strings.ReplaceAll(nftFixture("11", "22"), `"right": 22`, `"right": 2222`), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if changed == first {
+ t.Fatal("managed hash must change when a managed rule changes")
+ }
+
+ missing, _, err := collect(`{"nftables":[{"table":{"family":"ip","name":"filter"}}]}`, nil)
+ if err == nil || missing != "" || !strings.Contains(err.Error(), "managed nft table inet lattice_guard not found") {
+ t.Fatalf("missing table = %q, %v; want fail-closed error", missing, err)
+ }
+ empty, _, err := collect(`{"nftables":[{"table":{"family":"inet","name":"lattice_guard"}}]}`, nil)
+ if err == nil || empty != "" || !strings.Contains(err.Error(), "managed nft table inet lattice_guard is empty") {
+ t.Fatalf("empty managed table = %q, %v; want fail-closed error", empty, err)
+ }
+ malformed, _, err := collect(`{"nftables":`, nil)
+ if err == nil || malformed != "" || !strings.Contains(err.Error(), "parse nft ruleset") {
+ t.Fatalf("malformed ruleset = %q, %v; want fail-closed parse error", malformed, err)
+ }
+ boom := errors.New("nft unavailable")
+ failed, _, err := collect("", boom)
+ if !errors.Is(err, boom) || failed != "" || !strings.Contains(err.Error(), "custom-nft -j list ruleset") {
+ t.Fatalf("runner failure = %q, %v; want contextual fail-closed error", failed, err)
+ }
+}
+
func TestParseSSListenersSkipsNonNumericPortsAndSorts(t *testing.T) {
got, err := ParseSSListeners([]byte(strings.Join([]string{
`udp UNCONN 0 0 *:51820 *:* users:(("wg",pid=9,fd=3))`,
diff --git a/internal/taskexec/taskexec.go b/internal/taskexec/taskexec.go
index db3fa42..e6c26b3 100644
--- a/internal/taskexec/taskexec.go
+++ b/internal/taskexec/taskexec.go
@@ -159,6 +159,10 @@ type Runner struct {
// When set, it must be an absolute non-group/world-writable directory; task
// launch fails before script execution if the root cannot be prepared safely.
WorkdirRoot string
+ // AgentBinary is the absolute path of the running lattice-agent executable.
+ // Trusted server-authored scripts use it for one-shot helper modes without
+ // depending on PATH or inheriting the service process environment.
+ AgentBinary string
// getUID returns the effective uid of the agent process. It is a field so
// tests can simulate "running as root" without actually being root. When
// nil it defaults to os.Geteuid.
@@ -381,6 +385,9 @@ func (r Runner) Run(task model.Task) model.TaskResult {
"LATTICE_TASK_ID=" + task.ID,
"LATTICE_TASK_LEASE_ID=" + task.LeaseID,
}
+ if filepath.IsAbs(r.AgentBinary) {
+ cmd.Env = append(cmd.Env, "LATTICE_AGENT_BIN="+r.AgentBinary)
+ }
var stdout, stderr cappedBuffer
stdout.limit = limit
diff --git a/internal/taskexec/taskexec_test.go b/internal/taskexec/taskexec_test.go
index ff7886a..9e169c7 100644
--- a/internal/taskexec/taskexec_test.go
+++ b/internal/taskexec/taskexec_test.go
@@ -186,6 +186,21 @@ func TestRunnerPropagatesLeaseID(t *testing.T) {
}
}
+func TestRunnerPropagatesAbsoluteAgentBinary(t *testing.T) {
+ agentBinary := filepath.Join(t.TempDir(), "lattice-agent")
+ r := Runner{AllowExec: true, AgentBinary: agentBinary, getUID: nonRootUID}
+ result := r.Run(model.Task{
+ ID: "task_agent_binary",
+ Interpreter: "sh",
+ Script: `printf '%s' "$LATTICE_AGENT_BIN"`,
+ TimeoutSec: 5,
+ OutputLimit: 512,
+ })
+ if result.ExitCode != 0 || result.Stdout != agentBinary {
+ t.Fatalf("agent binary env = %#v, want %q", result, agentBinary)
+ }
+}
+
func TestRunnerSetsPrivateTaskTempEnvironment(t *testing.T) {
r := Runner{AllowExec: true, getUID: nonRootUID}
result := r.Run(model.Task{
diff --git a/internal/taskoutbox/lock_other.go b/internal/taskoutbox/lock_other.go
new file mode 100644
index 0000000..cca2dc9
--- /dev/null
+++ b/internal/taskoutbox/lock_other.go
@@ -0,0 +1,14 @@
+//go:build !linux && !darwin && !freebsd
+
+package taskoutbox
+
+import (
+ "fmt"
+ "os"
+)
+
+func lockOutbox(string) (*os.File, error) {
+ return nil, fmt.Errorf("durable task outbox locking is unsupported on this operating system")
+}
+
+func unlockOutbox(f *os.File) error { return f.Close() }
diff --git a/internal/taskoutbox/lock_unix.go b/internal/taskoutbox/lock_unix.go
new file mode 100644
index 0000000..516dba9
--- /dev/null
+++ b/internal/taskoutbox/lock_unix.go
@@ -0,0 +1,57 @@
+//go:build linux || darwin || freebsd
+
+package taskoutbox
+
+import (
+ "fmt"
+ "os"
+ "syscall"
+)
+
+func lockOutbox(path string) (*os.File, error) {
+ fd, err := syscall.Open(path, syscall.O_CREAT|syscall.O_RDWR|syscall.O_NOFOLLOW|syscall.O_CLOEXEC, 0o600)
+ if err != nil {
+ return nil, fmt.Errorf("open task result outbox lock: %w", err)
+ }
+ f := os.NewFile(uintptr(fd), path)
+ if f == nil {
+ _ = syscall.Close(fd)
+ return nil, fmt.Errorf("open task result outbox lock: invalid file descriptor")
+ }
+ info, err := f.Stat()
+ if err != nil {
+ f.Close()
+ return nil, fmt.Errorf("inspect task result outbox lock: %w", err)
+ }
+ stat, ok := info.Sys().(*syscall.Stat_t)
+ if !ok {
+ f.Close()
+ return nil, fmt.Errorf("inspect task result outbox lock ownership: unsupported stat data")
+ }
+ if !info.Mode().IsRegular() {
+ f.Close()
+ return nil, fmt.Errorf("task result outbox lock must be a regular file")
+ }
+ if stat.Uid != uint32(os.Geteuid()) {
+ f.Close()
+ return nil, fmt.Errorf("task result outbox lock must be owned by effective user %d", os.Geteuid())
+ }
+ if info.Mode().Perm() != 0o600 {
+ f.Close()
+ return nil, fmt.Errorf("task result outbox lock permissions are %o, want 600", info.Mode().Perm())
+ }
+ if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
+ f.Close()
+ return nil, fmt.Errorf("task result outbox is already owned by another agent process: %w", err)
+ }
+ return f, nil
+}
+
+func unlockOutbox(f *os.File) error {
+ unlockErr := syscall.Flock(int(f.Fd()), syscall.LOCK_UN)
+ closeErr := f.Close()
+ if unlockErr != nil {
+ return fmt.Errorf("unlock task result outbox: %w", unlockErr)
+ }
+ return closeErr
+}
diff --git a/internal/taskoutbox/lock_unix_test.go b/internal/taskoutbox/lock_unix_test.go
new file mode 100644
index 0000000..876d2f5
--- /dev/null
+++ b/internal/taskoutbox/lock_unix_test.go
@@ -0,0 +1,66 @@
+//go:build linux || darwin || freebsd
+
+package taskoutbox
+
+import (
+ "os"
+ "path/filepath"
+ "testing"
+)
+
+func TestOpenRefusesSymlinkLockWithoutChangingTarget(t *testing.T) {
+ dir := t.TempDir()
+ victim := filepath.Join(t.TempDir(), "victim")
+ const contents = "do not touch"
+ if err := os.WriteFile(victim, []byte(contents), 0o640); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chmod(victim, 0o640); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(victim, filepath.Join(dir, ".lock")); err != nil {
+ t.Fatal(err)
+ }
+
+ if store, err := Open(dir); err == nil {
+ _ = store.Close()
+ t.Fatal("Open() succeeded with a symlink lock, want rejection")
+ }
+ got, err := os.ReadFile(victim)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if string(got) != contents {
+ t.Fatalf("victim content = %q, want %q", got, contents)
+ }
+ info, err := os.Stat(victim)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if info.Mode().Perm() != 0o640 {
+ t.Fatalf("victim mode = %o, want 640", info.Mode().Perm())
+ }
+}
+
+func TestOpenRefusesInsecurePreexistingLockPermissions(t *testing.T) {
+ dir := t.TempDir()
+ lockPath := filepath.Join(dir, ".lock")
+ if err := os.WriteFile(lockPath, nil, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chmod(lockPath, 0o644); err != nil {
+ t.Fatal(err)
+ }
+
+ if store, err := Open(dir); err == nil {
+ _ = store.Close()
+ t.Fatal("Open() succeeded with an insecure preexisting lock, want rejection")
+ }
+ info, err := os.Stat(lockPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if info.Mode().Perm() != 0o644 {
+ t.Fatalf("lock mode = %o, want unchanged 644", info.Mode().Perm())
+ }
+}
diff --git a/internal/taskoutbox/outbox.go b/internal/taskoutbox/outbox.go
new file mode 100644
index 0000000..66d835a
--- /dev/null
+++ b/internal/taskoutbox/outbox.go
@@ -0,0 +1,495 @@
+// Package taskoutbox persists leased tasks and their results so an agent
+// restart or transient control-plane failure cannot cause silent re-execution
+// or result loss.
+package taskoutbox
+
+import (
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+ "reflect"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/LatticeNet/lattice-sdk/model"
+)
+
+const (
+ entryVersion = 1
+ stateLeased = "leased"
+ stateDone = "completed"
+ maxEntries = 1024
+ maxEntrySize = 4 << 20 // 4 MiB: bounds scripts plus capped task output.
+)
+
+var ErrCapacity = errors.New("task result outbox capacity exceeded")
+
+// Entry is a durable task execution journal record.
+type Entry struct {
+ Version int `json:"version"`
+ State string `json:"state"`
+ Task model.Task `json:"task"`
+ Result *model.TaskResult `json:"result,omitempty"`
+ ExecutionStartedAt time.Time `json:"execution_started_at"`
+ UpdatedAt time.Time `json:"updated_at"`
+ key string
+}
+
+// Store is a bounded, file-backed task result outbox.
+type Store struct {
+ dir string
+ syncDir func(string) error
+ lock *os.File
+ durabilityUnconfirmed bool
+}
+
+// Open creates or validates a private outbox directory.
+func Open(dir string) (*Store, error) {
+ return openWithSync(dir, syncDir)
+}
+
+func openWithSync(dir string, syncDirectory func(string) error) (*Store, error) {
+ dir = strings.TrimSpace(dir)
+ if dir == "" {
+ return nil, fmt.Errorf("task result outbox directory is empty")
+ }
+ if !filepath.IsAbs(dir) {
+ return nil, fmt.Errorf("task result outbox directory must be absolute: %q", dir)
+ }
+ if syncDirectory == nil {
+ syncDirectory = syncDir
+ }
+ if err := makeDurablePrivateDir(dir, syncDirectory); err != nil {
+ return nil, fmt.Errorf("create task result outbox: %w", err)
+ }
+ info, err := os.Lstat(dir)
+ if err != nil {
+ return nil, fmt.Errorf("inspect task result outbox: %w", err)
+ }
+ if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
+ return nil, fmt.Errorf("task result outbox must be a real directory: %s", dir)
+ }
+ if err := os.Chmod(dir, 0o700); err != nil {
+ return nil, fmt.Errorf("secure task result outbox: %w", err)
+ }
+ lock, err := lockOutbox(filepath.Join(dir, ".lock"))
+ if err != nil {
+ return nil, err
+ }
+ store := &Store{dir: dir, syncDir: syncDirectory, lock: lock}
+ if err := store.cleanupTemps(); err != nil {
+ _ = store.Close()
+ return nil, err
+ }
+ if _, err := store.readAll(); err != nil {
+ _ = store.Close()
+ return nil, err
+ }
+ return store, nil
+}
+
+// makeDurablePrivateDir creates every missing path component with private
+// permissions and fsyncs its parent immediately after publication. It also
+// re-confirms the deepest visible component so a retry cannot mistake a prior
+// sync-failed mkdir for a crash-durable directory entry.
+func makeDurablePrivateDir(dir string, syncDirectory func(string) error) error {
+ dir = filepath.Clean(dir)
+ missing := []string{}
+ current := dir
+ for {
+ info, err := os.Lstat(current)
+ if err == nil {
+ if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
+ return fmt.Errorf("path component must be a real directory: %s", current)
+ }
+ break
+ }
+ if !errors.Is(err, os.ErrNotExist) {
+ return fmt.Errorf("inspect directory %s: %w", current, err)
+ }
+ missing = append(missing, current)
+ parent := filepath.Dir(current)
+ if parent == current {
+ return fmt.Errorf("no existing parent for task result outbox: %s", dir)
+ }
+ current = parent
+ }
+
+ // Confirm the deepest existing component before extending it. A prior Open
+ // may have created exactly this component and then failed its parent fsync;
+ // creation proceeds only one confirmed component at a time, so this retry
+ // closes that ambiguity without trusting mere directory visibility.
+ if parent := filepath.Dir(current); parent != current {
+ if err := syncDirectory(parent); err != nil {
+ return fmt.Errorf("confirm existing directory %s: %w", current, err)
+ }
+ }
+ for i := len(missing) - 1; i >= 0; i-- {
+ path := missing[i]
+ if err := os.Mkdir(path, 0o700); err != nil {
+ return fmt.Errorf("create private directory %s: %w", path, err)
+ }
+ if err := syncDirectory(filepath.Dir(path)); err != nil {
+ return fmt.Errorf("sync parent after creating %s: %w", path, err)
+ }
+ }
+ return nil
+}
+
+// Close releases this process's exclusive ownership of the outbox.
+func (s *Store) Close() error {
+ if s == nil || s.lock == nil {
+ return nil
+ }
+ err := unlockOutbox(s.lock)
+ s.lock = nil
+ return err
+}
+
+// Begin durably records a task lease before any task code is executed. A false,
+// nil result means the exact executable task and lease are already journaled and
+// must not be executed again. A true result means this call published the new
+// lease journal, even if a subsequent directory sync reported an error.
+func (s *Store) Begin(task model.Task) (committed bool, err error) {
+ if strings.TrimSpace(task.ID) == "" || strings.TrimSpace(task.LeaseID) == "" {
+ return false, fmt.Errorf("task id and lease id are required for durable execution")
+ }
+ entries, err := s.readAll()
+ if err != nil {
+ return false, err
+ }
+ key := entryKey(task.ID, task.LeaseID)
+ for _, entry := range entries {
+ if entry.Task.ID == task.ID {
+ if entry.key == key && reflect.DeepEqual(entry.Task, task) {
+ return false, nil
+ }
+ return false, fmt.Errorf("task %s was redelivered with a different lease or content", task.ID)
+ }
+ }
+ if len(entries) >= maxEntries {
+ return false, ErrCapacity
+ }
+ now := time.Now().UTC()
+ return s.writeNew(key, Entry{
+ Version: entryVersion,
+ State: stateLeased,
+ Task: task,
+ ExecutionStartedAt: now,
+ UpdatedAt: now,
+ })
+}
+
+// Complete replaces the lease journal with the full result before upload. The
+// committed return value has the same publication semantics as Begin.
+func (s *Store) Complete(result model.TaskResult) (committed bool, err error) {
+ key := entryKey(result.TaskID, result.LeaseID)
+ entry, err := s.read(key)
+ if err != nil {
+ return false, err
+ }
+ if entry.State != stateLeased {
+ return false, fmt.Errorf("task lease %s is not awaiting completion", result.TaskID)
+ }
+ if entry.Task.ID != result.TaskID || entry.Task.LeaseID != result.LeaseID {
+ return false, fmt.Errorf("task result does not match durable lease")
+ }
+ entry.State = stateDone
+ entry.Result = &result
+ entry.UpdatedAt = time.Now().UTC()
+ return s.write(key, entry)
+}
+
+// ConfirmDurability retries the directory sync that makes a published journal
+// transition crash-durable. A completed result must not be exposed to the
+// server after Complete reports committed=true with an error until this call
+// succeeds.
+func (s *Store) ConfirmDurability() error {
+ if err := s.syncDir(s.dir); err != nil {
+ return fmt.Errorf("confirm task result outbox durability: %w", err)
+ }
+ s.durabilityUnconfirmed = false
+ return nil
+}
+
+// RecoverInterrupted turns every pre-execution/unknown-outcome lease journal
+// into an honest synthetic result. It never re-runs the task.
+func (s *Store) RecoverInterrupted(nodeID string) error {
+ // runTasks always calls recovery before Pending. Re-confirm the journal
+ // directory on every cycle so a completed entry that was published by a
+ // prior recovery attempt but whose directory fsync failed can never be
+ // uploaded merely because it is visible after the failed call.
+ if err := s.ConfirmDurability(); err != nil {
+ return fmt.Errorf("confirm task result outbox before recovery: %w", err)
+ }
+ entries, err := s.readAll()
+ if err != nil {
+ return err
+ }
+ for _, entry := range entries {
+ if entry.State != stateLeased {
+ continue
+ }
+ now := time.Now().UTC()
+ result := model.TaskResult{
+ TaskID: entry.Task.ID,
+ LeaseID: entry.Task.LeaseID,
+ NodeID: nodeID,
+ ExitCode: -1,
+ Error: "agent restarted or lost durable result after task launch; execution outcome is unknown and the task was not re-executed",
+ StartedAt: entry.ExecutionStartedAt,
+ FinishedAt: now,
+ }
+ entry.State = stateDone
+ entry.Result = &result
+ entry.UpdatedAt = now
+ committed, writeErr := s.write(entry.key, entry)
+ if writeErr == nil {
+ continue
+ }
+ if !committed {
+ return fmt.Errorf("recover interrupted task %s: %w", entry.Task.ID, writeErr)
+ }
+ // The unknown-outcome result is visible after rename, but it must not
+ // escape through Pending until its directory entry is crash-durable.
+ // Try immediately; if this also fails, the next recovery cycle's leading
+ // confirmation blocks upload and preserves the exact published result.
+ if confirmErr := s.ConfirmDurability(); confirmErr != nil {
+ return fmt.Errorf("recover interrupted task %s: %v; confirm published recovery: %w", entry.Task.ID, writeErr, confirmErr)
+ }
+ }
+ return nil
+}
+
+// Pending returns completed results in deterministic oldest-first order.
+func (s *Store) Pending() ([]Entry, error) {
+ if s.durabilityUnconfirmed {
+ return nil, fmt.Errorf("task result outbox durability is unconfirmed")
+ }
+ entries, err := s.readAll()
+ if err != nil {
+ return nil, err
+ }
+ pending := entries[:0]
+ for _, entry := range entries {
+ if entry.State == stateDone && entry.Result != nil {
+ pending = append(pending, entry)
+ }
+ }
+ sort.Slice(pending, func(i, j int) bool {
+ if pending[i].UpdatedAt.Equal(pending[j].UpdatedAt) {
+ return pending[i].key < pending[j].key
+ }
+ return pending[i].UpdatedAt.Before(pending[j].UpdatedAt)
+ })
+ return pending, nil
+}
+
+// Remove atomically unlinks an acknowledged outbox entry and syncs the
+// directory so the acknowledgement survives a crash.
+func (s *Store) Remove(entry Entry) error {
+ if entry.key == "" {
+ entry.key = entryKey(entry.Task.ID, entry.Task.LeaseID)
+ }
+ if err := os.Remove(s.path(entry.key)); err != nil {
+ return fmt.Errorf("remove acknowledged task result: %w", err)
+ }
+ if err := s.syncDir(s.dir); err != nil {
+ s.durabilityUnconfirmed = true
+ return err
+ }
+ return nil
+}
+
+func (s *Store) readAll() ([]Entry, error) {
+ dirEntries, err := os.ReadDir(s.dir)
+ if err != nil {
+ return nil, fmt.Errorf("read task result outbox: %w", err)
+ }
+ entries := make([]Entry, 0, len(dirEntries))
+ for _, dirEntry := range dirEntries {
+ if dirEntry.IsDir() || !strings.HasSuffix(dirEntry.Name(), ".json") {
+ continue
+ }
+ if len(entries) >= maxEntries {
+ return nil, ErrCapacity
+ }
+ key := strings.TrimSuffix(dirEntry.Name(), ".json")
+ entry, err := s.read(key)
+ if err != nil {
+ return nil, err
+ }
+ entries = append(entries, entry)
+ }
+ return entries, nil
+}
+
+func (s *Store) read(key string) (Entry, error) {
+ path := s.path(key)
+ info, err := os.Lstat(path)
+ if err != nil {
+ return Entry{}, fmt.Errorf("inspect task result journal: %w", err)
+ }
+ if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 {
+ return Entry{}, fmt.Errorf("task result journal is not a regular file: %s", path)
+ }
+ if info.Mode().Perm()&0o077 != 0 {
+ return Entry{}, fmt.Errorf("task result journal permissions are not private: %s", path)
+ }
+ if info.Size() > maxEntrySize {
+ return Entry{}, fmt.Errorf("task result journal exceeds %d bytes: %s", maxEntrySize, path)
+ }
+ f, err := os.Open(path)
+ if err != nil {
+ return Entry{}, fmt.Errorf("open task result journal: %w", err)
+ }
+ defer f.Close()
+ var entry Entry
+ dec := json.NewDecoder(io.LimitReader(f, maxEntrySize+1))
+ if err := dec.Decode(&entry); err != nil {
+ return Entry{}, fmt.Errorf("decode task result journal: %w", err)
+ }
+ var trailing any
+ if err := dec.Decode(&trailing); err == nil {
+ return Entry{}, fmt.Errorf("decode task result journal: unexpected trailing data")
+ } else if !errors.Is(err, io.EOF) {
+ return Entry{}, fmt.Errorf("decode task result journal trailing data: %w", err)
+ }
+ if entry.Version != entryVersion || entry.Task.ID == "" || entry.Task.LeaseID == "" {
+ return Entry{}, fmt.Errorf("invalid task result journal: %s", path)
+ }
+ if key != entryKey(entry.Task.ID, entry.Task.LeaseID) {
+ return Entry{}, fmt.Errorf("task result journal key mismatch: %s", path)
+ }
+ if entry.State != stateLeased && entry.State != stateDone {
+ return Entry{}, fmt.Errorf("invalid task result journal state %q", entry.State)
+ }
+ if entry.State == stateLeased && entry.Result != nil {
+ return Entry{}, fmt.Errorf("leased task result journal unexpectedly contains a result")
+ }
+ if entry.State == stateDone {
+ if entry.Result == nil || entry.Result.TaskID != entry.Task.ID || entry.Result.LeaseID != entry.Task.LeaseID {
+ return Entry{}, fmt.Errorf("completed task result journal does not match its lease")
+ }
+ }
+ entry.key = key
+ return entry, nil
+}
+
+func (s *Store) cleanupTemps() error {
+ entries, err := os.ReadDir(s.dir)
+ if err != nil {
+ return fmt.Errorf("read task result outbox: %w", err)
+ }
+ removed := false
+ for _, entry := range entries {
+ if !strings.HasPrefix(entry.Name(), ".task-result-") || !strings.HasSuffix(entry.Name(), ".tmp") {
+ continue
+ }
+ path := filepath.Join(s.dir, entry.Name())
+ info, err := os.Lstat(path)
+ if err != nil {
+ return fmt.Errorf("inspect stale task result journal: %w", err)
+ }
+ if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 {
+ return fmt.Errorf("stale task result journal is not a regular file: %s", path)
+ }
+ if err := os.Remove(path); err != nil {
+ return fmt.Errorf("remove stale task result journal: %w", err)
+ }
+ removed = true
+ }
+ if removed {
+ return s.syncDir(s.dir)
+ }
+ return nil
+}
+
+func (s *Store) write(key string, entry Entry) (bool, error) {
+ return s.writeEntry(key, entry, false)
+}
+
+func (s *Store) writeNew(key string, entry Entry) (bool, error) {
+ return s.writeEntry(key, entry, true)
+}
+
+func (s *Store) writeEntry(key string, entry Entry, createOnly bool) (bool, error) {
+ data, err := json.Marshal(entry)
+ if err != nil {
+ return false, fmt.Errorf("encode task result journal: %w", err)
+ }
+ if len(data) > maxEntrySize {
+ return false, fmt.Errorf("task result journal exceeds %d bytes", maxEntrySize)
+ }
+ tmp, err := os.CreateTemp(s.dir, ".task-result-*.tmp")
+ if err != nil {
+ return false, fmt.Errorf("create task result journal: %w", err)
+ }
+ tmpPath := tmp.Name()
+ cleanup := func() {
+ tmp.Close()
+ _ = os.Remove(tmpPath)
+ }
+ if err := tmp.Chmod(0o600); err != nil {
+ cleanup()
+ return false, fmt.Errorf("secure task result journal: %w", err)
+ }
+ if _, err := tmp.Write(data); err != nil {
+ cleanup()
+ return false, fmt.Errorf("write task result journal: %w", err)
+ }
+ if err := tmp.Sync(); err != nil {
+ cleanup()
+ return false, fmt.Errorf("sync task result journal: %w", err)
+ }
+ if err := tmp.Close(); err != nil {
+ _ = os.Remove(tmpPath)
+ return false, fmt.Errorf("close task result journal: %w", err)
+ }
+ finalPath := s.path(key)
+ if createOnly {
+ // Linking a fully written same-directory temp file publishes the initial
+ // lease without overwriting another agent process's journal.
+ if err := os.Link(tmpPath, finalPath); err != nil {
+ _ = os.Remove(tmpPath)
+ return false, fmt.Errorf("publish new task result journal: %w", err)
+ }
+ if err := os.Remove(tmpPath); err != nil {
+ s.durabilityUnconfirmed = true
+ return true, fmt.Errorf("remove published task result temp file: %w", err)
+ }
+ } else if err := os.Rename(tmpPath, finalPath); err != nil {
+ _ = os.Remove(tmpPath)
+ return false, fmt.Errorf("publish task result journal: %w", err)
+ }
+ if err := s.syncDir(s.dir); err != nil {
+ s.durabilityUnconfirmed = true
+ return true, fmt.Errorf("sync task result outbox: %w", err)
+ }
+ s.durabilityUnconfirmed = false
+ return true, nil
+}
+
+func (s *Store) path(key string) string {
+ return filepath.Join(s.dir, key+".json")
+}
+
+func entryKey(taskID, leaseID string) string {
+ sum := sha256.Sum256([]byte(taskID + "\x00" + leaseID))
+ return hex.EncodeToString(sum[:])
+}
+
+func syncDir(dir string) error {
+ f, err := os.Open(dir)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ return f.Sync()
+}
diff --git a/internal/taskoutbox/outbox_test.go b/internal/taskoutbox/outbox_test.go
new file mode 100644
index 0000000..8607d7e
--- /dev/null
+++ b/internal/taskoutbox/outbox_test.go
@@ -0,0 +1,403 @@
+package taskoutbox
+
+import (
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "reflect"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/LatticeNet/lattice-sdk/model"
+)
+
+func TestCompletedResultSurvivesReopenUntilAcknowledged(t *testing.T) {
+ dir := t.TempDir()
+ store, err := Open(dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ task := testTask()
+ if _, err := store.Begin(task); err != nil {
+ t.Fatal(err)
+ }
+ result := model.TaskResult{
+ TaskID: task.ID, LeaseID: task.LeaseID, NodeID: "node-a", ExitCode: 0,
+ Stdout: "done", StartedAt: time.Now().UTC(), FinishedAt: time.Now().UTC(),
+ }
+ if _, err := store.Complete(result); err != nil {
+ t.Fatal(err)
+ }
+
+ if err := store.Close(); err != nil {
+ t.Fatal(err)
+ }
+ reopened, err := Open(dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer reopened.Close()
+ pending, err := reopened.Pending()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(pending) != 1 || pending[0].Result == nil || pending[0].Result.Stdout != "done" {
+ t.Fatalf("unexpected durable result: %+v", pending)
+ }
+ files, err := os.ReadDir(dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var journals []os.DirEntry
+ for _, file := range files {
+ if strings.HasSuffix(file.Name(), ".json") {
+ journals = append(journals, file)
+ }
+ }
+ if len(journals) != 1 {
+ t.Fatalf("journal files = %d, want 1 (all files: %d)", len(journals), len(files))
+ }
+ info, err := journals[0].Info()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if info.Mode().Perm() != 0o600 {
+ t.Fatalf("journal mode = %o, want 600", info.Mode().Perm())
+ }
+ if err := reopened.Remove(pending[0]); err != nil {
+ t.Fatal(err)
+ }
+ if pending, err := reopened.Pending(); err != nil || len(pending) != 0 {
+ t.Fatalf("pending after acknowledgement = %+v, err=%v", pending, err)
+ }
+}
+
+func TestOpenDurablyCreatesEveryMissingDirectoryComponent(t *testing.T) {
+ base := t.TempDir()
+ target := filepath.Join(base, "task-outbox", "node-hash")
+ var synced []string
+ store, err := openWithSync(target, func(dir string) error {
+ synced = append(synced, dir)
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ want := []string{filepath.Dir(base), base, filepath.Join(base, "task-outbox")}
+ if !reflect.DeepEqual(synced, want) {
+ t.Fatalf("directory sync order = %q, want %q", synced, want)
+ }
+ for _, dir := range []string{filepath.Join(base, "task-outbox"), target} {
+ info, err := os.Stat(dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if info.Mode().Perm() != 0o700 {
+ t.Fatalf("directory %s mode = %o, want 700", dir, info.Mode().Perm())
+ }
+ }
+}
+
+func TestOpenReconfirmsVisibleDirectoryAfterPriorSyncFailure(t *testing.T) {
+ base := t.TempDir()
+ taskRoot := filepath.Join(base, "task-outbox")
+ target := filepath.Join(taskRoot, "node-hash")
+ calls := 0
+ _, err := openWithSync(target, func(string) error {
+ calls++
+ if calls == 3 {
+ return errors.New("injected directory sync failure")
+ }
+ return nil
+ })
+ if err == nil || !strings.Contains(err.Error(), "directory sync failure") {
+ t.Fatalf("Open() error = %v, want parent sync failure", err)
+ }
+ if _, statErr := os.Lstat(target); statErr != nil {
+ t.Fatalf("published but unconfirmed node directory is not visible: %v", statErr)
+ }
+ var retrySynced []string
+ store, err := openWithSync(target, func(dir string) error {
+ retrySynced = append(retrySynced, dir)
+ return nil
+ })
+ if err != nil {
+ t.Fatalf("retry Open() did not confirm visible directory: %v", err)
+ }
+ defer store.Close()
+ if len(retrySynced) == 0 || retrySynced[0] != taskRoot {
+ t.Fatalf("retry sync order = %q, want first confirmation of %q", retrySynced, taskRoot)
+ }
+}
+
+func TestRecoverInterruptedCreatesUnknownOutcomeWithoutTaskOutput(t *testing.T) {
+ store, err := Open(t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ if _, err := store.Begin(testTask()); err != nil {
+ t.Fatal(err)
+ }
+ if err := store.RecoverInterrupted("node-a"); err != nil {
+ t.Fatal(err)
+ }
+ pending, err := store.Pending()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(pending) != 1 || pending[0].Result == nil {
+ t.Fatalf("unexpected recovered entries: %+v", pending)
+ }
+ result := pending[0].Result
+ if result.NodeID != "node-a" || result.ExitCode != -1 || result.Stdout != "" || result.Stderr != "" {
+ t.Fatalf("unexpected recovered result: %+v", result)
+ }
+ if !strings.Contains(result.Error, "outcome is unknown") || !strings.Contains(result.Error, "not re-executed") {
+ t.Fatalf("recovery error is not honest enough: %q", result.Error)
+ }
+}
+
+func TestRecoverInterruptedReconfirmsPublishedResultBeforeLaterUpload(t *testing.T) {
+ store, err := Open(t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ if _, err := store.Begin(testTask()); err != nil {
+ t.Fatal(err)
+ }
+
+ syncCalls := 0
+ store.syncDir = func(string) error {
+ syncCalls++
+ // The cycle-start confirmation succeeds, then both the recovery write's
+ // directory sync and its immediate confirmation fail.
+ if syncCalls >= 2 {
+ return errors.New("injected recovery directory sync failure")
+ }
+ return nil
+ }
+ if err := store.RecoverInterrupted("node-a"); err == nil || !strings.Contains(err.Error(), "confirm published recovery") {
+ t.Fatalf("first RecoverInterrupted() error = %v, want unconfirmed published recovery", err)
+ }
+ if pending, err := store.Pending(); err == nil || pending != nil || !strings.Contains(err.Error(), "durability is unconfirmed") {
+ t.Fatalf("Pending() after unconfirmed recovery = %+v, %v; want upload blocked", pending, err)
+ }
+ entries, err := store.readAll()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(entries) != 1 || entries[0].State != stateDone || entries[0].Result == nil {
+ t.Fatalf("published recovery was not retained: %+v", entries)
+ }
+ finishedAt := entries[0].Result.FinishedAt
+
+ // A second runTasks cycle enters RecoverInterrupted before Pending. If
+ // confirmation still fails, the cycle must stop and cannot upload.
+ if err := store.RecoverInterrupted("node-a"); err == nil || !strings.Contains(err.Error(), "before recovery") {
+ t.Fatalf("second RecoverInterrupted() error = %v, want pre-upload confirmation failure", err)
+ }
+
+ store.syncDir = syncDir
+ if err := store.RecoverInterrupted("node-a"); err != nil {
+ t.Fatalf("RecoverInterrupted() after durability recovery: %v", err)
+ }
+ pending, err := store.Pending()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(pending) != 1 || pending[0].Result == nil || !pending[0].Result.FinishedAt.Equal(finishedAt) {
+ t.Fatalf("recovery result changed across durability retry: %+v", pending)
+ }
+}
+
+func TestBeginReportsJournalPublishedWhenDirectorySyncFails(t *testing.T) {
+ store, err := Open(t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ store.syncDir = func(string) error { return errors.New("injected directory sync failure") }
+ committed, err := store.Begin(testTask())
+ if !committed || err == nil || !strings.Contains(err.Error(), "directory sync failure") {
+ t.Fatalf("Begin() = committed %v, err %v; want published journal plus sync error", committed, err)
+ }
+ store.syncDir = syncDir
+ if err := store.RecoverInterrupted("node-a"); err != nil {
+ t.Fatal(err)
+ }
+ pending, err := store.Pending()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(pending) != 1 || pending[0].Result == nil || pending[0].Result.ExitCode != -1 {
+ t.Fatalf("published lease was not recoverable: %+v", pending)
+ }
+}
+
+func TestBeginTreatsExactLeaseRedeliveryAsNoOpAndRejectsChangedScript(t *testing.T) {
+ store, err := Open(t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ task := testTask()
+ if created, err := store.Begin(task); err != nil || !created {
+ t.Fatalf("first Begin() = created %v, err %v", created, err)
+ }
+ if created, err := store.Begin(task); err != nil || created {
+ t.Fatalf("exact redelivery Begin() = created %v, err %v; want no-op", created, err)
+ }
+ changed := task
+ changed.Script = "echo changed"
+ if created, err := store.Begin(changed); err == nil || created || !strings.Contains(err.Error(), "different lease or content") {
+ t.Fatalf("changed redelivery Begin() = created %v, err %v", created, err)
+ }
+ changedLease := task
+ changedLease.LeaseID = "lease-b"
+ if created, err := store.Begin(changedLease); err == nil || created || !strings.Contains(err.Error(), "different lease or content") {
+ t.Fatalf("new-lease redelivery Begin() = created %v, err %v", created, err)
+ }
+ changedControlPlaneField := task
+ changedControlPlaneField.ActorID = "actor-changed"
+ changedControlPlaneField.Targets = []string{"node-other"}
+ if created, err := store.Begin(changedControlPlaneField); err == nil || created || !strings.Contains(err.Error(), "different lease or content") {
+ t.Fatalf("full-task redelivery Begin() = created %v, err %v", created, err)
+ }
+ entry, err := store.read(entryKey(task.ID, task.LeaseID))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if entry.Task.Script != task.Script || entry.State != stateLeased {
+ t.Fatalf("redelivery changed durable journal: %+v", entry)
+ }
+}
+
+func TestBeginExactCompletedRedeliveryPreservesResult(t *testing.T) {
+ store, err := Open(t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ task := testTask()
+ if _, err := store.Begin(task); err != nil {
+ t.Fatal(err)
+ }
+ result := model.TaskResult{TaskID: task.ID, LeaseID: task.LeaseID, NodeID: "node-a", ExitCode: 0, Stdout: "exact"}
+ if _, err := store.Complete(result); err != nil {
+ t.Fatal(err)
+ }
+ if created, err := store.Begin(task); err != nil || created {
+ t.Fatalf("completed redelivery Begin() = created %v, err %v", created, err)
+ }
+ pending, err := store.Pending()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(pending) != 1 || pending[0].Result == nil || !reflect.DeepEqual(*pending[0].Result, result) {
+ t.Fatalf("completed redelivery changed exact result: %+v", pending)
+ }
+}
+
+func TestBeginExactRedeliveryBypassesCapacity(t *testing.T) {
+ store, err := Open(t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ task := testTask()
+ if _, err := store.Begin(task); err != nil {
+ t.Fatal(err)
+ }
+ for i := 1; i < maxEntries; i++ {
+ other := model.Task{ID: fmt.Sprintf("task-%04d", i), LeaseID: fmt.Sprintf("lease-%04d", i)}
+ entry := Entry{Version: entryVersion, State: stateLeased, Task: other, UpdatedAt: time.Now().UTC()}
+ raw, err := json.Marshal(entry)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(store.path(entryKey(other.ID, other.LeaseID)), raw, 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if created, err := store.Begin(task); err != nil || created {
+ t.Fatalf("exact full-capacity redelivery = created %v, err %v", created, err)
+ }
+ if created, err := store.Begin(model.Task{ID: "task-overflow", LeaseID: "lease-overflow"}); !errors.Is(err, ErrCapacity) || created {
+ t.Fatalf("new full-capacity task = created %v, err %v; want ErrCapacity", created, err)
+ }
+}
+
+func TestCompleteReportsResultPublishedWhenDirectorySyncFails(t *testing.T) {
+ store, err := Open(t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close()
+ task := testTask()
+ if _, err := store.Begin(task); err != nil {
+ t.Fatal(err)
+ }
+ store.syncDir = func(string) error { return errors.New("injected directory sync failure") }
+ result := model.TaskResult{TaskID: task.ID, LeaseID: task.LeaseID, NodeID: "node-a", ExitCode: 0, Stdout: "exact"}
+ committed, err := store.Complete(result)
+ if !committed || err == nil || !strings.Contains(err.Error(), "directory sync failure") {
+ t.Fatalf("Complete() = committed %v, err %v; want published result plus sync error", committed, err)
+ }
+ if err := store.ConfirmDurability(); err == nil || !strings.Contains(err.Error(), "directory sync failure") {
+ t.Fatalf("ConfirmDurability() error = %v, want persistent sync failure", err)
+ }
+ store.syncDir = syncDir
+ if err := store.ConfirmDurability(); err != nil {
+ t.Fatalf("ConfirmDurability() after recovery: %v", err)
+ }
+ pending, err := store.Pending()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(pending) != 1 || pending[0].Result == nil || pending[0].Result.Stdout != "exact" {
+ t.Fatalf("published result was not retained exactly: %+v", pending)
+ }
+}
+
+func TestOpenRefusesConcurrentOwnerWithoutChangingJournal(t *testing.T) {
+ dir := t.TempDir()
+ first, err := Open(dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := first.Begin(testTask()); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := Open(dir); err == nil || !strings.Contains(err.Error(), "already owned") {
+ t.Fatalf("concurrent Open() error = %v, want exclusive-owner rejection", err)
+ }
+ entry, err := first.read(entryKey(testTask().ID, testTask().LeaseID))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if entry.State != stateLeased || entry.Result != nil {
+ t.Fatalf("concurrent open modified active journal: %+v", entry)
+ }
+ if err := first.Close(); err != nil {
+ t.Fatal(err)
+ }
+ second, err := Open(dir)
+ if err != nil {
+ t.Fatalf("Open() after release: %v", err)
+ }
+ defer second.Close()
+}
+
+func testTask() model.Task {
+ return model.Task{
+ ID: "task-a", LeaseID: "lease-a", Interpreter: "sh", Script: "echo done",
+ TimeoutSec: 10, OutputLimit: 1024,
+ }
+}
diff --git a/scripts/install.sh b/scripts/install.sh
index f37f415..d2400ec 100755
--- a/scripts/install.sh
+++ b/scripts/install.sh
@@ -82,6 +82,7 @@ if [ "$(id -u)" -ne 0 ]; then
LATTICE_TASK_CGROUP_PIDS_MAX="${LATTICE_TASK_CGROUP_PIDS_MAX:-}" \
LATTICE_TASK_CGROUP_CPU_MAX="${LATTICE_TASK_CGROUP_CPU_MAX:-}" \
LATTICE_TASK_WORK_ROOT="${LATTICE_TASK_WORK_ROOT:-}" \
+ LATTICE_TASK_OUTBOX_DIR="${LATTICE_TASK_OUTBOX_DIR:-}" \
LATTICE_AGENT_ALLOW_TERMINAL="${LATTICE_AGENT_ALLOW_TERMINAL:-}" \
LATTICE_TERMINAL_TRANSPORT="${LATTICE_TERMINAL_TRANSPORT:-}" \
LATTICE_IP_MODE="${LATTICE_IP_MODE:-}" LATTICE_IP_RESOLVERS="${LATTICE_IP_RESOLVERS:-}" \
@@ -195,7 +196,7 @@ load_existing_config() {
LATTICE_AGENT_ALLOW_EXEC LATTICE_AGENT_ALLOW_ROOT_EXEC LATTICE_NO_EXEC \
LATTICE_TASK_CGROUP_ROOT LATTICE_TASK_CGROUP_MEMORY_MAX \
LATTICE_TASK_CGROUP_PIDS_MAX LATTICE_TASK_CGROUP_CPU_MAX \
- LATTICE_TASK_WORK_ROOT \
+ LATTICE_TASK_WORK_ROOT LATTICE_TASK_OUTBOX_DIR \
LATTICE_AGENT_ALLOW_TERMINAL LATTICE_TERMINAL_TRANSPORT LATTICE_IP_MODE \
LATTICE_IP_RESOLVERS LATTICE_IP_SCRIPT LATTICE_PUBLIC_IP LATTICE_PUBLIC_IP6 \
LATTICE_SSH_ALERTS LATTICE_SINGBOX_DISCOVER LATTICE_SINGBOX_BIN \
@@ -298,6 +299,10 @@ apply_service_identity_permissions() {
chown "$run_user:$run_group" "$LATTICE_TASK_WORK_ROOT" || die "cannot assign $LATTICE_TASK_WORK_ROOT to $run_user:$run_group"
chmod 0700 "$LATTICE_TASK_WORK_ROOT" 2>/dev/null || true
fi
+ if [ -n "${task_outbox_leaf:-}" ]; then
+ chown "$run_user:$run_group" "$task_outbox_leaf" || die "cannot assign $task_outbox_leaf to $run_user:$run_group"
+ chmod 0700 "$task_outbox_leaf" 2>/dev/null || true
+ fi
}
prepare_task_work_root() {
@@ -310,6 +315,46 @@ prepare_task_work_root() {
chmod 0700 "$LATTICE_TASK_WORK_ROOT" 2>/dev/null || true
}
+prepare_task_outbox_root() {
+ task_outbox_leaf=""
+ [ -n "${LATTICE_TASK_OUTBOX_DIR:-}" ] || return 0
+ case "$LATTICE_TASK_OUTBOX_DIR" in
+ /|/bin|/boot|/dev|/etc|/lib|/lib32|/lib64|/proc|/root|/sbin|/sys|/usr|/var)
+ die "LATTICE_TASK_OUTBOX_DIR must not be a filesystem or system root: $LATTICE_TASK_OUTBOX_DIR" ;;
+ /*) ;;
+ *) die "LATTICE_TASK_OUTBOX_DIR must be an absolute path" ;;
+ esac
+ [ ! -L "$LATTICE_TASK_OUTBOX_DIR" ] || die "LATTICE_TASK_OUTBOX_DIR must not traverse symlinks: $LATTICE_TASK_OUTBOX_DIR"
+ if [ -e "$LATTICE_TASK_OUTBOX_DIR" ] && [ ! -d "$LATTICE_TASK_OUTBOX_DIR" ]; then
+ die "LATTICE_TASK_OUTBOX_DIR must be a directory: $LATTICE_TASK_OUTBOX_DIR"
+ fi
+ mkdir -p "$LATTICE_TASK_OUTBOX_DIR" || die "cannot create LATTICE_TASK_OUTBOX_DIR=$LATTICE_TASK_OUTBOX_DIR"
+ current="$LATTICE_TASK_OUTBOX_DIR"
+ while [ "$current" != "/" ]; do
+ if [ -L "$current" ]; then
+ root_owned_link="$(find "$current" -prune -user root -print 2>/dev/null || true)"
+ [ "$root_owned_link" = "$current" ] || die "LATTICE_TASK_OUTBOX_DIR must not traverse a non-root-owned symlink: $current"
+ fi
+ current="${current%/*}"
+ [ -n "$current" ] || current="/"
+ done
+ case "$LATTICE_TASK_OUTBOX_DIR/" in
+ "$LATTICE_HOME/"*|"$state_dir/"*) ;;
+ *)
+ secure_base="$(find "$LATTICE_TASK_OUTBOX_DIR" -prune -user root ! -perm -022 -print 2>/dev/null || true)"
+ [ "$secure_base" = "$LATTICE_TASK_OUTBOX_DIR" ] || \
+ die "external LATTICE_TASK_OUTBOX_DIR must be root-owned and not group/world writable: $LATTICE_TASK_OUTBOX_DIR"
+ ;;
+ esac
+ task_outbox_leaf="$LATTICE_TASK_OUTBOX_DIR/task-outbox"
+ if [ -e "$task_outbox_leaf" ] || [ -L "$task_outbox_leaf" ]; then
+ [ ! -L "$task_outbox_leaf" ] && [ -d "$task_outbox_leaf" ] || die "task outbox leaf must be a real directory: $task_outbox_leaf"
+ else
+ mkdir "$task_outbox_leaf" || die "cannot create task outbox leaf=$task_outbox_leaf"
+ fi
+ chmod 0700 "$task_outbox_leaf" || die "cannot secure task outbox leaf=$task_outbox_leaf"
+}
+
# ---- service helpers (systemd / openrc / launchd) --------------------------
svc_kind() {
if [ "$os" = "darwin" ]; then echo launchd; return; fi
@@ -375,6 +420,7 @@ fi
[ -w "$LATTICE_HOME" ] || die "$LATTICE_HOME is not writable"
mkdir -p "$state_dir"
prepare_task_work_root
+prepare_task_outbox_root
chmod 0750 "$LATTICE_HOME" 2>/dev/null || true
apply_service_identity_permissions
@@ -428,6 +474,7 @@ LATTICE_TASK_CGROUP_MEMORY_MAX=$(quote_env "${LATTICE_TASK_CGROUP_MEMORY_MAX:-53
LATTICE_TASK_CGROUP_PIDS_MAX=$(quote_env "${LATTICE_TASK_CGROUP_PIDS_MAX:-64}")
LATTICE_TASK_CGROUP_CPU_MAX=$(quote_env "${LATTICE_TASK_CGROUP_CPU_MAX:-100000 100000}")
LATTICE_TASK_WORK_ROOT=$(quote_env "${LATTICE_TASK_WORK_ROOT:-}")
+LATTICE_TASK_OUTBOX_DIR=$(quote_env "${LATTICE_TASK_OUTBOX_DIR:-}")
LATTICE_AGENT_ALLOW_TERMINAL=$(quote_env "${LATTICE_AGENT_ALLOW_TERMINAL:-0}")
LATTICE_TERMINAL_TRANSPORT=$(quote_env "${LATTICE_TERMINAL_TRANSPORT:-poll}")
LATTICE_IP_MODE=$(quote_env "${LATTICE_IP_MODE:-auto}")
@@ -519,6 +566,7 @@ EOF
LATTICE_TASK_CGROUP_PIDS_MAX$(xml_escape "${LATTICE_TASK_CGROUP_PIDS_MAX:-64}")
LATTICE_TASK_CGROUP_CPU_MAX$(xml_escape "${LATTICE_TASK_CGROUP_CPU_MAX:-100000 100000}")
LATTICE_TASK_WORK_ROOT$(xml_escape "${LATTICE_TASK_WORK_ROOT:-}")
+ LATTICE_TASK_OUTBOX_DIR$(xml_escape "${LATTICE_TASK_OUTBOX_DIR:-}")
LATTICE_AGENT_ALLOW_TERMINAL$(xml_escape "${LATTICE_AGENT_ALLOW_TERMINAL:-0}")
LATTICE_TERMINAL_TRANSPORT$(xml_escape "${LATTICE_TERMINAL_TRANSPORT:-poll}")
LATTICE_IP_MODE$(xml_escape "${LATTICE_IP_MODE:-auto}")
diff --git a/scripts/test-install-integrity.sh b/scripts/test-install-integrity.sh
index 7a181c2..0ce1714 100755
--- a/scripts/test-install-integrity.sh
+++ b/scripts/test-install-integrity.sh
@@ -118,7 +118,10 @@ for expected in \
'User=$run_user' \
'Group=$run_group' \
'Delegate=yes' \
- 'chown "$run_user:$run_group" "$state_dir"'
+ 'chown "$run_user:$run_group" "$state_dir"' \
+ 'LATTICE_TASK_OUTBOX_DIR="${LATTICE_TASK_OUTBOX_DIR:-}"' \
+ 'LATTICE_TASK_OUTBOX_DIR=$(quote_env "${LATTICE_TASK_OUTBOX_DIR:-}")' \
+ 'chown "$run_user:$run_group" "$task_outbox_leaf"'
do
if ! grep -Fq "$expected" "$ROOT/scripts/install.sh"; then
echo "installer non-root systemd contract missing: $expected" >&2
@@ -126,6 +129,67 @@ do
fi
done
+UNSAFE_LOG="$TMP/unsafe-outbox.log"
+if PATH="$FAKEBIN:/usr/bin:/bin" \
+ LATTICE_HOME="$HOME_DIR" \
+ LATTICE_AGENT_BIN="$HOME_DIR/lattice-agent" \
+ LATTICE_AGENT_ENV="$HOME_DIR/lattice-agent.env" \
+ LATTICE_AGENT_STATE="$HOME_DIR/state" \
+ LATTICE_TASK_OUTBOX_DIR="/" \
+ LATTICE_SERVER="https://lattice.example.com" \
+ LATTICE_NODE_ID="node-test" \
+ LATTICE_NODE_TOKEN="node-token-test" \
+ sh "$ROOT/scripts/install.sh" >"$UNSAFE_LOG" 2>&1; then
+ echo "installer accepted filesystem root as task outbox" >&2
+ exit 1
+fi
+if ! grep -Fq "must not be a filesystem or system root" "$UNSAFE_LOG"; then
+ echo "unsafe outbox rejection was not explicit" >&2
+ cat "$UNSAFE_LOG" >&2
+ exit 1
+fi
+
+SYMLINK_BASE="$TMP/outbox-link"
+ln -s "$HOME_DIR" "$SYMLINK_BASE"
+if PATH="$FAKEBIN:/usr/bin:/bin" \
+ LATTICE_HOME="$HOME_DIR" \
+ LATTICE_AGENT_BIN="$HOME_DIR/lattice-agent" \
+ LATTICE_AGENT_ENV="$HOME_DIR/lattice-agent.env" \
+ LATTICE_AGENT_STATE="$HOME_DIR/state" \
+ LATTICE_TASK_OUTBOX_DIR="$SYMLINK_BASE" \
+ LATTICE_SERVER="https://lattice.example.com" \
+ LATTICE_NODE_ID="node-test" \
+ LATTICE_NODE_TOKEN="node-token-test" \
+ sh "$ROOT/scripts/install.sh" >"$UNSAFE_LOG" 2>&1; then
+ echo "installer accepted symlinked task outbox" >&2
+ exit 1
+fi
+if ! grep -Fq "must not traverse symlinks" "$UNSAFE_LOG"; then
+ echo "symlinked outbox rejection was not explicit" >&2
+ cat "$UNSAFE_LOG" >&2
+ exit 1
+fi
+
+NESTED_SYMLINK_PARENT="$TMP/outbox-parent-link"
+ln -s "$HOME_DIR" "$NESTED_SYMLINK_PARENT"
+if PATH="$FAKEBIN:/usr/bin:/bin" \
+ LATTICE_HOME="$HOME_DIR" \
+ LATTICE_AGENT_BIN="$HOME_DIR/lattice-agent" \
+ LATTICE_AGENT_ENV="$HOME_DIR/lattice-agent.env" \
+ LATTICE_AGENT_STATE="$HOME_DIR/state" \
+ LATTICE_TASK_OUTBOX_DIR="$NESTED_SYMLINK_PARENT/nested" \
+ LATTICE_SERVER="https://lattice.example.com" \
+ LATTICE_NODE_ID="node-test" \
+ LATTICE_NODE_TOKEN="node-token-test" \
+ sh "$ROOT/scripts/install.sh" >"$UNSAFE_LOG" 2>&1; then
+ echo "installer accepted non-root-owned parent symlink for task outbox" >&2
+ exit 1
+fi
+if ! grep -Fq "must not traverse a non-root-owned symlink" "$UNSAFE_LOG"; then
+ echo "nested symlinked outbox rejection was not explicit" >&2
+ cat "$UNSAFE_LOG" >&2
+ exit 1
+fi
DARWIN_TMP="$TMP/darwin"
DARWIN_BIN="$DARWIN_TMP/bin"
DARWIN_HOME="$DARWIN_TMP/home"
@@ -207,6 +271,7 @@ if ! PATH="$DARWIN_BIN:/usr/bin:/bin" \
LATTICE_AGENT_BIN="$DARWIN_HOME/lattice-agent" \
LATTICE_AGENT_ENV="$DARWIN_HOME/lattice-agent.env" \
LATTICE_AGENT_STATE="$DARWIN_HOME/state&logs" \
+ LATTICE_TASK_OUTBOX_DIR="$DARWIN_HOME/outbox&state" \
LATTICE_AGENT_PLIST="$DARWIN_PLIST" \
LATTICE_SERVER="$special_server" \
LATTICE_NODE_ID="node&mac" \
@@ -239,7 +304,8 @@ for expected in \
'LATTICE_IP_RESOLVERShttps://api.example.com/ip?format=json&node=<self>' \
'LATTICE_IP_SCRIPT/usr/local/bin/ip&probe' \
'LATTICE_PUBLIC_IP203.0.113.7' \
- 'LATTICE_PUBLIC_IP62001:db8::7'
+ 'LATTICE_PUBLIC_IP62001:db8::7' \
+ 'LATTICE_TASK_OUTBOX_DIR'"$DARWIN_HOME"'/outbox&state'
do
if ! grep -Fq "$expected" "$DARWIN_PLIST"; then
echo "launchd plist missing persisted IP config: $expected" >&2