From 917f427770ddb9bdc2d55f9b9a5bebb9beb3ac3f Mon Sep 17 00:00:00 2001 From: gaoming <2168502171@qq.com> Date: Fri, 14 Aug 2026 15:23:17 +0800 Subject: [PATCH 1/2] feat(sandbox): add bounded Docker container I/O contract Schema v98 adds the sealed read-only input projection ledger, bounded stdout/stderr log capture receipts, strict output staging receipts, and atomic output commit receipts with per-file entries. The sandbox domain pins one path rule set for Windows and Linux, demuxes and redacts the attach stream under byte/line/deadline caps, walks the exported archive with traversal/symlink/duplicate/size rejection, and commits accepted manifests in one replay-safe store transaction. The I/O transport exposes exactly two closed daemon endpoints (attach with stream=0 and archive of the dedicated output mount). The application service stays unwired from any product entry, matching the lifecycle slice boundary. --- README.md | 1 + ...98-bounded-docker-container-io-contract.md | 72 ++ internal/application/docker_container_io.go | 153 +++++ .../application/docker_container_io_test.go | 269 ++++++++ internal/events/event.go | 14 + .../sandbox/docker_container_io_export.go | 615 ++++++++++++++++++ internal/sandbox/docker_container_io_http.go | 236 +++++++ .../sandbox/docker_container_io_http_test.go | 195 ++++++ internal/sandbox/docker_container_io_input.go | 236 +++++++ internal/sandbox/docker_container_io_log.go | 386 +++++++++++ internal/sandbox/docker_container_io_test.go | 507 +++++++++++++++ internal/store/analyzer_execution_test.go | 4 +- internal/store/docker_container_io.go | 173 +++++ internal/store/docker_container_io_test.go | 151 +++++ internal/store/migration_v98.go | 196 ++++++ internal/store/migrations.go | 2 +- internal/store/sqlite.go | 1 + 17 files changed, 3208 insertions(+), 3 deletions(-) create mode 100644 docs/adr/0098-bounded-docker-container-io-contract.md create mode 100644 internal/application/docker_container_io.go create mode 100644 internal/application/docker_container_io_test.go create mode 100644 internal/sandbox/docker_container_io_export.go create mode 100644 internal/sandbox/docker_container_io_http.go create mode 100644 internal/sandbox/docker_container_io_http_test.go create mode 100644 internal/sandbox/docker_container_io_input.go create mode 100644 internal/sandbox/docker_container_io_log.go create mode 100644 internal/sandbox/docker_container_io_test.go create mode 100644 internal/store/docker_container_io.go create mode 100644 internal/store/docker_container_io_test.go create mode 100644 internal/store/migration_v98.go diff --git a/README.md b/README.md index af9970d7..3fca2d70 100644 --- a/README.md +++ b/README.md @@ -271,6 +271,7 @@ go run ./cmd/cyberagent tui | v95 | Analyzer 结果、Artifact 与审计事件原子提交 | atomic Analyzer result, Artifact, and audit-event commit | | v96 | 用户审批档的精确宿主命令提案、审阅与恰好一次执行 | exact approval-mode host-command proposals, reviews, and exactly-once execution | | v97 | 持久 Docker 生命周期所有权、代际租约与崩溃恢复 | durable Docker lifecycle ownership, generation leases, and crash recovery | +| v98 | 有界 Docker 容器 I/O 合同：只读输入投影、日志限额与原子输出提交 | bounded Docker container I/O contract: read-only input projection, log capture limits, and atomic output commit | diff --git a/docs/adr/0098-bounded-docker-container-io-contract.md b/docs/adr/0098-bounded-docker-container-io-contract.md new file mode 100644 index 00000000..443ea86a --- /dev/null +++ b/docs/adr/0098-bounded-docker-container-io-contract.md @@ -0,0 +1,72 @@ +# ADR 0098: Bounded Docker Container I/O Contract + +Date: 2026-08-14 + +## Status + +Accepted. + +## Context + +The durable Docker lifecycle slice (schema v97, ADR 0096) proved exact +Stage/create/start/wait/terminate/cleanup ownership, but a container still had +no product-grade I/O contract: the Workspace boundary, log capture, output +extraction, and Artifact commit were not composed into one bounded, +fail-closed transaction set. Issue #39 requires that combination while +keeping product admission closed: no CLI, HTTP, Desktop, Agent, or Tool path +may reach it. + +Container output is untrusted evidence. Nothing a container writes may +automatically become prompt context, an approval decision, a command, or host +executable content, and the original Workspace must stay read-only from the +container perspective. + +## Decision + +1. Read-only input projection (sandbox_docker_input_projection.v1): a sealed + manifest of canonical container-relative paths, SHA-256, size, and media + type, bound to the lifecycle attempt/generation, plan, observation, Run, + Mission, Workspace, and spec fingerprints. The input mount target stays + fixed at /run/cyberagent/inputs with read-only access; the spec invariant + of exactly one writable (dedicated output) mount is unchanged. A mount + isolation verifier checks real inspection mounts: the output target is the + only writable mount in its tree and the input/workspace trees are + read-only. +2. Bounded log capture (sandbox_docker_log_capture.v1): stdout/stderr arrive + only through the fixed POST containers/{id}/attach with + logs=1&stderr=1&stdout=1&stream=0 (no live streaming). A demuxer bounds + each stream by bytes (256 KiB) and lines (4096), enforces a wall-clock + deadline, rejects malformed/oversized frames, replaces invalid UTF-8 and + counts the violations, redacts secrets, and records only metadata plus + digest-only receipts. Raw bytes never persist. +3. Output staging (sandbox_docker_output_staging.v1): the dedicated output + mount is exported only through GET containers/{id}/archive?path=. + The tar walker rejects absolute/traversal paths, Windows separators, + drive letters, symlinks/hardlinks/devices, duplicates, and non-canonical + names; caps file count (64), per-file size (4 MiB), and total size + (16 MiB); detects media types; redacts text content; and writes only + validated regular files into a Run-scoped process-local staging directory. + Rejected archives carry no trusted manifest. +4. Atomic output commit (sandbox_docker_output_commit.v1): an accepted + manifest must exactly match completed staging entries (path, digest, + size, media type). The staged files are re-read and re-hashed, then the + receipt and every entry commit in one SQLite transaction keyed by a + replay-safe operation key digest. Failure leaves no partial rows. +5. Schema v98 persists the four ledgers with strict CHECKs and foreign keys; + events record prepared/acquired/taken-over/failed/completed transitions. + The DockerContainerIOService is exported for tests only; no product entry + is wired, matching the lifecycle slice boundary. +6. Windows and Linux share one path rule set, so the same adversarial path + matrix is exercised on both CI hosts. + +## Consequences + +- Every container I/O step is bounded, digest-bound, replay-safe, and + fail-closed; the Workspace can only be modified through the already-gated + FileEdit boundaries, never by the container. +- Product admission (Policy/Approval/Budget/Network gates and CLI/HTTP/ + Desktop wiring) remains a separate slice; this contract does not authorize + anything. +- Raw logs and staged files are process-local and never reach SQLite or + events; only receipts and digests are durable. + diff --git a/internal/application/docker_container_io.go b/internal/application/docker_container_io.go new file mode 100644 index 00000000..562ffb0d --- /dev/null +++ b/internal/application/docker_container_io.go @@ -0,0 +1,153 @@ +package application + +import ( + "context" + "errors" + "time" + + "cyberagent-workbench/internal/idgen" + "cyberagent-workbench/internal/sandbox" +) + +// dockerContainerIOStore is the durable boundary satisfied by *store.SQLiteStore. +// Raw container bytes never reach it. +type dockerContainerIOStore interface { + InsertDockerInputProjection(ctx context.Context, + projection sandbox.DockerInputProjection) (bool, error) + InsertDockerLogCaptureReceipt(ctx context.Context, + receipt sandbox.DockerLogCaptureReceipt) (bool, error) + InsertDockerOutputStagingReceipt(ctx context.Context, + receipt sandbox.DockerOutputStagingReceipt) (bool, error) + CommitDockerOutputs(ctx context.Context, request sandbox.DockerOutputCommitRequest, + receipt sandbox.DockerOutputCommitReceipt) (bool, error) +} + +// DockerContainerIOService owns the bounded container I/O contract: sealed +// read-only input projections, bounded log capture, strict output staging, +// and atomic output commit. It grants no product execution authority and is +// not wired to any CLI, HTTP, or Desktop entry. +type DockerContainerIOService struct { + store dockerContainerIOStore + transport sandbox.DockerContainerIOTransport + now func() time.Time +} + +func NewDockerContainerIOService(store dockerContainerIOStore, + transport sandbox.DockerContainerIOTransport, +) *DockerContainerIOService { + return &DockerContainerIOService{store: store, transport: transport, now: time.Now} +} + +// ProjectInputs durably records the sealed read-only input projection. The +// returned flag is false when an identical projection already exists. +func (service *DockerContainerIOService) ProjectInputs(ctx context.Context, + projection sandbox.DockerInputProjection, +) (bool, error) { + if service == nil || service.store == nil { + return false, errors.New("docker container I/O store is required") + } + if err := ctx.Err(); err != nil { + return false, err + } + return service.store.InsertDockerInputProjection(ctx, projection) +} + +// CaptureLogs attaches to the exited container under the plan bounds, +// demuxes and redacts the raw stream in memory, and durably records the +// content-free receipt. Raw bytes never persist. +func (service *DockerContainerIOService) CaptureLogs(ctx context.Context, + plan sandbox.DockerLogCapturePlan, +) (sandbox.DockerLogCaptureReceipt, bool, error) { + if service == nil || service.store == nil || service.transport == nil { + return sandbox.DockerLogCaptureReceipt{}, false, errors.New("docker container I/O service is unavailable") + } + if err := plan.Validate(); err != nil { + return sandbox.DockerLogCaptureReceipt{}, false, err + } + if err := ctx.Err(); err != nil { + return sandbox.DockerLogCaptureReceipt{}, false, err + } + captureCtx, cancel := context.WithTimeout(ctx, time.Duration(plan.DurationSeconds)*time.Second) + defer cancel() + stream, err := service.transport.AttachLogs(captureCtx, plan) + if err != nil { + return sandbox.DockerLogCaptureReceipt{}, false, err + } + defer stream.Close() + records, status, err := sandbox.DecodeDockerLogFrames(captureCtx, plan, stream) + if err != nil { + return sandbox.DockerLogCaptureReceipt{}, false, err + } + receipt, err := sandbox.NewDockerLogCaptureReceipt( + idgen.New("sandbox-docker-log-capture"), plan.AttemptID, plan.Generation, plan.RunID, + plan.ContainerIDFingerprint, plan, records, status, service.now().UTC()) + if err != nil { + return sandbox.DockerLogCaptureReceipt{}, false, err + } + inserted, err := service.store.InsertDockerLogCaptureReceipt(ctx, receipt) + if err != nil { + return sandbox.DockerLogCaptureReceipt{}, false, err + } + return receipt, inserted, nil +} + +// StageOutputs exports the dedicated output mount as a tar archive, walks it +// under the plan bounds, redacts text content, and records the staging +// receipt. Staged files stay in the caller-owned staging directory. +func (service *DockerContainerIOService) StageOutputs(ctx context.Context, + plan sandbox.DockerOutputExportPlan, stagingRoot string, +) (sandbox.DockerOutputStagingReceipt, bool, error) { + if service == nil || service.store == nil || service.transport == nil { + return sandbox.DockerOutputStagingReceipt{}, false, errors.New("docker container I/O service is unavailable") + } + if err := plan.Validate(); err != nil { + return sandbox.DockerOutputStagingReceipt{}, false, err + } + if err := ctx.Err(); err != nil { + return sandbox.DockerOutputStagingReceipt{}, false, err + } + stream, err := service.transport.ExportOutputs(ctx, plan) + if err != nil { + return sandbox.DockerOutputStagingReceipt{}, false, err + } + defer stream.Close() + receipt, err := sandbox.StageDockerOutputArchive(ctx, plan, stream, stagingRoot, + idgen.New("sandbox-docker-output-staging"), service.now().UTC()) + if err != nil { + return sandbox.DockerOutputStagingReceipt{}, false, err + } + inserted, err := service.store.InsertDockerOutputStagingReceipt(ctx, receipt) + if err != nil { + return sandbox.DockerOutputStagingReceipt{}, false, err + } + return receipt, inserted, nil +} + +// CommitOutputs verifies the accepted manifest against the completed staging +// receipt and the staged files, then commits the receipt and every entry in +// one atomic store transaction. A failed commit leaves no partial rows. +func (service *DockerContainerIOService) CommitOutputs(ctx context.Context, + request sandbox.DockerOutputCommitRequest, staging sandbox.DockerOutputStagingReceipt, + stagingRoot string, +) (sandbox.DockerOutputCommitReceipt, bool, error) { + if service == nil || service.store == nil { + return sandbox.DockerOutputCommitReceipt{}, false, errors.New("docker container I/O store is required") + } + if err := request.Binds(staging); err != nil { + return sandbox.DockerOutputCommitReceipt{}, false, err + } + if _, err := sandbox.VerifyDockerOutputCommit(stagingRoot, request); err != nil { + return sandbox.DockerOutputCommitReceipt{}, false, err + } + receipt, err := sandbox.NewDockerOutputCommitReceipt( + idgen.New("sandbox-docker-output-commit"), request.AttemptID, request.Generation, + request.RunID, request.WorkspaceID, request, service.now().UTC()) + if err != nil { + return sandbox.DockerOutputCommitReceipt{}, false, err + } + committed, err := service.store.CommitDockerOutputs(ctx, request, receipt) + if err != nil { + return sandbox.DockerOutputCommitReceipt{}, false, err + } + return receipt, committed, nil +} diff --git a/internal/application/docker_container_io_test.go b/internal/application/docker_container_io_test.go new file mode 100644 index 00000000..1a8a5c04 --- /dev/null +++ b/internal/application/docker_container_io_test.go @@ -0,0 +1,269 @@ +package application + +import ( + "archive/tar" + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "io" + "testing" + "time" + + "cyberagent-workbench/internal/idgen" + "cyberagent-workbench/internal/sandbox" +) + +func serviceTestDigest(value string) string { + sum := sha256.Sum256([]byte(value)) + return hex.EncodeToString(sum[:]) +} + +type fakeDockerContainerIOStore struct { + failProjection bool + failLogReceipt bool + failStaging bool + failCommit bool + projections []sandbox.DockerInputProjection + logReceipts []sandbox.DockerLogCaptureReceipt + stagingReceipts []sandbox.DockerOutputStagingReceipt + commits []sandbox.DockerOutputCommitReceipt +} + +func (store *fakeDockerContainerIOStore) InsertDockerInputProjection(ctx context.Context, + projection sandbox.DockerInputProjection) (bool, error) { + if store.failProjection { + return false, errors.New("projection store failed") + } + for _, existing := range store.projections { + if existing.ProjectionFingerprint == projection.ProjectionFingerprint { + return false, nil + } + } + store.projections = append(store.projections, projection) + return true, nil +} + +func (store *fakeDockerContainerIOStore) InsertDockerLogCaptureReceipt(ctx context.Context, + receipt sandbox.DockerLogCaptureReceipt) (bool, error) { + if store.failLogReceipt { + return false, errors.New("log store failed") + } + store.logReceipts = append(store.logReceipts, receipt) + return true, nil +} + +func (store *fakeDockerContainerIOStore) InsertDockerOutputStagingReceipt(ctx context.Context, + receipt sandbox.DockerOutputStagingReceipt) (bool, error) { + if store.failStaging { + return false, errors.New("staging store failed") + } + store.stagingReceipts = append(store.stagingReceipts, receipt) + return true, nil +} + +func (store *fakeDockerContainerIOStore) CommitDockerOutputs(ctx context.Context, + request sandbox.DockerOutputCommitRequest, receipt sandbox.DockerOutputCommitReceipt) (bool, error) { + if store.failCommit { + return false, errors.New("commit store failed") + } + for _, existing := range store.commits { + if existing.OperationKeyDigest == receipt.OperationKeyDigest { + return false, nil + } + } + store.commits = append(store.commits, receipt) + return true, nil +} + +type fakeDockerContainerIOTransport struct { + failAttach bool + failExport bool + attachBody []byte + exportBody []byte + attaches int + exports int +} + +func (transport *fakeDockerContainerIOTransport) Endpoint() sandbox.DockerObservationEndpoint { + endpoint, _ := sandbox.NewDockerObservationEndpoint(sandbox.DockerObservationEndpointLocalUnix) + return endpoint +} + +func (transport *fakeDockerContainerIOTransport) AttachLogs(ctx context.Context, + plan sandbox.DockerLogCapturePlan) (io.ReadCloser, error) { + transport.attaches++ + if transport.failAttach { + return nil, errors.New("attach failed") + } + return io.NopCloser(bytes.NewReader(transport.attachBody)), nil +} + +func (transport *fakeDockerContainerIOTransport) ExportOutputs(ctx context.Context, + plan sandbox.DockerOutputExportPlan) (io.ReadCloser, error) { + transport.exports++ + if transport.failExport { + return nil, errors.New("export failed") + } + return io.NopCloser(bytes.NewReader(transport.exportBody)), nil +} + +func dockerLogFramePayload(stream byte, payload string) []byte { + frame := make([]byte, 8+len(payload)) + frame[0] = stream + frame[4] = byte(len(payload) >> 24) + frame[5] = byte(len(payload) >> 16) + frame[6] = byte(len(payload) >> 8) + frame[7] = byte(len(payload)) + copy(frame[8:], payload) + return frame +} + +func serviceTestExportPlan(t *testing.T, attemptID string) sandbox.DockerOutputExportPlan { + t.Helper() + plan, err := sandbox.NewDockerOutputExportPlan(attemptID, 1, + idgen.New("sandbox-docker-run"), serviceTestDigest("container"), + "/run/cyberagent/outputs", sandbox.MaxDockerOutputFiles, + sandbox.MaxDockerOutputFileBytes, sandbox.MaxDockerOutputTotalBytes) + if err != nil { + t.Fatal(err) + } + return plan +} + +func TestDockerContainerIOServiceFullContractFlow(t *testing.T) { + store := &fakeDockerContainerIOStore{} + transport := &fakeDockerContainerIOTransport{} + service := NewDockerContainerIOService(store, transport) + ctx := context.Background() + attemptID := idgen.New("sandbox-docker-attempt") + + t.Run("input projection", func(t *testing.T) { + projection, err := sandbox.NewDockerInputProjection(idgen.New("sandbox-docker-input"), + attemptID, 1, idgen.New("sandbox-docker-plan"), idgen.New("sandbox-docker-observation"), + idgen.New("sandbox-docker-run"), idgen.New("sandbox-docker-mission"), + idgen.New("sandbox-docker-workspace"), serviceTestDigest("inputs"), + serviceTestDigest("spec"), serviceTestDigest("authority"), + sandbox.DockerInputArtifactMountTarget, + []sandbox.DockerInputProjectionEntry{{Ordinal: 1, Path: "data/in.json", + SHA256: serviceTestDigest("in"), SizeBytes: 2, MediaType: "application/json"}}, + time.Now().UTC()) + if err != nil { + t.Fatal(err) + } + inserted, err := service.ProjectInputs(ctx, projection) + if err != nil || !inserted || len(store.projections) != 1 { + t.Fatalf("project = %t, %v", inserted, err) + } + store.failProjection = true + if _, err := service.ProjectInputs(ctx, projection); err == nil { + t.Fatal("store failure was swallowed") + } + store.failProjection = false + }) + + t.Run("log capture", func(t *testing.T) { + transport.attachBody = append(dockerLogFramePayload(1, "hello\n"), + dockerLogFramePayload(2, "oops\n")...) + plan, err := sandbox.NewDockerLogCapturePlan(attemptID, 1, + idgen.New("sandbox-docker-run"), serviceTestDigest("container"), 4096, 64, 30) + if err != nil { + t.Fatal(err) + } + receipt, inserted, err := service.CaptureLogs(ctx, plan) + if err != nil || !inserted || receipt.Status != sandbox.DockerLogCaptureStatusCompleted || + receipt.TotalBytes != 11 || len(store.logReceipts) != 1 { + t.Fatalf("capture = %#v, %t, %v", receipt, inserted, err) + } + transport.failAttach = true + if _, _, err := service.CaptureLogs(ctx, plan); err == nil { + t.Fatal("attach failure was swallowed") + } + transport.failAttach = false + }) + + t.Run("staging and atomic commit", func(t *testing.T) { + exportPlan := serviceTestExportPlan(t, attemptID) + transport.exportBody = buildServiceOutputTar(t, map[string]string{ + "report/result.json": "{\"ok\": true}\n", + }) + stagingRoot := t.TempDir() + staging, inserted, err := service.StageOutputs(ctx, exportPlan, stagingRoot) + if err != nil || !inserted || staging.Status != sandbox.DockerOutputStagingStatusCompleted || + len(store.stagingReceipts) != 1 || staging.FileCount != 1 { + t.Fatalf("stage = %#v, %t, %v", staging, inserted, err) + } + accepted := make([]sandbox.DockerOutputCommitEntry, 0, len(staging.Entries)) + for _, entry := range staging.Entries { + accepted = append(accepted, sandbox.DockerOutputCommitEntry{ + Path: entry.Path, SHA256: entry.SHA256, SizeBytes: entry.SizeBytes, + MediaType: entry.MediaType, + }) + } + request, err := sandbox.NewDockerOutputCommitRequest(attemptID, 1, staging.RunID, + idgen.New("sandbox-docker-workspace"), staging.ID, serviceTestDigest("operation"), + accepted) + if err != nil { + t.Fatal(err) + } + receipt, committed, err := service.CommitOutputs(ctx, request, staging, stagingRoot) + if err != nil || !committed || len(store.commits) != 1 || receipt.CommittedCount != 1 { + t.Fatalf("commit = %#v, %t, %v", receipt, committed, err) + } + _, replayed, err := service.CommitOutputs(ctx, request, staging, stagingRoot) + if err != nil || replayed || len(store.commits) != 1 { + t.Fatalf("commit replay = %t, %v", replayed, err) + } + store.failCommit = true + if _, _, err := service.CommitOutputs(ctx, request, staging, stagingRoot); err == nil { + t.Fatal("commit failure was swallowed") + } + if len(store.commits) != 1 { + t.Fatal("failed commit left partial rows") + } + store.failCommit = false + transport.failExport = true + if _, _, err := service.StageOutputs(ctx, exportPlan, t.TempDir()); err == nil { + t.Fatal("export failure was swallowed") + } + transport.failExport = false + }) +} + +func buildServiceOutputTar(t *testing.T, files map[string]string) []byte { + t.Helper() + var buffer bytes.Buffer + writer := tar.NewWriter(&buffer) + for name, content := range files { + if err := writer.WriteHeader(&tar.Header{Name: name, Mode: 0o644, + Size: int64(len(content)), Typeflag: tar.TypeReg}); err != nil { + t.Fatal(err) + } + if _, err := writer.Write([]byte(content)); err != nil { + t.Fatal(err) + } + } + if err := writer.Close(); err != nil { + t.Fatal(err) + } + return buffer.Bytes() +} + +func TestDockerContainerIOServiceUnavailableTransport(t *testing.T) { + store := &fakeDockerContainerIOStore{} + transport := sandbox.NewUnavailableDockerContainerIOTransport("unavailable", "unsupported") + service := NewDockerContainerIOService(store, transport) + plan, err := sandbox.NewDockerLogCapturePlan(idgen.New("sandbox-docker-attempt"), 1, + idgen.New("sandbox-docker-run"), serviceTestDigest("container"), 1024, 64, 60) + if err != nil { + t.Fatal(err) + } + if _, _, err := service.CaptureLogs(context.Background(), plan); err == nil { + t.Fatal("unavailable attach succeeded") + } + if _, _, err := service.StageOutputs(context.Background(), + serviceTestExportPlan(t, idgen.New("sandbox-docker-attempt")), t.TempDir()); err == nil { + t.Fatal("unavailable export succeeded") + } +} diff --git a/internal/events/event.go b/internal/events/event.go index 04d732a1..e7c6f008 100644 --- a/internal/events/event.go +++ b/internal/events/event.go @@ -238,6 +238,20 @@ const ( SandboxDockerProductionEvidenceHarnessReconciledEvent = "sandbox.docker_production_evidence_harness_reconciled" SandboxDockerProductionEvidenceHarnessCompletedEvent = "sandbox.docker_production_evidence_harness_completed" SandboxDockerProductionEvidenceReviewedEvent = "sandbox.docker_production_evidence_reviewed" + SandboxDockerInputProjectionPreparedEvent = "sandbox.docker_input_projection_prepared" + SandboxDockerLogCapturePreparedEvent = "sandbox.docker_log_capture_prepared" + SandboxDockerLogCaptureAcquiredEvent = "sandbox.docker_log_capture_acquired" + SandboxDockerLogCaptureTakenOverEvent = "sandbox.docker_log_capture_taken_over" + SandboxDockerLogCaptureFailedEvent = "sandbox.docker_log_capture_failed" + SandboxDockerLogCaptureCompletedEvent = "sandbox.docker_log_capture_completed" + SandboxDockerOutputStagingPreparedEvent = "sandbox.docker_output_staging_prepared" + SandboxDockerOutputStagingAcquiredEvent = "sandbox.docker_output_staging_acquired" + SandboxDockerOutputStagingTakenOverEvent = "sandbox.docker_output_staging_taken_over" + SandboxDockerOutputStagingFailedEvent = "sandbox.docker_output_staging_failed" + SandboxDockerOutputStagingCompletedEvent = "sandbox.docker_output_staging_completed" + SandboxDockerOutputCommitPreparedEvent = "sandbox.docker_output_commit_prepared" + SandboxDockerOutputCommitFailedEvent = "sandbox.docker_output_commit_failed" + SandboxDockerOutputCommitCompletedEvent = "sandbox.docker_output_commit_completed" ) type Event struct { diff --git a/internal/sandbox/docker_container_io_export.go b/internal/sandbox/docker_container_io_export.go new file mode 100644 index 00000000..515c4f69 --- /dev/null +++ b/internal/sandbox/docker_container_io_export.go @@ -0,0 +1,615 @@ +package sandbox + +import ( + "archive/tar" + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "io" + "mime" + "os" + "path" + "path/filepath" + "sort" + "strconv" + "strings" + "time" + "unicode/utf8" + + "cyberagent-workbench/internal/redact" +) + +const ( + DockerOutputStagingProtocolVersion = "sandbox_docker_output_staging.v1" + DockerOutputCommitProtocolVersion = "sandbox_docker_output_commit.v1" + DockerOutputStagingStatusCompleted = "completed" + DockerOutputStagingStatusTruncated = "truncated_bytes" + DockerOutputStagingStatusInvalid = "invalid_archive" + DockerOutputStagingStatusPath = "rejected_path" + DockerOutputStagingStatusLink = "rejected_link" + DockerOutputStagingStatusDuplicate = "rejected_duplicate" + DockerOutputCommitStatusCommitted = "committed" + MaxDockerOutputFiles = 64 + MaxDockerOutputFileBytes = 4 * 1024 * 1024 + MaxDockerOutputTotalBytes = 16 * 1024 * 1024 + MaxDockerOutputArchiveHeaderBytes = 1024 * 1024 +) + +// DockerOutputExportPlan pins the exact dedicated output mount export for one +// lifecycle attempt. The archive is read through the strict staging walker; +// nothing lands on the host outside the Run-scoped staging directory. +type DockerOutputExportPlan struct { + ProtocolVersion string + AttemptID string + Generation int64 + RunID string + ContainerIDFingerprint string + OutputMountTarget string + MaxFiles int + MaxFileBytes int64 + MaxTotalBytes int64 + ExportFingerprint string +} + +func NewDockerOutputExportPlan(attemptID string, generation int64, runID, + containerIDFingerprint, outputMountTarget string, maxFiles int, + maxFileBytes, maxTotalBytes int64, +) (DockerOutputExportPlan, error) { + plan := DockerOutputExportPlan{ + ProtocolVersion: DockerOutputStagingProtocolVersion, AttemptID: attemptID, + Generation: generation, RunID: runID, ContainerIDFingerprint: containerIDFingerprint, + OutputMountTarget: outputMountTarget, MaxFiles: maxFiles, MaxFileBytes: maxFileBytes, + MaxTotalBytes: maxTotalBytes, + } + plan.ExportFingerprint = dockerOutputExportPlanFingerprint(plan) + return plan, plan.Validate() +} + +func (plan DockerOutputExportPlan) Validate() error { + for label, value := range map[string]string{ + "docker output export attempt id": plan.AttemptID, + "docker output export Run id": plan.RunID, + } { + if validateStoredIdentity(label, value) != nil { + return errors.New("docker output export identity is invalid") + } + } + if plan.ProtocolVersion != DockerOutputStagingProtocolVersion || plan.Generation != 1 || + !validDigest(plan.ContainerIDFingerprint) || + validateVirtualPath("docker output mount target", plan.OutputMountTarget) != nil || + plan.MaxFiles < 1 || plan.MaxFiles > MaxDockerOutputFiles || + plan.MaxFileBytes < 1 || plan.MaxFileBytes > MaxDockerOutputFileBytes || + plan.MaxTotalBytes < plan.MaxFileBytes || plan.MaxTotalBytes > MaxDockerOutputTotalBytes || + plan.ExportFingerprint != dockerOutputExportPlanFingerprint(plan) { + return errors.New("docker output export plan violates a fixed bound") + } + return nil +} + +func dockerOutputExportPlanFingerprint(plan DockerOutputExportPlan) string { + return fingerprint(DockerOutputStagingProtocolVersion, plan.AttemptID, + strconv.FormatInt(plan.Generation, 10), plan.RunID, plan.ContainerIDFingerprint, + plan.OutputMountTarget, strconv.Itoa(plan.MaxFiles), + strconv.FormatInt(plan.MaxFileBytes, 10), strconv.FormatInt(plan.MaxTotalBytes, 10)) +} + +// DockerStagedOutputEntry is one validated, redacted, staged output file. The +// container path is canonical; the digest binds the exact staged bytes. +type DockerStagedOutputEntry struct { + Path string + SHA256 string + SizeBytes int64 + MediaType string + Redacted bool +} + +func (entry DockerStagedOutputEntry) Validate() error { + if validateContainerRelativePath("docker staged output path", entry.Path) != nil || + !validDigest(entry.SHA256) || entry.SizeBytes < 1 || entry.SizeBytes > MaxDockerOutputFileBytes || + validateBoundedText("docker staged media type", entry.MediaType, 256, false) != nil { + return errors.New("docker staged output entry is invalid") + } + if _, _, err := mime.ParseMediaType(entry.MediaType); err != nil { + return errors.New("docker staged output media type is invalid") + } + return nil +} + +// DockerOutputStagingReceipt is the durable, content-free evidence of one +// bounded output export. Raw file bytes stay in the process-local staging +// directory and never enter SQLite or events. +type DockerOutputStagingReceipt struct { + ProtocolVersion string + ID string + AttemptID string + Generation int64 + RunID string + ContainerIDFingerprint string + Status string + Entries []DockerStagedOutputEntry + FileCount int + TotalBytes int64 + RedactedCount int + EntryDigestSet string + ExportFingerprint string + ReceiptFingerprint string + CreatedAt time.Time +} + +func NewDockerOutputStagingReceipt(id, attemptID string, generation int64, runID, + containerIDFingerprint string, plan DockerOutputExportPlan, + entries []DockerStagedOutputEntry, status string, createdAt time.Time, +) (DockerOutputStagingReceipt, error) { + receipt := DockerOutputStagingReceipt{ + ProtocolVersion: DockerOutputStagingProtocolVersion, ID: id, AttemptID: attemptID, + Generation: generation, RunID: runID, ContainerIDFingerprint: containerIDFingerprint, + Status: status, Entries: append([]DockerStagedOutputEntry(nil), entries...), + ExportFingerprint: plan.ExportFingerprint, CreatedAt: createdAt, + } + receipt.FileCount = len(receipt.Entries) + for _, entry := range receipt.Entries { + receipt.TotalBytes += entry.SizeBytes + if entry.Redacted { + receipt.RedactedCount++ + } + } + receipt.EntryDigestSet = dockerStagedOutputDigestSet(receipt.Entries) + receipt.ReceiptFingerprint = dockerOutputStagingReceiptFingerprint(receipt) + return receipt, receipt.Validate() +} + +func (receipt DockerOutputStagingReceipt) Validate() error { + for label, value := range map[string]string{ + "docker staging receipt id": receipt.ID, "docker staging receipt attempt id": receipt.AttemptID, + "docker staging receipt Run id": receipt.RunID, + } { + if validateStoredIdentity(label, value) != nil { + return errors.New("docker output staging receipt identity is invalid") + } + } + if receipt.ProtocolVersion != DockerOutputStagingProtocolVersion || receipt.Generation != 1 || + !validDigest(receipt.ContainerIDFingerprint) || !validDigest(receipt.ExportFingerprint) || + !validDockerOutputStagingStatus(receipt.Status) || receipt.CreatedAt.IsZero() || + receipt.FileCount != len(receipt.Entries) || len(receipt.Entries) > MaxDockerOutputFiles { + return errors.New("docker output staging receipt is invalid") + } + var total int64 + redactedCount := 0 + previous := "" + for _, entry := range receipt.Entries { + if entry.Validate() != nil || (previous != "" && previous >= entry.Path) { + return errors.New("docker staged output entry sequence is invalid") + } + previous = entry.Path + total += entry.SizeBytes + if entry.Redacted { + redactedCount++ + } + } + if receipt.RedactedCount != redactedCount || receipt.TotalBytes != total || + receipt.EntryDigestSet != dockerStagedOutputDigestSet(receipt.Entries) || + receipt.ReceiptFingerprint != dockerOutputStagingReceiptFingerprint(receipt) { + return errors.New("docker output staging receipt aggregate is invalid") + } + switch receipt.Status { + case DockerOutputStagingStatusCompleted, DockerOutputStagingStatusTruncated: + case DockerOutputStagingStatusInvalid, DockerOutputStagingStatusPath, + DockerOutputStagingStatusLink, DockerOutputStagingStatusDuplicate: + if len(receipt.Entries) != 0 { + return errors.New("rejected docker staging receipt carries entries") + } + default: + return errors.New("docker output staging status is invalid") + } + return nil +} + +func validDockerOutputStagingStatus(status string) bool { + switch status { + case DockerOutputStagingStatusCompleted, DockerOutputStagingStatusTruncated, + DockerOutputStagingStatusInvalid, DockerOutputStagingStatusPath, + DockerOutputStagingStatusLink, DockerOutputStagingStatusDuplicate: + return true + default: + return false + } +} + +func dockerOutputStagingReceiptFingerprint(receipt DockerOutputStagingReceipt) string { + parts := []string{DockerOutputStagingProtocolVersion, receipt.ID, receipt.AttemptID, + strconv.FormatInt(receipt.Generation, 10), receipt.RunID, + receipt.ContainerIDFingerprint, receipt.Status, receipt.ExportFingerprint, + receipt.EntryDigestSet, strconv.Itoa(receipt.FileCount)} + return fingerprint(parts...) +} + +func dockerStagedOutputDigestSet(entries []DockerStagedOutputEntry) string { + parts := make([]string, 0, len(entries)) + for _, entry := range entries { + parts = append(parts, entry.Path+":"+entry.SHA256) + } + sort.Strings(parts) + return fingerprint(parts...) +} + +// StageDockerOutputArchive walks the exported tar stream under the plan +// bounds, rejects escaping or linked entries, redacts text content, and writes +// only validated regular files into stagingRoot. The returned receipt records +// metadata and digests; raw bytes remain on the process-local filesystem. +func StageDockerOutputArchive(ctx context.Context, plan DockerOutputExportPlan, + tarStream io.Reader, stagingRoot, id string, createdAt time.Time, +) (DockerOutputStagingReceipt, error) { + if err := plan.Validate(); err != nil { + return DockerOutputStagingReceipt{}, err + } + if validateStoredIdentity("docker staging receipt id", id) != nil || createdAt.IsZero() { + return DockerOutputStagingReceipt{}, errors.New("docker output staging identity is invalid") + } + if tarStream == nil { + return DockerOutputStagingReceipt{}, errors.New("docker output archive is required") + } + info, err := os.Stat(stagingRoot) + if err != nil || !info.IsDir() { + return DockerOutputStagingReceipt{}, errors.New("docker output staging root is required") + } + stagingRoot, err = filepath.Abs(stagingRoot) + if err != nil { + return DockerOutputStagingReceipt{}, err + } + reader := tar.NewReader(io.LimitReader(tarStream, plan.MaxTotalBytes+MaxDockerOutputArchiveHeaderBytes)) + entries := make([]DockerStagedOutputEntry, 0, plan.MaxFiles) + seen := make(map[string]bool) + status := DockerOutputStagingStatusCompleted + var total int64 + headers := 0 + for { + if err := ctx.Err(); err != nil { + return DockerOutputStagingReceipt{}, err + } + header, err := reader.Next() + if err == io.EOF { + break + } + if err != nil { + status = DockerOutputStagingStatusInvalid + break + } + headers++ + if headers > MaxDockerOutputFiles*4 { + status = DockerOutputStagingStatusTruncated + break + } + relative, err := normalizeDockerArchiveName(header.Name) + if err != nil { + status = DockerOutputStagingStatusPath + break + } + switch header.Typeflag { + case tar.TypeDir: + continue + case tar.TypeReg, tar.TypeRegA: + default: + status = DockerOutputStagingStatusLink + break + } + if status != DockerOutputStagingStatusCompleted { + break + } + if seen[relative] { + status = DockerOutputStagingStatusDuplicate + break + } + if header.Size < 0 || header.Size > plan.MaxFileBytes || + total+header.Size > plan.MaxTotalBytes || len(entries) >= plan.MaxFiles { + status = DockerOutputStagingStatusTruncated + break + } + content, err := io.ReadAll(io.LimitReader(reader, header.Size+1)) + if err != nil || int64(len(content)) != header.Size { + status = DockerOutputStagingStatusInvalid + break + } + seen[relative] = true + redacted := utf8.Valid(content) && redact.String(string(content)) != string(content) + if utf8.Valid(content) { + redactedContent := redact.String(string(content)) + redacted = redactedContent != string(content) + content = []byte(redactedContent) + } + digest := sha256.Sum256(content) + mediaType := dockerOutputMediaType(relative, content) + entry := DockerStagedOutputEntry{ + Path: relative, SHA256: hex.EncodeToString(digest[:]), + SizeBytes: int64(len(content)), MediaType: mediaType, Redacted: redacted, + } + if err := writeDockerStagedFile(stagingRoot, relative, content); err != nil { + return DockerOutputStagingReceipt{}, err + } + entries = append(entries, entry) + total += entry.SizeBytes + } + // Rejected archives carry no trusted manifest: drop any partially staged + // entries so the receipt can never be used for a commit. + if status != DockerOutputStagingStatusCompleted && status != DockerOutputStagingStatusTruncated { + entries = nil + } + sort.Slice(entries, func(i, j int) bool { return entries[i].Path < entries[j].Path }) + return NewDockerOutputStagingReceipt(id, plan.AttemptID, plan.Generation, plan.RunID, + plan.ContainerIDFingerprint, plan, entries, status, createdAt.UTC()) +} + +// normalizeDockerArchiveName accepts a plain relative name or a leading "./" +// prefix (which the Docker archive API emits for directory entries) and then +// applies the strict container-relative path rules. +func normalizeDockerArchiveName(name string) (string, error) { + cleaned := strings.TrimPrefix(name, "./") + if cleaned == "" || cleaned == name && strings.HasPrefix(name, "../") { + return "", errors.New("docker archive name is invalid") + } + if err := validateContainerRelativePath("docker archive path", cleaned); err != nil { + return "", err + } + return cleaned, nil +} + +func writeDockerStagedFile(stagingRoot, relative string, content []byte) error { + full := filepath.Join(stagingRoot, filepath.FromSlash(relative)) + if !workspacePathWithin(full, stagingRoot) || filepath.Clean(full) != full { + return errors.New("docker staged output path escaped the staging root") + } + if err := os.MkdirAll(filepath.Dir(full), 0o700); err != nil { + return err + } + return os.WriteFile(full, content, 0o600) +} + +func dockerOutputMediaType(relative string, content []byte) string { + switch strings.ToLower(path.Ext(relative)) { + case ".json": + return "application/json" + case ".txt", ".log": + return "text/plain" + case ".md": + return "text/markdown" + case ".csv": + return "text/csv" + case ".html", ".htm": + return "text/html" + case ".xml": + return "application/xml" + case ".yaml", ".yml": + return "application/yaml" + } + if utf8.Valid(content) { + return "text/plain" + } + return "application/octet-stream" +} + +// DockerOutputCommitEntry is one accepted output file. Acceptance requires +// the exact path, digest, size, and media type of a staged entry. +type DockerOutputCommitEntry struct { + Path string + SHA256 string + SizeBytes int64 + MediaType string +} + +func (entry DockerOutputCommitEntry) Validate() error { + if validateContainerRelativePath("docker commit path", entry.Path) != nil || + !validDigest(entry.SHA256) || entry.SizeBytes < 1 || entry.SizeBytes > MaxDockerOutputFileBytes || + validateBoundedText("docker commit media type", entry.MediaType, 256, false) != nil { + return errors.New("docker output commit entry is invalid") + } + if _, _, err := mime.ParseMediaType(entry.MediaType); err != nil { + return errors.New("docker output commit media type is invalid") + } + return nil +} + +// DockerOutputCommitRequest freezes one explicit acceptance manifest. Only +// entries that exactly match the staging receipt can be committed, and the +// operation key makes the commit idempotent. +type DockerOutputCommitRequest struct { + ProtocolVersion string + AttemptID string + Generation int64 + RunID string + WorkspaceID string + StagingReceiptID string + OperationKeyDigest string + AcceptedEntries []DockerOutputCommitEntry + RequestFingerprint string +} + +func NewDockerOutputCommitRequest(attemptID string, generation int64, runID, workspaceID, + stagingReceiptID, operationKeyDigest string, accepted []DockerOutputCommitEntry, +) (DockerOutputCommitRequest, error) { + request := DockerOutputCommitRequest{ + ProtocolVersion: DockerOutputCommitProtocolVersion, AttemptID: attemptID, + Generation: generation, RunID: runID, WorkspaceID: workspaceID, + StagingReceiptID: stagingReceiptID, OperationKeyDigest: operationKeyDigest, + AcceptedEntries: append([]DockerOutputCommitEntry(nil), accepted...), + } + request.RequestFingerprint = DockerOutputCommitRequestFingerprint(request) + return request, request.Validate() +} + +func (request DockerOutputCommitRequest) Validate() error { + for label, value := range map[string]string{ + "docker commit attempt id": request.AttemptID, "docker commit Run id": request.RunID, + "docker commit workspace id": request.WorkspaceID, + "docker commit staging receipt id": request.StagingReceiptID, + } { + if validateStoredIdentity(label, value) != nil { + return errors.New("docker output commit request identity is invalid") + } + } + if request.ProtocolVersion != DockerOutputCommitProtocolVersion || request.Generation != 1 || + !validDigest(request.OperationKeyDigest) || + len(request.AcceptedEntries) == 0 || len(request.AcceptedEntries) > MaxDockerOutputFiles || + request.RequestFingerprint != DockerOutputCommitRequestFingerprint(request) { + return errors.New("docker output commit request is invalid") + } + previous := "" + for _, entry := range request.AcceptedEntries { + if entry.Validate() != nil || (previous != "" && previous >= entry.Path) { + return errors.New("docker output commit entry sequence is invalid") + } + previous = entry.Path + } + return nil +} + +// Binds checks that every accepted entry exactly matches a staged entry. +func (request DockerOutputCommitRequest) Binds(staging DockerOutputStagingReceipt) error { + if err := request.Validate(); err != nil { + return err + } + if staging.Status != DockerOutputStagingStatusCompleted || + staging.AttemptID != request.AttemptID || staging.Generation != request.Generation || + staging.RunID != request.RunID || staging.ID != request.StagingReceiptID { + return errors.New("docker output commit does not bind the completed staging receipt") + } + staged := make(map[string]DockerStagedOutputEntry, len(staging.Entries)) + for _, entry := range staging.Entries { + staged[entry.Path] = entry + } + for _, entry := range request.AcceptedEntries { + expected, ok := staged[entry.Path] + if !ok || expected.SHA256 != entry.SHA256 || expected.SizeBytes != entry.SizeBytes || + expected.MediaType != entry.MediaType { + return errors.New("docker output commit accepted an un-staged entry") + } + } + return nil +} + +func DockerOutputCommitRequestFingerprint(request DockerOutputCommitRequest) string { + parts := []string{DockerOutputCommitProtocolVersion, request.AttemptID, + strconv.FormatInt(request.Generation, 10), request.RunID, request.WorkspaceID, + request.StagingReceiptID, request.OperationKeyDigest, + strconv.Itoa(len(request.AcceptedEntries))} + for _, entry := range request.AcceptedEntries { + parts = append(parts, entry.Path, entry.SHA256, strconv.FormatInt(entry.SizeBytes, 10), + entry.MediaType) + } + return fingerprint(parts...) +} + +// DockerOutputCommitReceipt is the durable, content-free evidence of one +// atomic commit. All entries are inserted with the receipt in one +// transaction; a failed commit leaves no partial rows. +type DockerOutputCommitReceipt struct { + ProtocolVersion string + ID string + AttemptID string + Generation int64 + RunID string + WorkspaceID string + OperationKeyDigest string + RequestFingerprint string + Status string + Entries []DockerOutputCommitEntry + CommittedCount int + CommittedDigestSet string + ReceiptFingerprint string + CreatedAt time.Time +} + +func NewDockerOutputCommitReceipt(id, attemptID string, generation int64, runID, + workspaceID string, request DockerOutputCommitRequest, createdAt time.Time, +) (DockerOutputCommitReceipt, error) { + receipt := DockerOutputCommitReceipt{ + ProtocolVersion: DockerOutputCommitProtocolVersion, ID: id, AttemptID: attemptID, + Generation: generation, RunID: runID, WorkspaceID: workspaceID, + OperationKeyDigest: request.OperationKeyDigest, RequestFingerprint: request.RequestFingerprint, + Status: DockerOutputCommitStatusCommitted, + Entries: append([]DockerOutputCommitEntry(nil), request.AcceptedEntries...), + CreatedAt: createdAt, + } + receipt.CommittedCount = len(receipt.Entries) + receipt.CommittedDigestSet = dockerCommittedDigestSet(receipt.Entries) + receipt.ReceiptFingerprint = dockerOutputCommitReceiptFingerprint(receipt) + return receipt, receipt.Validate() +} + +func (receipt DockerOutputCommitReceipt) Validate() error { + for label, value := range map[string]string{ + "docker commit receipt id": receipt.ID, "docker commit receipt attempt id": receipt.AttemptID, + "docker commit receipt Run id": receipt.RunID, "docker commit receipt workspace id": receipt.WorkspaceID, + } { + if validateStoredIdentity(label, value) != nil { + return errors.New("docker output commit receipt identity is invalid") + } + } + if receipt.ProtocolVersion != DockerOutputCommitProtocolVersion || receipt.Generation != 1 || + !validDigest(receipt.OperationKeyDigest) || !validDigest(receipt.RequestFingerprint) || + receipt.Status != DockerOutputCommitStatusCommitted || receipt.CreatedAt.IsZero() || + receipt.CommittedCount != len(receipt.Entries) || len(receipt.Entries) == 0 || + len(receipt.Entries) > MaxDockerOutputFiles { + return errors.New("docker output commit receipt is invalid") + } + previous := "" + for _, entry := range receipt.Entries { + if entry.Validate() != nil || (previous != "" && previous >= entry.Path) { + return errors.New("docker output commit entry sequence is invalid") + } + previous = entry.Path + } + if receipt.CommittedDigestSet != dockerCommittedDigestSet(receipt.Entries) || + receipt.ReceiptFingerprint != dockerOutputCommitReceiptFingerprint(receipt) { + return errors.New("docker output commit receipt aggregate is invalid") + } + return nil +} + +func dockerCommittedDigestSet(entries []DockerOutputCommitEntry) string { + parts := make([]string, 0, len(entries)) + for _, entry := range entries { + parts = append(parts, entry.Path+":"+entry.SHA256) + } + sort.Strings(parts) + return fingerprint(parts...) +} + +func dockerOutputCommitReceiptFingerprint(receipt DockerOutputCommitReceipt) string { + parts := []string{DockerOutputCommitProtocolVersion, receipt.ID, receipt.AttemptID, + strconv.FormatInt(receipt.Generation, 10), receipt.RunID, receipt.WorkspaceID, + receipt.OperationKeyDigest, receipt.RequestFingerprint, receipt.Status, + receipt.CommittedDigestSet, strconv.Itoa(receipt.CommittedCount)} + return fingerprint(parts...) +} + +// VerifyDockerOutputCommit re-reads every accepted file from the staging +// directory and confirms the exact digest and size. It returns the canonical +// committed entries used by the store insert. +func VerifyDockerOutputCommit(stagingRoot string, + request DockerOutputCommitRequest, +) ([]DockerOutputCommitEntry, error) { + if err := request.Validate(); err != nil { + return nil, err + } + info, err := os.Stat(stagingRoot) + if err != nil || !info.IsDir() { + return nil, errors.New("docker output staging root is unavailable") + } + entries := make([]DockerOutputCommitEntry, 0, len(request.AcceptedEntries)) + for _, entry := range request.AcceptedEntries { + full := filepath.Join(stagingRoot, filepath.FromSlash(entry.Path)) + if !workspacePathWithin(full, stagingRoot) || filepath.Clean(full) != full { + return nil, errors.New("docker output commit path escaped the staging root") + } + content, err := os.ReadFile(full) + if err != nil || int64(len(content)) != entry.SizeBytes { + return nil, errors.New("docker output commit file is unavailable") + } + digest := sha256.Sum256(content) + if hex.EncodeToString(digest[:]) != entry.SHA256 { + return nil, errors.New("docker output commit digest does not match staging") + } + entries = append(entries, entry) + } + return entries, nil +} diff --git a/internal/sandbox/docker_container_io_http.go b/internal/sandbox/docker_container_io_http.go new file mode 100644 index 00000000..5308dd5f --- /dev/null +++ b/internal/sandbox/docker_container_io_http.go @@ -0,0 +1,236 @@ +package sandbox + +import ( + "context" + "errors" + "io" + "mime" + "net/http" + "net/url" + "strings" +) + +const ( + DockerContainerIOFailureInvalidResponse = "invalid_response" + DockerContainerIOFailureConnection = "connection_failed" + DockerContainerIOFailureUnavailable = "unavailable" +) + +// DockerContainerIOTransport is the closed daemon surface for the container +// I/O contract: exactly one bounded attach for logs and one archive export of +// the dedicated output mount. Both return raw streams that the domain layer +// bounds and validates; no other verb is reachable. +type DockerContainerIOTransport interface { + Endpoint() DockerObservationEndpoint + AttachLogs(ctx context.Context, plan DockerLogCapturePlan) (io.ReadCloser, error) + ExportOutputs(ctx context.Context, plan DockerOutputExportPlan) (io.ReadCloser, error) +} + +type dockerEngineContainerIOTransport struct { + doer dockerContainerWriteHTTPDoer + endpoint DockerObservationEndpoint +} + +func newDockerEngineContainerIOTransport(doer dockerContainerWriteHTTPDoer, + endpoint DockerObservationEndpoint, +) (dockerEngineContainerIOTransport, error) { + if doer == nil { + return dockerEngineContainerIOTransport{}, errors.New("docker I/O HTTP client is required") + } + if !validDockerContainerLocalEndpoint(endpoint) { + return dockerEngineContainerIOTransport{}, errors.New("docker I/O transport requires a fixed local endpoint") + } + return dockerEngineContainerIOTransport{doer: doer, endpoint: endpoint}, nil +} + +func (transport dockerEngineContainerIOTransport) Endpoint() DockerObservationEndpoint { + return transport.endpoint +} + +// AttachLogs opens the bounded historical log stream. stream=0 means the +// daemon returns the collected logs and closes; live streaming is never used. +func (transport dockerEngineContainerIOTransport) AttachLogs(ctx context.Context, + plan DockerLogCapturePlan, +) (io.ReadCloser, error) { + if err := plan.Validate(); err != nil { + return nil, err + } + path := "/v" + DockerContainerWriteAPIVersion + "/containers/" + + url.PathEscape(plan.ContainerIDFingerprint) + "/attach" + return transport.doStream(ctx, http.MethodPost, path, + "logs=1&stderr=1&stdout=1&stream=0") +} + +// ExportOutputs opens the tar archive of the dedicated output mount. +func (transport dockerEngineContainerIOTransport) ExportOutputs(ctx context.Context, + plan DockerOutputExportPlan, +) (io.ReadCloser, error) { + if err := plan.Validate(); err != nil { + return nil, err + } + path := "/v" + DockerContainerWriteAPIVersion + "/containers/" + + url.PathEscape(plan.ContainerIDFingerprint) + "/archive" + return transport.doStream(ctx, http.MethodGet, path, + "path="+url.QueryEscape(plan.OutputMountTarget)) +} + +func (transport dockerEngineContainerIOTransport) doStream(ctx context.Context, + method, path, rawQuery string, +) (io.ReadCloser, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + if !validDockerContainerIOOperation(method, path, rawQuery) { + return nil, newDockerContainerIOError(DockerContainerIOFailureInvalidResponse) + } + requestURL := "http://docker" + path + if rawQuery != "" { + requestURL += "?" + rawQuery + } + request, err := http.NewRequestWithContext(ctx, method, requestURL, nil) + if err != nil { + return nil, newDockerContainerIOError(DockerContainerIOFailureInvalidResponse) + } + request.Header.Set("User-Agent", "cyberagent-workbench/docker-container-io-v1") + if method == http.MethodGet { + request.Header.Set("Accept", "application/x-tar") + } + response, err := transport.doer.Do(request) + if err != nil { + if ctx.Err() != nil { + return nil, ctx.Err() + } + return nil, newDockerContainerIOError(DockerContainerIOFailureConnection) + } + if response == nil || response.Body == nil { + return nil, newDockerContainerIOError(DockerContainerIOFailureInvalidResponse) + } + if response.Request == nil || response.Request.URL == nil || + response.Request.Method != method || response.Request.URL.Scheme != "http" || + response.Request.URL.Host != "docker" || response.Request.URL.Path != path || + response.Request.URL.RawQuery != rawQuery { + _ = response.Body.Close() + return nil, newDockerContainerIOError(DockerContainerIOFailureInvalidResponse) + } + if response.StatusCode == http.StatusNotFound { + _ = response.Body.Close() + return nil, newDockerContainerIOError(DockerContainerIOFailureUnavailable) + } + if response.StatusCode != http.StatusOK { + _ = response.Body.Close() + return nil, newDockerContainerIOError(DockerContainerIOFailureInvalidResponse) + } + if method == http.MethodGet { + if contentType := response.Header.Get("Content-Type"); contentType != "" { + mediaType, _, parseErr := mime.ParseMediaType(contentType) + if parseErr != nil || !strings.EqualFold(mediaType, "application/x-tar") { + _ = response.Body.Close() + return nil, newDockerContainerIOError(DockerContainerIOFailureInvalidResponse) + } + } + } + return response.Body, nil +} + +// validDockerContainerIOOperation pins the exact attach/archive surface. +// Anything else, including every other query shape, fails closed. +func validDockerContainerIOOperation(method, path, rawQuery string) bool { + prefix := "/v" + DockerContainerWriteAPIVersion + "/containers/" + if !strings.HasPrefix(path, prefix) { + return false + } + tail := strings.TrimPrefix(path, prefix) + switch method { + case http.MethodPost: + if !strings.HasSuffix(tail, "/attach") { + return false + } + reference, err := url.PathUnescape(strings.TrimSuffix(tail, "/attach")) + if err != nil || !validDockerContainerID(reference) || + tail != url.PathEscape(reference)+"/attach" { + return false + } + return rawQuery == "logs=1&stderr=1&stdout=1&stream=0" + case http.MethodGet: + if !strings.HasSuffix(tail, "/archive") { + return false + } + reference, err := url.PathUnescape(strings.TrimSuffix(tail, "/archive")) + if err != nil || !validDockerContainerID(reference) || + tail != url.PathEscape(reference)+"/archive" { + return false + } + values, err := url.ParseQuery(rawQuery) + if err != nil || len(values) != 1 || len(values["path"]) != 1 { + return false + } + target := values.Get("path") + return rawQuery == "path="+url.QueryEscape(target) && + validateVirtualPath("docker archive target", target) == nil + default: + return false + } +} + +type unavailableDockerContainerIOTransport struct { + endpoint DockerObservationEndpoint + reason string +} + +// NewUnavailableDockerContainerIOTransport returns the fixed fail-closed +// transport for environments without the local Docker endpoint. It is the +// only constructor exported from this package for the I/O surface. +func NewUnavailableDockerContainerIOTransport(endpointClass, + reason string, +) DockerContainerIOTransport { + endpoint, err := NewDockerObservationEndpoint(endpointClass) + if err != nil { + endpoint, _ = NewDockerObservationEndpoint(DockerObservationEndpointLocalUnix) + } + return unavailableDockerContainerIOTransport{endpoint: endpoint, reason: reason} +} + +func (transport unavailableDockerContainerIOTransport) Endpoint() DockerObservationEndpoint { + return transport.endpoint +} + +func (transport unavailableDockerContainerIOTransport) AttachLogs(ctx context.Context, + plan DockerLogCapturePlan, +) (io.ReadCloser, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + return nil, newDockerContainerIOError(DockerContainerIOFailureUnavailable) +} + +func (transport unavailableDockerContainerIOTransport) ExportOutputs(ctx context.Context, + plan DockerOutputExportPlan, +) (io.ReadCloser, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + return nil, newDockerContainerIOError(DockerContainerIOFailureUnavailable) +} + +type DockerContainerIOError struct { + code string +} + +func (err *DockerContainerIOError) Error() string { + if err == nil { + return "docker container I/O error" + } + return "docker container I/O " + err.code +} + +func newDockerContainerIOError(code string) error { + return &DockerContainerIOError{code: code} +} + +func DockerContainerIOErrorCode(err error) string { + var target *DockerContainerIOError + if errors.As(err, &target) && target != nil { + return target.code + } + return "" +} diff --git a/internal/sandbox/docker_container_io_http_test.go b/internal/sandbox/docker_container_io_http_test.go new file mode 100644 index 00000000..2a636993 --- /dev/null +++ b/internal/sandbox/docker_container_io_http_test.go @@ -0,0 +1,195 @@ +package sandbox + +import ( + "context" + "io" + "net/http" + "strings" + "testing" + + "cyberagent-workbench/internal/idgen" +) + +type scriptedDockerIOHTTPDoer struct { + status int + contentType string + body string + breakEcho bool + requests []*http.Request +} + +func (doer *scriptedDockerIOHTTPDoer) Do(request *http.Request) (*http.Response, error) { + doer.requests = append(doer.requests, request) + response := &http.Response{ + StatusCode: doer.status, + Body: io.NopCloser(strings.NewReader(doer.body)), + Header: http.Header{}, + } + if doer.contentType != "" { + response.Header.Set("Content-Type", doer.contentType) + } + if !doer.breakEcho { + response.Request = request + } + return response, nil +} + +func testDockerIOTransport(t *testing.T, doer *scriptedDockerIOHTTPDoer) dockerEngineContainerIOTransport { + t.Helper() + endpoint, err := NewDockerObservationEndpoint(DockerObservationEndpointLocalUnix) + if err != nil { + t.Fatal(err) + } + transport, err := newDockerEngineContainerIOTransport(doer, endpoint) + if err != nil { + t.Fatal(err) + } + return transport +} + +func TestDockerContainerIOTransportAttachAndArchiveAllowlist(t *testing.T) { + doer := &scriptedDockerIOHTTPDoer{status: http.StatusOK, body: "raw"} + transport := testDockerIOTransport(t, doer) + plan, err := NewDockerLogCapturePlan(idgen.New("sandbox-docker-attempt"), 1, + idgen.New("sandbox-docker-run"), testDockerContainerFingerprint(), 1024, 64, 60) + if err != nil { + t.Fatal(err) + } + stream, err := transport.AttachLogs(context.Background(), plan) + if err != nil { + t.Fatalf("attach rejected: %v", err) + } + _ = stream.Close() + if len(doer.requests) != 1 || doer.requests[0].Method != http.MethodPost || + doer.requests[0].URL.RawQuery != "logs=1&stderr=1&stdout=1&stream=0" || + doer.requests[0].URL.Host != "docker" { + t.Fatalf("unexpected attach request: %#v", doer.requests[0]) + } + exportPlan, err := NewDockerOutputExportPlan(idgen.New("sandbox-docker-attempt"), 1, + idgen.New("sandbox-docker-run"), testDockerContainerFingerprint(), + "/run/cyberagent/outputs", MaxDockerOutputFiles, MaxDockerOutputFileBytes, + MaxDockerOutputTotalBytes) + if err != nil { + t.Fatal(err) + } + doer.contentType = "application/x-tar" + archive, err := transport.ExportOutputs(context.Background(), exportPlan) + if err != nil { + t.Fatalf("archive rejected: %v", err) + } + _ = archive.Close() + if len(doer.requests) != 2 || doer.requests[1].Method != http.MethodGet || + doer.requests[1].URL.RawQuery != "path=%2Frun%2Fcyberagent%2Foutputs" { + t.Fatalf("unexpected archive request: %#v", doer.requests[1]) + } +} + +func TestDockerContainerIOTransportRejectsEverythingElse(t *testing.T) { + base := "/v" + DockerContainerWriteAPIVersion + "/containers/" + testDockerContainerFingerprint() + tests := []struct { + name string + method string + path string + rawQuery string + }{ + {name: "attach live stream", method: http.MethodPost, path: base + "/attach", + rawQuery: "logs=1&stderr=1&stdout=1&stream=1"}, + {name: "attach missing logs", method: http.MethodPost, path: base + "/attach", + rawQuery: "stderr=1&stdout=1&stream=0"}, + {name: "attach extra parameter", method: http.MethodPost, path: base + "/attach", + rawQuery: "logs=1&stderr=1&stdout=1&stream=0×tamps=1"}, + {name: "archive root target", method: http.MethodGet, path: base + "/archive", + rawQuery: "path=%2F"}, + {name: "archive unclean target", method: http.MethodGet, path: base + "/archive", + rawQuery: "path=%2Frun%2Fcyberagent%2Foutputs%2F"}, + {name: "archive extra parameter", method: http.MethodGet, path: base + "/archive", + rawQuery: "path=%2Frun%2Fcyberagent%2Foutputs&force=1"}, + {name: "archive relative target", method: http.MethodGet, path: base + "/archive", + rawQuery: "path=outputs"}, + {name: "wrong verb", method: http.MethodDelete, path: base, rawQuery: "v=1"}, + {name: "wrong path", method: http.MethodPost, path: base + "/exec", rawQuery: ""}, + {name: "bad reference", method: http.MethodGet, path: "/v" + DockerContainerWriteAPIVersion + "/containers/nope/archive", + rawQuery: "path=%2Frun"}, + } + for _, current := range tests { + t.Run(current.name, func(t *testing.T) { + if validDockerContainerIOOperation(current.method, current.path, current.rawQuery) { + t.Fatal("non-allowlisted operation was accepted") + } + }) + } +} + +func TestDockerContainerIOTransportResponseValidation(t *testing.T) { + tests := []struct { + name string + status int + contentType string + breakEcho bool + wantCode string + }{ + {name: "not found", status: http.StatusNotFound, wantCode: DockerContainerIOFailureUnavailable}, + {name: "server error", status: http.StatusInternalServerError, wantCode: DockerContainerIOFailureInvalidResponse}, + {name: "bad archive content type", status: http.StatusOK, contentType: "text/html", + wantCode: DockerContainerIOFailureInvalidResponse}, + {name: "echo broken", status: http.StatusOK, contentType: "application/x-tar", + breakEcho: true, wantCode: DockerContainerIOFailureInvalidResponse}, + } + for _, current := range tests { + t.Run(current.name, func(t *testing.T) { + doer := &scriptedDockerIOHTTPDoer{status: current.status, + contentType: current.contentType, breakEcho: current.breakEcho, body: "x"} + transport := testDockerIOTransport(t, doer) + plan, err := NewDockerOutputExportPlan(idgen.New("sandbox-docker-attempt"), 1, + idgen.New("sandbox-docker-run"), testDockerContainerFingerprint(), + "/run/cyberagent/outputs", MaxDockerOutputFiles, MaxDockerOutputFileBytes, + MaxDockerOutputTotalBytes) + if err != nil { + t.Fatal(err) + } + _, err = transport.ExportOutputs(context.Background(), plan) + if err == nil || DockerContainerIOErrorCode(err) != current.wantCode { + t.Fatalf("export error = %v, want code %q", err, current.wantCode) + } + }) + } +} + +func TestDockerContainerIOTransportRequiresFixedEndpoint(t *testing.T) { + doer := &scriptedDockerIOHTTPDoer{status: http.StatusOK} + if _, err := newDockerEngineContainerIOTransport(doer, DockerObservationEndpoint{}); err == nil { + t.Fatal("invalid endpoint was accepted") + } + if _, err := newDockerEngineContainerIOTransport(nil, DockerObservationEndpoint{}); err == nil { + t.Fatal("missing doer was accepted") + } + endpoint, err := NewDockerObservationEndpoint(DockerObservationEndpointLocalUnix) + if err != nil { + t.Fatal(err) + } + transport, err := newDockerEngineContainerIOTransport(doer, endpoint) + if err != nil { + t.Fatal(err) + } + if transport.Endpoint() != endpoint { + t.Fatal("endpoint was not preserved") + } +} + +func TestDockerContainerIOTransportCancelPropagates(t *testing.T) { + doer := &scriptedDockerIOHTTPDoer{status: http.StatusOK} + transport := testDockerIOTransport(t, doer) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + plan, err := NewDockerLogCapturePlan(idgen.New("sandbox-docker-attempt"), 1, + idgen.New("sandbox-docker-run"), testDockerContainerFingerprint(), 1024, 64, 60) + if err != nil { + t.Fatal(err) + } + if _, err := transport.AttachLogs(ctx, plan); err == nil { + t.Fatal("cancelled attach succeeded") + } + if len(doer.requests) != 0 { + t.Fatal("cancelled context still sent a request") + } +} diff --git a/internal/sandbox/docker_container_io_input.go b/internal/sandbox/docker_container_io_input.go new file mode 100644 index 00000000..d350b613 --- /dev/null +++ b/internal/sandbox/docker_container_io_input.go @@ -0,0 +1,236 @@ +package sandbox + +import ( + "errors" + "fmt" + "mime" + "path" + "sort" + "strconv" + "strings" + "time" + "unicode/utf8" +) + +const ( + DockerInputProjectionProtocolVersion = "sandbox_docker_input_projection.v1" + MaxDockerInputProjectionEntries = MaxInputArtifacts + MaxDockerInputProjectionFileBytes = 4 * 1024 * 1024 + MaxDockerInputProjectionTotalBytes = MaxCapturedOutputBytes +) + +// DockerInputProjectionEntry is one sealed read-only input file. The path is +// a canonical container-relative path, the digest binds the exact bytes, and +// nothing in the entry grants execution authority. +type DockerInputProjectionEntry struct { + Ordinal int + Path string + SHA256 string + SizeBytes int64 + MediaType string +} + +func (entry DockerInputProjectionEntry) Validate() error { + if entry.Ordinal < 1 || entry.Ordinal > MaxDockerInputProjectionEntries || + validateContainerRelativePath("docker input projection path", entry.Path) != nil || + !validDigest(entry.SHA256) || entry.SizeBytes < 1 || + entry.SizeBytes > MaxDockerInputProjectionFileBytes || + validateBoundedText("docker input media type", entry.MediaType, 256, false) != nil { + return errors.New("docker input projection entry is invalid") + } + if _, _, err := mime.ParseMediaType(entry.MediaType); err != nil { + return errors.New("docker input projection media type is invalid") + } + return nil +} + +// DockerInputProjection seals the read-only input boundary for one container +// lifecycle attempt. It binds the lifecycle generation/attempt identity, the +// exact plan and observation, and a fingerprint over every entry. The daemon +// mount stays read-only and the container cannot modify the original +// Workspace through this boundary. +type DockerInputProjection struct { + ProtocolVersion string + ID string + AttemptID string + Generation int64 + PlanID string + ObservationID string + RunID string + MissionID string + WorkspaceID string + InputArtifactDigest string + SpecFingerprint string + AuthorityFingerprint string + MountTarget string + MountReadOnly bool + Entries []DockerInputProjectionEntry + EntryCount int + TotalBytes int64 + ProjectionFingerprint string + CreatedAt time.Time +} + +func NewDockerInputProjection(id, attemptID string, generation int64, planID, + observationID, runID, missionID, workspaceID, inputArtifactDigest, + specFingerprint, authorityFingerprint, mountTarget string, + entries []DockerInputProjectionEntry, createdAt time.Time, +) (DockerInputProjection, error) { + projection := DockerInputProjection{ + ProtocolVersion: DockerInputProjectionProtocolVersion, ID: id, AttemptID: attemptID, + Generation: generation, PlanID: planID, ObservationID: observationID, RunID: runID, + MissionID: missionID, WorkspaceID: workspaceID, InputArtifactDigest: inputArtifactDigest, + SpecFingerprint: specFingerprint, AuthorityFingerprint: authorityFingerprint, + MountTarget: mountTarget, MountReadOnly: true, + Entries: append([]DockerInputProjectionEntry(nil), entries...), CreatedAt: createdAt, + } + projection.EntryCount = len(projection.Entries) + for _, entry := range projection.Entries { + projection.TotalBytes += entry.SizeBytes + } + projection.ProjectionFingerprint = dockerInputProjectionFingerprint(projection) + return projection, projection.Validate() +} + +func (projection DockerInputProjection) Validate() error { + for label, value := range map[string]string{ + "docker input projection id": projection.ID, + "docker input projection attempt id": projection.AttemptID, + "docker input projection plan id": projection.PlanID, + "docker input projection observation id": projection.ObservationID, + "docker input projection Run id": projection.RunID, + "docker input projection Mission id": projection.MissionID, + "docker input projection workspace id": projection.WorkspaceID, + } { + if validateStoredIdentity(label, value) != nil { + return errors.New("docker input projection identity is invalid") + } + } + if projection.ProtocolVersion != DockerInputProjectionProtocolVersion || + projection.Generation != 1 || !validDigest(projection.InputArtifactDigest) || + !validDigest(projection.SpecFingerprint) || !validDigest(projection.AuthorityFingerprint) || + projection.MountTarget != DockerInputArtifactMountTarget || !projection.MountReadOnly || + projection.CreatedAt.IsZero() { + return errors.New("docker input projection violates a fixed read-only invariant") + } + if len(projection.Entries) == 0 || len(projection.Entries) > MaxDockerInputProjectionEntries || + projection.EntryCount != len(projection.Entries) { + return errors.New("docker input projection entry count is invalid") + } + var total int64 + previousPath := "" + for index, entry := range projection.Entries { + if entry.Ordinal != index+1 || entry.Validate() != nil || + (previousPath != "" && previousPath >= entry.Path) { + return errors.New("docker input projection entry sequence is invalid") + } + previousPath = entry.Path + total += entry.SizeBytes + } + if total < 1 || total > MaxDockerInputProjectionTotalBytes || projection.TotalBytes != total || + projection.ProjectionFingerprint != dockerInputProjectionFingerprint(projection) { + return errors.New("docker input projection aggregate is invalid") + } + return nil +} + +func dockerInputProjectionFingerprint(projection DockerInputProjection) string { + parts := []string{DockerInputProjectionProtocolVersion, projection.ID, projection.AttemptID, + strconv.FormatInt(projection.Generation, 10), projection.PlanID, projection.ObservationID, + projection.RunID, projection.MissionID, projection.WorkspaceID, + projection.InputArtifactDigest, projection.SpecFingerprint, + projection.AuthorityFingerprint, projection.MountTarget, + strconv.FormatBool(projection.MountReadOnly), strconv.Itoa(len(projection.Entries))} + for _, entry := range projection.Entries { + parts = append(parts, strconv.Itoa(entry.Ordinal), entry.Path, entry.SHA256, + strconv.FormatInt(entry.SizeBytes, 10), entry.MediaType) + } + return fingerprint(parts...) +} + +// validateContainerRelativePath rejects absolute paths, traversal, Windows +// separators, drive letters, NUL, control characters, and non-canonical +// forms. The same rules run on every host OS so a Linux daemon and a Windows +// operator share one path boundary. +func validateContainerRelativePath(label, value string) error { + if err := validateBoundedText(label, value, 1024, false); err != nil { + return err + } + if strings.HasPrefix(value, "/") || strings.Contains(value, "\\") || + strings.ContainsRune(value, 0) || path.IsAbs(value) || path.Clean(value) != value || + value == "." || strings.HasPrefix(value, "../") || strings.Contains(value, "/../") || + utf8.RuneCountInString(value) > 512 { + return fmt.Errorf("%s must be a canonical container-relative path", label) + } + if len(value) >= 2 && value[1] == ':' { + return fmt.Errorf("%s must not carry a drive letter", label) + } + for _, component := range strings.Split(value, "/") { + if component == "" || component == "." || component == ".." || + strings.HasSuffix(component, " ") || strings.HasSuffix(component, ".") { + return fmt.Errorf("%s contains an invalid path component", label) + } + } + return nil +} + +// DockerContainerMountState is the bounded projection of one daemon mount +// used to verify the read-only input and dedicated writable output boundary +// after a real container has been created. +type DockerContainerMountState struct { + Target string + Access string +} + +func validDockerContainerMountAccess(value string) bool { + return value == "ro" || value == "rw" +} + +// VerifyDockerContainerMountIsolation checks the real inspection mounts: +// the dedicated output target must be the only writable mount inside the +// output tree, the input target tree must be read-only, and the workspace +// tree must be read-only. Daemon-managed mounts elsewhere stay untouched. +func VerifyDockerContainerMountIsolation(mounts []DockerContainerMountState, + inputTarget, outputTarget, workspaceTarget string, +) error { + if len(mounts) < 2 || len(mounts) > MaxMounts+8 || + validateVirtualPath("docker input mount target", inputTarget) != nil || + validateVirtualPath("docker output mount target", outputTarget) != nil || + validateVirtualPath("docker workspace mount target", workspaceTarget) != nil || + pathWithin(outputTarget, inputTarget) || pathWithin(outputTarget, workspaceTarget) { + return errors.New("docker mount isolation request is invalid") + } + outputFound := false + for _, mount := range mounts { + if validateVirtualPath("docker mount target", mount.Target) != nil || + !validDockerContainerMountAccess(mount.Access) { + return errors.New("docker mount state is invalid") + } + if mount.Target == outputTarget { + if outputFound || mount.Access != "rw" { + return errors.New("docker dedicated output mount is not unique and writable") + } + outputFound = true + continue + } + if pathWithin(mount.Target, outputTarget) { + return errors.New("docker writable mount escaped into the output tree") + } + if pathWithin(mount.Target, inputTarget) || pathWithin(mount.Target, workspaceTarget) { + if mount.Access != "ro" { + return errors.New("docker input or workspace mount is writable") + } + } + } + if !outputFound { + return errors.New("docker dedicated output mount is missing") + } + return nil +} + +// SortedDockerInputProjectionEntries returns a stable copy for receipts. +func SortedDockerInputProjectionEntries(projection DockerInputProjection) []DockerInputProjectionEntry { + entries := append([]DockerInputProjectionEntry(nil), projection.Entries...) + sort.Slice(entries, func(i, j int) bool { return entries[i].Path < entries[j].Path }) + return entries +} diff --git a/internal/sandbox/docker_container_io_log.go b/internal/sandbox/docker_container_io_log.go new file mode 100644 index 00000000..1ac99e7a --- /dev/null +++ b/internal/sandbox/docker_container_io_log.go @@ -0,0 +1,386 @@ +package sandbox + +import ( + "context" + "crypto/sha256" + "encoding/binary" + "encoding/hex" + "errors" + "io" + "strconv" + "strings" + "time" + "unicode/utf8" + + "cyberagent-workbench/internal/redact" +) + +const ( + DockerLogCaptureProtocolVersion = "sandbox_docker_log_capture.v1" + DockerLogCaptureStatusCompleted = "completed" + DockerLogCaptureStatusTruncatedBytes = "truncated_bytes" + DockerLogCaptureStatusTruncatedLines = "truncated_lines" + DockerLogCaptureStatusTruncatedDeadline = "truncated_deadline" + DockerLogCaptureStatusInvalidStream = "invalid_stream" + MaxDockerLogCaptureBytes = 256 * 1024 + MaxDockerLogCaptureLines = 4096 + MaxDockerLogCaptureDurationSeconds = 300 + MaxDockerLogFrameBytes = 256 * 1024 +) + +// DockerLogCapturePlan pins the bounded stdout/stderr capture contract for +// one lifecycle attempt. Raw content never persists: only the bounded, +// redacted in-memory digest set becomes part of the receipt. +type DockerLogCapturePlan struct { + ProtocolVersion string + AttemptID string + Generation int64 + RunID string + ContainerIDFingerprint string + MaxBytes int64 + MaxLines int + DurationSeconds int + RedactSecrets bool + RequireUTF8 bool + CaptureFingerprint string +} + +func NewDockerLogCapturePlan(attemptID string, generation int64, runID, + containerIDFingerprint string, maxBytes int64, maxLines, durationSeconds int, +) (DockerLogCapturePlan, error) { + plan := DockerLogCapturePlan{ + ProtocolVersion: DockerLogCaptureProtocolVersion, AttemptID: attemptID, + Generation: generation, RunID: runID, ContainerIDFingerprint: containerIDFingerprint, + MaxBytes: maxBytes, MaxLines: maxLines, DurationSeconds: durationSeconds, + RedactSecrets: true, RequireUTF8: true, + } + plan.CaptureFingerprint = dockerLogCapturePlanFingerprint(plan) + return plan, plan.Validate() +} + +func (plan DockerLogCapturePlan) Validate() error { + for label, value := range map[string]string{ + "docker log capture attempt id": plan.AttemptID, + "docker log capture Run id": plan.RunID, + } { + if validateStoredIdentity(label, value) != nil { + return errors.New("docker log capture identity is invalid") + } + } + if plan.ProtocolVersion != DockerLogCaptureProtocolVersion || plan.Generation != 1 || + !validDigest(plan.ContainerIDFingerprint) || + plan.MaxBytes < 1 || plan.MaxBytes > MaxDockerLogCaptureBytes || + plan.MaxLines < 1 || plan.MaxLines > MaxDockerLogCaptureLines || + plan.DurationSeconds < 1 || plan.DurationSeconds > MaxDockerLogCaptureDurationSeconds || + !plan.RedactSecrets || !plan.RequireUTF8 || + plan.CaptureFingerprint != dockerLogCapturePlanFingerprint(plan) { + return errors.New("docker log capture plan violates a fixed bound") + } + return nil +} + +func dockerLogCapturePlanFingerprint(plan DockerLogCapturePlan) string { + return fingerprint(DockerLogCaptureProtocolVersion, plan.AttemptID, + strconv.FormatInt(plan.Generation, 10), plan.RunID, plan.ContainerIDFingerprint, + strconv.FormatInt(plan.MaxBytes, 10), strconv.Itoa(plan.MaxLines), + strconv.Itoa(plan.DurationSeconds), strconv.FormatBool(plan.RedactSecrets), + strconv.FormatBool(plan.RequireUTF8)) +} + +// DockerLogStreamRecord is the bounded, digest-only projection of one stream. +// It never carries raw content. +type DockerLogStreamRecord struct { + Stream string + ByteCount int64 + LineCount int + TruncatedBytes bool + TruncatedLines bool + TruncatedDeadline bool + UTF8Violations int + RedactedSegments int + ContentDigest string +} + +func (record DockerLogStreamRecord) Validate() error { + if record.Stream != "stdout" && record.Stream != "stderr" || + record.ByteCount < 0 || record.LineCount < 0 || record.UTF8Violations < 0 || + record.RedactedSegments < 0 || !validDigest(record.ContentDigest) { + return errors.New("docker log stream record is invalid") + } + if record.TruncatedBytes && record.TruncatedLines { + return errors.New("docker log stream cannot truncate on two bounds") + } + return nil +} + +// DockerLogCaptureReceipt is the durable, content-free evidence of one +// bounded capture. Raw stdout/stderr bytes never enter SQLite or events. +type DockerLogCaptureReceipt struct { + ProtocolVersion string + ID string + AttemptID string + Generation int64 + RunID string + ContainerIDFingerprint string + Status string + Streams []DockerLogStreamRecord + StreamCount int + TotalBytes int64 + TotalLines int + UTF8Violations int + RedactedSegments int + CaptureMaxBytes int64 + CaptureMaxLines int + CaptureFingerprint string + ReceiptFingerprint string + CreatedAt time.Time +} + +func NewDockerLogCaptureReceipt(id, attemptID string, generation int64, runID, + containerIDFingerprint string, plan DockerLogCapturePlan, + streams []DockerLogStreamRecord, status string, createdAt time.Time, +) (DockerLogCaptureReceipt, error) { + receipt := DockerLogCaptureReceipt{ + ProtocolVersion: DockerLogCaptureProtocolVersion, ID: id, AttemptID: attemptID, + Generation: generation, RunID: runID, ContainerIDFingerprint: containerIDFingerprint, + Status: status, Streams: append([]DockerLogStreamRecord(nil), streams...), + CaptureMaxBytes: plan.MaxBytes, CaptureMaxLines: plan.MaxLines, + CaptureFingerprint: plan.CaptureFingerprint, CreatedAt: createdAt, + } + receipt.StreamCount = len(receipt.Streams) + for _, stream := range receipt.Streams { + receipt.TotalBytes += stream.ByteCount + receipt.TotalLines += stream.LineCount + receipt.UTF8Violations += stream.UTF8Violations + receipt.RedactedSegments += stream.RedactedSegments + } + receipt.ReceiptFingerprint = dockerLogCaptureReceiptFingerprint(receipt) + return receipt, receipt.Validate() +} + +func (receipt DockerLogCaptureReceipt) Validate() error { + for label, value := range map[string]string{ + "docker log receipt id": receipt.ID, "docker log receipt attempt id": receipt.AttemptID, + "docker log receipt Run id": receipt.RunID, + } { + if validateStoredIdentity(label, value) != nil { + return errors.New("docker log capture receipt identity is invalid") + } + } + if receipt.ProtocolVersion != DockerLogCaptureProtocolVersion || receipt.Generation != 1 || + !validDigest(receipt.ContainerIDFingerprint) || !validDigest(receipt.CaptureFingerprint) || + !validDockerLogCaptureStatus(receipt.Status) || receipt.CreatedAt.IsZero() || + receipt.StreamCount != len(receipt.Streams) || len(receipt.Streams) != 2 { + return errors.New("docker log capture receipt is invalid") + } + seen := map[string]bool{} + var totalBytes int64 + totalLines := 0 + truncated := false + for index, stream := range receipt.Streams { + if index == 0 && stream.Stream != "stdout" || index == 1 && stream.Stream != "stderr" || + seen[stream.Stream] || stream.Validate() != nil || + stream.ByteCount > receipt.CaptureMaxBytes || stream.LineCount > receipt.CaptureMaxLines { + return errors.New("docker log capture stream sequence is invalid") + } + seen[stream.Stream] = true + totalBytes += stream.ByteCount + totalLines += stream.LineCount + if stream.TruncatedBytes && stream.ByteCount != receipt.CaptureMaxBytes || + stream.TruncatedLines && stream.LineCount != receipt.CaptureMaxLines { + return errors.New("docker log truncation flag does not bind its cap") + } + truncated = truncated || stream.TruncatedBytes || stream.TruncatedLines || stream.TruncatedDeadline + } + if receipt.CaptureMaxBytes < 1 || receipt.CaptureMaxBytes > MaxDockerLogCaptureBytes || + receipt.CaptureMaxLines < 1 || receipt.CaptureMaxLines > MaxDockerLogCaptureLines || + receipt.TotalBytes != totalBytes || receipt.TotalLines != totalLines || + receipt.UTF8Violations != receipt.Streams[0].UTF8Violations+receipt.Streams[1].UTF8Violations || + receipt.RedactedSegments != receipt.Streams[0].RedactedSegments+receipt.Streams[1].RedactedSegments || + receipt.ReceiptFingerprint != dockerLogCaptureReceiptFingerprint(receipt) { + return errors.New("docker log capture receipt aggregate is invalid") + } + switch receipt.Status { + case DockerLogCaptureStatusCompleted: + if truncated { + return errors.New("completed docker log receipt carries a truncation flag") + } + case DockerLogCaptureStatusInvalidStream, DockerLogCaptureStatusTruncatedDeadline: + default: + if !truncated { + return errors.New("truncated docker log receipt misses its truncation flag") + } + } + return nil +} + +func validDockerLogCaptureStatus(status string) bool { + switch status { + case DockerLogCaptureStatusCompleted, DockerLogCaptureStatusTruncatedBytes, + DockerLogCaptureStatusTruncatedLines, DockerLogCaptureStatusTruncatedDeadline, + DockerLogCaptureStatusInvalidStream: + return true + default: + return false + } +} + +func dockerLogCaptureReceiptFingerprint(receipt DockerLogCaptureReceipt) string { + parts := []string{DockerLogCaptureProtocolVersion, receipt.ID, receipt.AttemptID, + strconv.FormatInt(receipt.Generation, 10), receipt.RunID, + receipt.ContainerIDFingerprint, receipt.Status, receipt.CaptureFingerprint, + strconv.FormatInt(receipt.CaptureMaxBytes, 10), strconv.Itoa(receipt.CaptureMaxLines), + strconv.Itoa(receipt.StreamCount)} + for _, stream := range receipt.Streams { + parts = append(parts, stream.Stream, strconv.FormatInt(stream.ByteCount, 10), + strconv.Itoa(stream.LineCount), strconv.FormatBool(stream.TruncatedBytes), + strconv.FormatBool(stream.TruncatedLines), strconv.FormatBool(stream.TruncatedDeadline), + strconv.Itoa(stream.UTF8Violations), strconv.Itoa(stream.RedactedSegments), + stream.ContentDigest) + } + return fingerprint(parts...) +} + +type dockerLogStreamAccumulator struct { + record DockerLogStreamRecord + pending []byte + budgetBytes int64 + budgetLines int + content strings.Builder +} + +// DecodeDockerLogFrames consumes the raw Docker attach multiplexed stream +// (8-byte header: stream type, three zero bytes, big-endian size) under the +// plan bounds. It returns bounded, redacted, digest-only stream records plus +// a status. Raw bytes stay in memory and are never persisted. +func DecodeDockerLogFrames(ctx context.Context, plan DockerLogCapturePlan, + src io.Reader, +) ([]DockerLogStreamRecord, string, error) { + if err := plan.Validate(); err != nil { + return nil, "", err + } + if src == nil { + return nil, "", errors.New("docker log source is required") + } + streams := [2]*dockerLogStreamAccumulator{ + {record: DockerLogStreamRecord{Stream: "stdout"}, budgetBytes: plan.MaxBytes, budgetLines: plan.MaxLines}, + {record: DockerLogStreamRecord{Stream: "stderr"}, budgetBytes: plan.MaxBytes, budgetLines: plan.MaxLines}, + } + status := DockerLogCaptureStatusCompleted + for { + if err := ctx.Err(); err != nil { + streams[0].record.TruncatedDeadline = true + streams[1].record.TruncatedDeadline = true + status = DockerLogCaptureStatusTruncatedDeadline + break + } + header := make([]byte, 8) + _, err := io.ReadFull(src, header) + if err != nil { + if err == io.EOF || err == io.ErrUnexpectedEOF { + break + } + return nil, "", err + } + if header[1] != 0 || header[2] != 0 || header[3] != 0 || + (header[0] != 1 && header[0] != 2) { + status = DockerLogCaptureStatusInvalidStream + break + } + size := binary.BigEndian.Uint32(header[4:8]) + if size == 0 { + continue + } + if size > MaxDockerLogFrameBytes { + status = DockerLogCaptureStatusInvalidStream + break + } + accumulator := streams[header[0]-1] + if accumulator.record.TruncatedBytes || accumulator.record.TruncatedLines { + if _, err := io.CopyN(io.Discard, src, int64(size)); err != nil { + return nil, "", err + } + continue + } + remaining := accumulator.budgetBytes - accumulator.record.ByteCount + readBytes := int64(size) + if readBytes > remaining { + readBytes = remaining + } + buffer := make([]byte, readBytes) + if _, err := io.ReadFull(src, buffer); err != nil { + return nil, "", err + } + if int64(size) > readBytes { + if _, err := io.CopyN(io.Discard, src, int64(size)-readBytes); err != nil { + return nil, "", err + } + } + accumulator.record.ByteCount += int64(len(buffer)) + decoded, pending, violations := decodeDockerLogUTF8(accumulator.pending, buffer) + accumulator.pending = pending + accumulator.record.UTF8Violations += violations + accumulator.record.LineCount += strings.Count(decoded, "\n") + accumulator.content.WriteString(decoded) + if accumulator.record.ByteCount >= accumulator.budgetBytes { + accumulator.record.TruncatedBytes = true + } else if accumulator.record.LineCount >= accumulator.budgetLines { + accumulator.record.TruncatedLines = true + // The crossing frame may overshoot the line cap; the receipt binds + // the exact capped count so the digest stays deterministic. + accumulator.record.LineCount = accumulator.budgetLines + } + } + records := make([]DockerLogStreamRecord, 2) + for index, accumulator := range streams { + record := accumulator.record + content := accumulator.content.String() + if record.TruncatedLines { + content = strings.Join(strings.SplitN(content, "\n", accumulator.budgetLines+1)[:accumulator.budgetLines], "\n") + } + redacted := redact.String(content) + if redacted != content { + record.RedactedSegments = strings.Count(redacted, "[REDACTED:") + } + record.ContentDigest = hashDockerLogContent(redacted) + records[index] = record + } + if status == DockerLogCaptureStatusCompleted { + switch { + case streams[0].record.TruncatedBytes || streams[1].record.TruncatedBytes: + status = DockerLogCaptureStatusTruncatedBytes + case streams[0].record.TruncatedLines || streams[1].record.TruncatedLines: + status = DockerLogCaptureStatusTruncatedLines + } + } + return records, status, nil +} + +// decodeDockerLogUTF8 decodes chunk on top of a pending incomplete-run tail, +// counts and replaces invalid sequences with U+FFFD, and returns the text, +// the new tail, and the violation count. +func decodeDockerLogUTF8(pending, chunk []byte) (string, []byte, int) { + data := append(pending, chunk...) + var builder strings.Builder + violations := 0 + for len(data) > 0 { + current, size := utf8.DecodeRune(data) + if current == utf8.RuneError && size == 1 { + if len(data) < utf8.UTFMax && !utf8.FullRune(data) { + break + } + violations++ + builder.WriteRune(utf8.RuneError) + data = data[1:] + continue + } + builder.WriteRune(current) + data = data[size:] + } + return builder.String(), append([]byte(nil), data...), violations +} + +func hashDockerLogContent(content string) string { + digest := sha256.Sum256([]byte(content)) + return hex.EncodeToString(digest[:]) +} diff --git a/internal/sandbox/docker_container_io_test.go b/internal/sandbox/docker_container_io_test.go new file mode 100644 index 00000000..aa5a55ba --- /dev/null +++ b/internal/sandbox/docker_container_io_test.go @@ -0,0 +1,507 @@ +package sandbox + +import ( + "archive/tar" + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "io" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "cyberagent-workbench/internal/idgen" +) + +func testDigest(value string) string { + sum := sha256.Sum256([]byte(value)) + return hex.EncodeToString(sum[:]) +} + +func testDockerContainerFingerprint() string { + return testDigest("container") +} + +func testDockerInputProjection(t *testing.T, entries ...DockerInputProjectionEntry) DockerInputProjection { + t.Helper() + projection, err := NewDockerInputProjection(idgen.New("sandbox-docker-input"), + idgen.New("sandbox-docker-attempt"), 1, idgen.New("sandbox-docker-plan"), + idgen.New("sandbox-docker-observation"), idgen.New("sandbox-docker-run"), + idgen.New("sandbox-docker-mission"), idgen.New("sandbox-docker-workspace"), + testDigest("input-artifacts"), testDigest("spec"), testDigest("authority"), + DockerInputArtifactMountTarget, entries, time.Now().UTC()) + if err != nil { + t.Fatal(err) + } + return projection +} + +func TestDockerInputProjectionRejectsEscapeAndTampering(t *testing.T) { + valid := []DockerInputProjectionEntry{ + {Ordinal: 1, Path: "data/input.json", SHA256: testDigest("one"), + SizeBytes: 4, MediaType: "application/json"}, + {Ordinal: 2, Path: "data/notes.txt", SHA256: testDigest("two"), + SizeBytes: 3, MediaType: "text/plain"}, + } + projection := testDockerInputProjection(t, valid...) + for name, mutate := range map[string]func(*DockerInputProjection){ + "absolute": func(p *DockerInputProjection) { p.Entries[0].Path = "/etc/passwd" }, + "traversal": func(p *DockerInputProjection) { p.Entries[0].Path = "../secret" }, + "windows separator": func(p *DockerInputProjection) { p.Entries[0].Path = "a\\b" }, + "drive letter": func(p *DockerInputProjection) { p.Entries[0].Path = "C:evil" }, + "duplicate path": func(p *DockerInputProjection) { p.Entries[1].Path = p.Entries[0].Path }, + "bad digest": func(p *DockerInputProjection) { p.Entries[0].SHA256 = strings.Repeat("g", 64) }, + "empty media": func(p *DockerInputProjection) { p.Entries[0].MediaType = "" }, + "mount writable": func(p *DockerInputProjection) { p.MountReadOnly = false }, + "wrong mount": func(p *DockerInputProjection) { p.MountTarget = "/tmp" }, + "generation": func(p *DockerInputProjection) { p.Generation = 2 }, + "total drift": func(p *DockerInputProjection) { p.TotalBytes = 1 }, + } { + t.Run(name, func(t *testing.T) { + candidate := projection + candidate.Entries = append([]DockerInputProjectionEntry(nil), projection.Entries...) + mutate(&candidate) + candidate.ProjectionFingerprint = dockerInputProjectionFingerprint(candidate) + if err := candidate.Validate(); err == nil { + t.Fatal("invalid projection was accepted") + } + }) + } + if err := projection.Validate(); err != nil { + t.Fatalf("valid projection rejected: %v", err) + } +} + +func TestVerifyDockerContainerMountIsolation(t *testing.T) { + input := "/run/cyberagent/inputs" + output := "/run/cyberagent/outputs" + workspace := "/workspace" + tests := []struct { + name string + mounts []DockerContainerMountState + wantErr bool + }{ + {name: "isolated", mounts: []DockerContainerMountState{ + {Target: workspace, Access: "ro"}, + {Target: input, Access: "ro"}, + {Target: output, Access: "rw"}, + {Target: "/etc/hosts", Access: "rw"}, + }, wantErr: false}, + {name: "missing output", mounts: []DockerContainerMountState{ + {Target: workspace, Access: "ro"}, + }, wantErr: true}, + {name: "output read only", mounts: []DockerContainerMountState{ + {Target: workspace, Access: "ro"}, + {Target: output, Access: "ro"}, + }, wantErr: true}, + {name: "input writable", mounts: []DockerContainerMountState{ + {Target: workspace, Access: "ro"}, + {Target: input, Access: "rw"}, + {Target: output, Access: "rw"}, + }, wantErr: true}, + {name: "workspace writable", mounts: []DockerContainerMountState{ + {Target: workspace, Access: "rw"}, + {Target: output, Access: "rw"}, + }, wantErr: true}, + {name: "escaped writable inside output tree", mounts: []DockerContainerMountState{ + {Target: workspace, Access: "ro"}, + {Target: output, Access: "rw"}, + {Target: output + "/nested", Access: "rw"}, + }, wantErr: true}, + {name: "duplicate output", mounts: []DockerContainerMountState{ + {Target: workspace, Access: "ro"}, + {Target: output, Access: "rw"}, + {Target: output, Access: "rw"}, + }, wantErr: true}, + {name: "invalid access", mounts: []DockerContainerMountState{ + {Target: workspace, Access: "ro"}, + {Target: output, Access: "shared"}, + }, wantErr: true}, + } + for _, current := range tests { + t.Run(current.name, func(t *testing.T) { + err := VerifyDockerContainerMountIsolation(current.mounts, input, output, workspace) + if (err != nil) != current.wantErr { + t.Fatalf("mount isolation error = %v, wantErr %t", err, current.wantErr) + } + }) + } +} + +func dockerLogFrame(stream byte, payload []byte) []byte { + frame := make([]byte, 8+len(payload)) + frame[0] = stream + frame[4] = byte(len(payload) >> 24) + frame[5] = byte(len(payload) >> 16) + frame[6] = byte(len(payload) >> 8) + frame[7] = byte(len(payload)) + copy(frame[8:], payload) + return frame +} + +func testDockerLogCapturePlan(t *testing.T, maxBytes int64, maxLines int) DockerLogCapturePlan { + t.Helper() + plan, err := NewDockerLogCapturePlan(idgen.New("sandbox-docker-attempt"), 1, + idgen.New("sandbox-docker-run"), testDockerContainerFingerprint(), maxBytes, maxLines, 60) + if err != nil { + t.Fatal(err) + } + return plan +} + +func TestDecodeDockerLogFramesGoldenVectors(t *testing.T) { + t.Run("completed", func(t *testing.T) { + plan := testDockerLogCapturePlan(t, 1024, 64) + stream := append(dockerLogFrame(1, []byte("hello\n")), + dockerLogFrame(2, []byte("oops\n"))...) + records, status, err := DecodeDockerLogFrames(context.Background(), plan, bytes.NewReader(stream)) + if err != nil || status != DockerLogCaptureStatusCompleted { + t.Fatalf("decode = %v, %q", err, status) + } + if records[0].ByteCount != 6 || records[0].LineCount != 1 || + records[1].ByteCount != 5 || records[1].LineCount != 1 { + t.Fatalf("unexpected records: %#v", records) + } + }) + t.Run("byte truncation", func(t *testing.T) { + plan := testDockerLogCapturePlan(t, 8, 64) + stream := append(dockerLogFrame(1, []byte("0123456789")), + dockerLogFrame(2, []byte("later"))...) + records, status, err := DecodeDockerLogFrames(context.Background(), plan, bytes.NewReader(stream)) + if err != nil || status != DockerLogCaptureStatusTruncatedBytes { + t.Fatalf("decode = %v, %q", err, status) + } + if !records[0].TruncatedBytes || records[0].ByteCount != 8 || records[1].ByteCount != 5 { + t.Fatalf("unexpected truncation: %#v", records) + } + }) + t.Run("line truncation clamps count", func(t *testing.T) { + plan := testDockerLogCapturePlan(t, 1024, 3) + stream := append(dockerLogFrame(1, []byte("a\nb\nc\nd\ne\n")), + dockerLogFrame(2, []byte("x\n"))...) + records, status, err := DecodeDockerLogFrames(context.Background(), plan, bytes.NewReader(stream)) + if err != nil || status != DockerLogCaptureStatusTruncatedLines { + t.Fatalf("decode = %v, %q", err, status) + } + if !records[0].TruncatedLines || records[0].LineCount != 3 { + t.Fatalf("unexpected line truncation: %#v", records) + } + }) + t.Run("invalid stream type", func(t *testing.T) { + plan := testDockerLogCapturePlan(t, 1024, 64) + records, status, err := DecodeDockerLogFrames(context.Background(), plan, + bytes.NewReader(dockerLogFrame(3, []byte("x")))) + if err != nil || status != DockerLogCaptureStatusInvalidStream { + t.Fatalf("decode = %v, %q", err, status) + } + _ = records + }) + t.Run("oversized frame", func(t *testing.T) { + plan := testDockerLogCapturePlan(t, 1024, 64) + stream := dockerLogFrame(1, make([]byte, MaxDockerLogFrameBytes+1)) + records, status, err := DecodeDockerLogFrames(context.Background(), plan, bytes.NewReader(stream)) + if err != nil || status != DockerLogCaptureStatusInvalidStream { + t.Fatalf("decode = %v, %q", err, status) + } + _ = records + }) + t.Run("utf8 violations and redaction", func(t *testing.T) { + plan := testDockerLogCapturePlan(t, 4096, 64) + stream := dockerLogFrame(1, []byte("token sk-abcdefghijklmnopqrstuvwxyz012345 \xff\xfe plain\n")) + records, status, err := DecodeDockerLogFrames(context.Background(), plan, bytes.NewReader(stream)) + if err != nil || status != DockerLogCaptureStatusCompleted { + t.Fatalf("decode = %v, %q", err, status) + } + if records[0].UTF8Violations != 2 { + t.Fatalf("utf8 violations = %d, want 2", records[0].UTF8Violations) + } + if records[0].RedactedSegments < 1 { + t.Fatalf("secret was not redacted: %#v", records[0]) + } + }) + t.Run("deadline", func(t *testing.T) { + plan := testDockerLogCapturePlan(t, 4096, 64) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + records, status, err := DecodeDockerLogFrames(ctx, plan, bytes.NewReader(nil)) + if err != nil || status != DockerLogCaptureStatusTruncatedDeadline { + t.Fatalf("decode = %v, %q", err, status) + } + if !records[0].TruncatedDeadline || !records[1].TruncatedDeadline { + t.Fatalf("deadline flags missing: %#v", records) + } + }) +} + +func TestDockerLogCaptureReceiptRejectsTampering(t *testing.T) { + plan := testDockerLogCapturePlan(t, 1024, 64) + records := []DockerLogStreamRecord{ + {Stream: "stdout", ByteCount: 2, LineCount: 1, ContentDigest: testDigest("a")}, + {Stream: "stderr", ContentDigest: testDigest("")}, + } + receipt, err := NewDockerLogCaptureReceipt(idgen.New("sandbox-docker-log"), + plan.AttemptID, 1, plan.RunID, plan.ContainerIDFingerprint, plan, records, + DockerLogCaptureStatusCompleted, time.Now().UTC()) + if err != nil { + t.Fatal(err) + } + for name, mutate := range map[string]func(*DockerLogCaptureReceipt){ + "status mismatch": func(r *DockerLogCaptureReceipt) { r.Status = DockerLogCaptureStatusTruncatedBytes }, + "stream order": func(r *DockerLogCaptureReceipt) { r.Streams[0], r.Streams[1] = r.Streams[1], r.Streams[0] }, + "flag without cap": func(r *DockerLogCaptureReceipt) { r.Streams[0].TruncatedBytes = true }, + "bytes over plan cap": func(r *DockerLogCaptureReceipt) { r.Streams[0].ByteCount = r.CaptureMaxBytes + 1 }, + "stream count drift": func(r *DockerLogCaptureReceipt) { r.StreamCount = 3 }, + } { + t.Run(name, func(t *testing.T) { + candidate := receipt + candidate.Streams = append([]DockerLogStreamRecord(nil), receipt.Streams...) + mutate(&candidate) + candidate.ReceiptFingerprint = dockerLogCaptureReceiptFingerprint(candidate) + if err := candidate.Validate(); err == nil { + t.Fatal("tampered receipt was accepted") + } + }) + } +} + +func buildDockerOutputTar(t *testing.T, files map[string]string) []byte { + t.Helper() + var buffer bytes.Buffer + writer := tar.NewWriter(&buffer) + names := make([]string, 0, len(files)) + for name := range files { + names = append(names, name) + } + sortStrings(names) + for _, name := range names { + if err := writer.WriteHeader(&tar.Header{Name: name, Mode: 0o644, + Size: int64(len(files[name])), Typeflag: tar.TypeReg}); err != nil { + t.Fatal(err) + } + if _, err := writer.Write([]byte(files[name])); err != nil { + t.Fatal(err) + } + } + if err := writer.Close(); err != nil { + t.Fatal(err) + } + return buffer.Bytes() +} + +func sortStrings(values []string) { + for i := 1; i < len(values); i++ { + for j := i; j > 0 && values[j] < values[j-1]; j-- { + values[j], values[j-1] = values[j-1], values[j] + } + } +} + +func testDockerOutputExportPlan(t *testing.T) DockerOutputExportPlan { + t.Helper() + plan, err := NewDockerOutputExportPlan(idgen.New("sandbox-docker-attempt"), 1, + idgen.New("sandbox-docker-run"), testDockerContainerFingerprint(), + "/run/cyberagent/outputs", MaxDockerOutputFiles, MaxDockerOutputFileBytes, + MaxDockerOutputTotalBytes) + if err != nil { + t.Fatal(err) + } + return plan +} + +func TestStageDockerOutputArchiveBoundary(t *testing.T) { + t.Run("completed with directories and redaction", func(t *testing.T) { + plan := testDockerOutputExportPlan(t) + archive := buildDockerOutputTar(t, map[string]string{ + "report/summary.json": "{\"ok\": true}\n", + "token.txt": "token sk-abcdefghijklmnopqrstuvwxyz012345\n", + }) + staging := t.TempDir() + receipt, err := StageDockerOutputArchive(context.Background(), plan, + bytes.NewReader(archive), staging, idgen.New("sandbox-docker-staging"), + time.Now().UTC()) + if err != nil || receipt.Status != DockerOutputStagingStatusCompleted { + t.Fatalf("stage = %v, %q", err, receipt.Status) + } + if receipt.FileCount != 2 || receipt.RedactedCount != 1 || receipt.TotalBytes == 0 { + t.Fatalf("unexpected receipt: %#v", receipt) + } + staged, err := os.ReadFile(filepath.Join(staging, "report", "summary.json")) + if err != nil || !strings.Contains(string(staged), "ok") { + t.Fatalf("staged file missing: %v", err) + } + redactedFile, err := os.ReadFile(filepath.Join(staging, "token.txt")) + if err != nil || strings.Contains(string(redactedFile), "sk-abcdefghijklmnopqrstuvwxyz012345") { + t.Fatal("secret survived staging redaction") + } + }) + t.Run("path traversal rejected", func(t *testing.T) { + plan := testDockerOutputExportPlan(t) + var buffer bytes.Buffer + writer := tar.NewWriter(&buffer) + _ = writer.WriteHeader(&tar.Header{Name: "../escape", Mode: 0o644, + Size: 3, Typeflag: tar.TypeReg}) + _, _ = writer.Write([]byte("x")) + _ = writer.Close() + receipt, err := StageDockerOutputArchive(context.Background(), plan, + bytes.NewReader(buffer.Bytes()), t.TempDir(), idgen.New("sandbox-docker-staging"), + time.Now().UTC()) + if err != nil || receipt.Status != DockerOutputStagingStatusPath { + t.Fatalf("stage = %v, %q", err, receipt.Status) + } + }) + t.Run("symlink rejected", func(t *testing.T) { + plan := testDockerOutputExportPlan(t) + var buffer bytes.Buffer + writer := tar.NewWriter(&buffer) + _ = writer.WriteHeader(&tar.Header{Name: "link", Mode: 0o777, + Linkname: "/etc/passwd", Typeflag: tar.TypeSymlink}) + _ = writer.Close() + receipt, err := StageDockerOutputArchive(context.Background(), plan, + bytes.NewReader(buffer.Bytes()), t.TempDir(), idgen.New("sandbox-docker-staging"), + time.Now().UTC()) + if err != nil || receipt.Status != DockerOutputStagingStatusLink { + t.Fatalf("stage = %v, %q", err, receipt.Status) + } + }) + t.Run("duplicate rejected", func(t *testing.T) { + plan := testDockerOutputExportPlan(t) + var buffer bytes.Buffer + writer := tar.NewWriter(&buffer) + _ = writer.WriteHeader(&tar.Header{Name: "dup.txt", Mode: 0o644, Size: 1, Typeflag: tar.TypeReg}) + _, _ = writer.Write([]byte("a")) + _ = writer.WriteHeader(&tar.Header{Name: "dup.txt", Mode: 0o644, Size: 1, Typeflag: tar.TypeReg}) + _, _ = writer.Write([]byte("b")) + _ = writer.Close() + receipt, err := StageDockerOutputArchive(context.Background(), plan, + bytes.NewReader(buffer.Bytes()), t.TempDir(), idgen.New("sandbox-docker-staging"), + time.Now().UTC()) + if err != nil || receipt.Status != DockerOutputStagingStatusDuplicate { + t.Fatalf("stage = %v, %q", err, receipt.Status) + } + }) + t.Run("total size truncation", func(t *testing.T) { + plan, err := NewDockerOutputExportPlan(idgen.New("sandbox-docker-attempt"), 1, + idgen.New("sandbox-docker-run"), testDockerContainerFingerprint(), + "/run/cyberagent/outputs", 8, 64, 128) + if err != nil { + t.Fatal(err) + } + archive := buildDockerOutputTar(t, map[string]string{"a.txt": strings.Repeat("a", 100)}) + receipt, err := StageDockerOutputArchive(context.Background(), plan, + bytes.NewReader(archive), t.TempDir(), idgen.New("sandbox-docker-staging"), + time.Now().UTC()) + if err != nil || receipt.Status != DockerOutputStagingStatusTruncated { + t.Fatalf("stage = %v, %q", err, receipt.Status) + } + }) +} + +func TestDockerOutputCommitBindsAndVerifiesStaging(t *testing.T) { + plan := testDockerOutputExportPlan(t) + archive := buildDockerOutputTar(t, map[string]string{"report/result.json": "{\"ok\": true}\n"}) + stagingRoot := t.TempDir() + staging, err := StageDockerOutputArchive(context.Background(), plan, + bytes.NewReader(archive), stagingRoot, idgen.New("sandbox-docker-staging"), + time.Now().UTC()) + if err != nil { + t.Fatal(err) + } + entry := DockerOutputCommitEntry{ + Path: staging.Entries[0].Path, SHA256: staging.Entries[0].SHA256, + SizeBytes: staging.Entries[0].SizeBytes, MediaType: staging.Entries[0].MediaType, + } + request, err := NewDockerOutputCommitRequest(staging.AttemptID, 1, staging.RunID, + idgen.New("sandbox-docker-workspace"), staging.ID, testDigest("operation"), + []DockerOutputCommitEntry{entry}) + if err != nil { + t.Fatal(err) + } + if err := request.Binds(staging); err != nil { + t.Fatalf("accepted manifest should bind: %v", err) + } + verified, err := VerifyDockerOutputCommit(stagingRoot, request) + if err != nil || len(verified) != 1 || verified[0].SHA256 != entry.SHA256 { + t.Fatalf("commit verification = %#v, %v", verified, err) + } + for name, mutate := range map[string]func(*DockerOutputCommitRequest){ + "un-staged entry": func(r *DockerOutputCommitRequest) { + r.AcceptedEntries[0].Path = "missing.txt" + r.RequestFingerprint = DockerOutputCommitRequestFingerprint(*r) + }, + "digest drift": func(r *DockerOutputCommitRequest) { + r.AcceptedEntries[0].SHA256 = testDigest("drift") + r.RequestFingerprint = DockerOutputCommitRequestFingerprint(*r) + }, + "empty manifest": func(r *DockerOutputCommitRequest) { + r.AcceptedEntries = nil + r.RequestFingerprint = DockerOutputCommitRequestFingerprint(*r) + }, + } { + t.Run(name, func(t *testing.T) { + candidate := request + candidate.AcceptedEntries = append([]DockerOutputCommitEntry(nil), request.AcceptedEntries...) + mutate(&candidate) + if err := candidate.Binds(staging); err == nil { + t.Fatal("invalid manifest bound to staging") + } + }) + } + stagedFile := filepath.Join(stagingRoot, filepath.FromSlash(entry.Path)) + if err := os.Remove(stagedFile); err != nil { + t.Fatal(err) + } + if _, err := VerifyDockerOutputCommit(stagingRoot, request); err == nil { + t.Fatal("missing staged file verified") + } +} + +func TestDockerContainerIOPathBoundaryMatrix(t *testing.T) { + for name, value := range map[string]string{ + "absolute": "/etc/passwd", + "parent traversal": "../secret", + "nested traversal": "a/../../secret", + "windows separator": "a\\b", + "drive letter": "C:secret", + "empty component": "a//b", + "dot component": "a/./b", + "nul byte": "a\x00b", + } { + t.Run(name, func(t *testing.T) { + if err := validateContainerRelativePath("matrix", value); err == nil { + t.Fatalf("path %q was accepted", value) + } + }) + } + for name, value := range map[string]string{ + "plain": "report/summary.json", + "nested": "a/b/c.txt", + } { + t.Run("valid "+name, func(t *testing.T) { + if err := validateContainerRelativePath("matrix", value); err != nil { + t.Fatalf("path %q rejected: %v", value, err) + } + }) + } +} + +func TestDockerContainerIOErrors(t *testing.T) { + if code := DockerContainerIOErrorCode(errors.New("other")); code != "" { + t.Fatalf("unexpected code %q", code) + } + if code := DockerContainerIOErrorCode(newDockerContainerIOError(DockerContainerIOFailureConnection)); code != DockerContainerIOFailureConnection { + t.Fatalf("code = %q", code) + } + unavailable := NewUnavailableDockerContainerIOTransport("missing", "unavailable") + if _, err := unavailable.AttachLogs(context.Background(), testDockerLogCapturePlan(t, 1024, 64)); err == nil || DockerContainerIOErrorCode(err) != DockerContainerIOFailureUnavailable { + t.Fatalf("unavailable attach = %v", err) + } + if _, err := unavailable.ExportOutputs(context.Background(), testDockerOutputExportPlan(t)); err == nil { + t.Fatal("unavailable export succeeded") + } +} + +var _ = io.EOF diff --git a/internal/store/analyzer_execution_test.go b/internal/store/analyzer_execution_test.go index 029f70f5..470074ba 100644 --- a/internal/store/analyzer_execution_test.go +++ b/internal/store/analyzer_execution_test.go @@ -219,7 +219,7 @@ func removeSchemaV96ForTestStatements() []string { } func removeSchemaV97ForTestStatements() []string { - return []string{ + return append(removeSchemaV98ForTestStatements(), []string{ `DROP TRIGGER trg_sandbox_docker_lifecycle_cleanup_receipt_delete_immutable`, `DROP TRIGGER trg_sandbox_docker_lifecycle_cleanup_receipt_update_immutable`, `DROP TRIGGER trg_sandbox_docker_lifecycle_transition_delete_immutable`, @@ -245,7 +245,7 @@ func removeSchemaV97ForTestStatements() []string { `DROP INDEX idx_sandbox_docker_lifecycle_intents_run_created`, `DROP TABLE sandbox_docker_lifecycle_intents`, `DELETE FROM schema_migrations WHERE version = 97`, - } + }...) } func analyzerExecutionRequestFixture(t *testing.T, requestID string) ([]byte, analyzer.InvocationCandidate) { diff --git a/internal/store/docker_container_io.go b/internal/store/docker_container_io.go new file mode 100644 index 00000000..d3918564 --- /dev/null +++ b/internal/store/docker_container_io.go @@ -0,0 +1,173 @@ +package store + +import ( + "context" + "database/sql" + "errors" + "strings" + + "cyberagent-workbench/internal/apperror" + "cyberagent-workbench/internal/sandbox" +) + +// DockerContainerIOStore is the durable boundary of the container I/O +// contract: input projections, log capture receipts, staging receipts, and +// atomic output commit receipts. Raw container bytes never reach this store. +type DockerContainerIOStore interface { + InsertDockerInputProjection(ctx context.Context, + projection sandbox.DockerInputProjection) (bool, error) + InsertDockerLogCaptureReceipt(ctx context.Context, + receipt sandbox.DockerLogCaptureReceipt) (bool, error) + InsertDockerOutputStagingReceipt(ctx context.Context, + receipt sandbox.DockerOutputStagingReceipt) (bool, error) + CommitDockerOutputs(ctx context.Context, request sandbox.DockerOutputCommitRequest, + receipt sandbox.DockerOutputCommitReceipt) (bool, error) +} + +const insertDockerInputProjectionSQL = "INSERT INTO sandbox_docker_input_projections (id, attempt_id, generation, plan_id, observation_id, run_id, mission_id, workspace_id, protocol_version, input_artifact_digest, spec_fingerprint, authority_fingerprint, mount_target, mount_read_only, entry_count, total_bytes, projection_fingerprint, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?, ?)" + +func (s *SQLiteStore) InsertDockerInputProjection(ctx context.Context, + projection sandbox.DockerInputProjection, +) (bool, error) { + if projection.Validate() != nil { + return false, apperror.New(apperror.CodeInvalidArgument, "docker input projection is invalid") + } + result, err := s.db.ExecContext(ctx, insertDockerInputProjectionSQL, + projection.ID, projection.AttemptID, projection.Generation, projection.PlanID, + projection.ObservationID, projection.RunID, projection.MissionID, + projection.WorkspaceID, projection.ProtocolVersion, projection.InputArtifactDigest, + projection.SpecFingerprint, projection.AuthorityFingerprint, projection.MountTarget, + projection.EntryCount, projection.TotalBytes, projection.ProjectionFingerprint, + ts(projection.CreatedAt)) + if err != nil { + if isUniqueConstraintError(err) { + return false, nil + } + return false, err + } + count, err := result.RowsAffected() + return err == nil && count == 1, err +} + +const insertDockerLogCaptureReceiptSQL = "INSERT INTO sandbox_docker_log_capture_receipts (id, attempt_id, generation, run_id, container_id_fingerprint, protocol_version, status, stdout_bytes, stdout_lines, stdout_truncated_bytes, stdout_truncated_lines, stdout_truncated_deadline, stdout_utf8_violations, stdout_redacted_segments, stdout_content_digest, stderr_bytes, stderr_lines, stderr_truncated_bytes, stderr_truncated_lines, stderr_truncated_deadline, stderr_utf8_violations, stderr_redacted_segments, stderr_content_digest, capture_max_bytes, capture_max_lines, capture_fingerprint, receipt_fingerprint, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" + +func (s *SQLiteStore) InsertDockerLogCaptureReceipt(ctx context.Context, + receipt sandbox.DockerLogCaptureReceipt, +) (bool, error) { + if receipt.Validate() != nil || len(receipt.Streams) != 2 { + return false, apperror.New(apperror.CodeInvalidArgument, "docker log capture receipt is invalid") + } + stdout := receipt.Streams[0] + stderr := receipt.Streams[1] + result, err := s.db.ExecContext(ctx, insertDockerLogCaptureReceiptSQL, + receipt.ID, receipt.AttemptID, receipt.Generation, receipt.RunID, + receipt.ContainerIDFingerprint, receipt.ProtocolVersion, receipt.Status, + stdout.ByteCount, stdout.LineCount, storeBool(stdout.TruncatedBytes), + storeBool(stdout.TruncatedLines), storeBool(stdout.TruncatedDeadline), + stdout.UTF8Violations, stdout.RedactedSegments, stdout.ContentDigest, + stderr.ByteCount, stderr.LineCount, storeBool(stderr.TruncatedBytes), + storeBool(stderr.TruncatedLines), storeBool(stderr.TruncatedDeadline), + stderr.UTF8Violations, stderr.RedactedSegments, stderr.ContentDigest, + receipt.CaptureMaxBytes, receipt.CaptureMaxLines, receipt.CaptureFingerprint, + receipt.ReceiptFingerprint, ts(receipt.CreatedAt)) + if err != nil { + if isUniqueConstraintError(err) { + return false, nil + } + return false, err + } + count, err := result.RowsAffected() + return err == nil && count == 1, err +} + +const insertDockerOutputStagingReceiptSQL = "INSERT INTO sandbox_docker_output_staging_receipts (id, attempt_id, generation, run_id, container_id_fingerprint, protocol_version, status, file_count, total_bytes, redacted_count, entry_digest_set, export_fingerprint, receipt_fingerprint, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" + +func (s *SQLiteStore) InsertDockerOutputStagingReceipt(ctx context.Context, + receipt sandbox.DockerOutputStagingReceipt, +) (bool, error) { + if receipt.Validate() != nil { + return false, apperror.New(apperror.CodeInvalidArgument, "docker output staging receipt is invalid") + } + result, err := s.db.ExecContext(ctx, insertDockerOutputStagingReceiptSQL, + receipt.ID, receipt.AttemptID, receipt.Generation, receipt.RunID, + receipt.ContainerIDFingerprint, receipt.ProtocolVersion, receipt.Status, + receipt.FileCount, receipt.TotalBytes, receipt.RedactedCount, + receipt.EntryDigestSet, receipt.ExportFingerprint, receipt.ReceiptFingerprint, + ts(receipt.CreatedAt)) + if err != nil { + if isUniqueConstraintError(err) { + return false, nil + } + return false, err + } + count, err := result.RowsAffected() + return err == nil && count == 1, err +} + +const insertDockerOutputCommitReceiptSQL = "INSERT INTO sandbox_docker_output_commit_receipts (id, attempt_id, generation, run_id, workspace_id, operation_key_digest, request_fingerprint, protocol_version, status, committed_count, committed_digest_set, receipt_fingerprint, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" + +const insertDockerOutputCommitEntrySQL = "INSERT INTO sandbox_docker_output_commit_entries (receipt_id, ordinal, path, sha256, size_bytes, media_type) VALUES (?, ?, ?, ?, ?, ?)" + +const selectDockerOutputCommitByKeySQL = "SELECT id FROM sandbox_docker_output_commit_receipts WHERE operation_key_digest = ?" + +// CommitDockerOutputs atomically inserts the commit receipt and every accepted +// entry in one transaction. A conflicting operation key digest replays the +// existing receipt instead; a failure leaves no partial rows. +func (s *SQLiteStore) CommitDockerOutputs(ctx context.Context, + request sandbox.DockerOutputCommitRequest, receipt sandbox.DockerOutputCommitReceipt, +) (bool, error) { + if request.Validate() != nil || receipt.Validate() != nil || + receipt.RequestFingerprint != request.RequestFingerprint || + receipt.OperationKeyDigest != request.OperationKeyDigest || + len(receipt.Entries) != len(request.AcceptedEntries) { + return false, apperror.New(apperror.CodeInvalidArgument, "docker output commit request or receipt is invalid") + } + tx, err := s.db.BeginTx(ctx, &sql.TxOptions{}) + if err != nil { + return false, err + } + defer tx.Rollback() + var existingID string + err = tx.QueryRowContext(ctx, selectDockerOutputCommitByKeySQL, + request.OperationKeyDigest).Scan(&existingID) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return false, err + } + if err == nil { + return false, nil + } + if _, err := tx.ExecContext(ctx, insertDockerOutputCommitReceiptSQL, + receipt.ID, receipt.AttemptID, receipt.Generation, receipt.RunID, + receipt.WorkspaceID, receipt.OperationKeyDigest, receipt.RequestFingerprint, + receipt.ProtocolVersion, receipt.Status, receipt.CommittedCount, + receipt.CommittedDigestSet, receipt.ReceiptFingerprint, ts(receipt.CreatedAt)); err != nil { + return false, err + } + for index, entry := range receipt.Entries { + if _, err := tx.ExecContext(ctx, insertDockerOutputCommitEntrySQL, + receipt.ID, index+1, entry.Path, entry.SHA256, entry.SizeBytes, + entry.MediaType); err != nil { + return false, err + } + } + if err := tx.Commit(); err != nil { + return false, err + } + return true, nil +} + +func storeBool(value bool) int { + if value { + return 1 + } + return 0 +} + +func isUniqueConstraintError(err error) bool { + if err == nil { + return false + } + text := strings.ToLower(err.Error()) + return strings.Contains(text, "unique constraint") || + strings.Contains(text, "constraint failed") +} diff --git a/internal/store/docker_container_io_test.go b/internal/store/docker_container_io_test.go new file mode 100644 index 00000000..20fb2d6f --- /dev/null +++ b/internal/store/docker_container_io_test.go @@ -0,0 +1,151 @@ +package store + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "path/filepath" + "testing" + "time" + + "cyberagent-workbench/internal/idgen" + "cyberagent-workbench/internal/sandbox" +) + +func storeTestDigest(value string) string { + sum := sha256.Sum256([]byte(value)) + return hex.EncodeToString(sum[:]) +} + +func newDockerContainerIOStoreAuthority(t *testing.T, ctx context.Context, + st *SQLiteStore, runID, root, prefix string, +) (sandbox.DockerContainerPlan, sandbox.DockerContainerSpec, sandbox.DockerObservation) { + t.Helper() + _, manifest, observation := createDockerContainerPlanStoreAuthority(t, ctx, st, + runID, root, prefix) + plan, operation := newDockerContainerPlanStoreRecord(t, ctx, observation, manifest, + prefix+"-plan") + if _, _, err := st.CreateDockerContainerPlan(ctx, plan, operation); err != nil { + t.Fatal(err) + } + spec, err := sandbox.CompileDockerContainerSpec(ctx, observation, manifest) + if err != nil { + t.Fatal(err) + } + return plan, spec, observation +} + +func TestDockerContainerIOStoreLedger(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "docker-container-io.db") + st, run, root := openSandboxManifestStoreAt(t, ctx, path) + t.Cleanup(func() { _ = st.Close() }) + plan, spec, observation := newDockerContainerIOStoreAuthority(t, ctx, st, run.ID, root, + "docker-container-io") + + t.Run("input projection", func(t *testing.T) { + entries := []sandbox.DockerInputProjectionEntry{{ + Ordinal: 1, Path: "data/input.json", SHA256: storeTestDigest("input"), + SizeBytes: 4, MediaType: "application/json", + }} + projection, err := sandbox.NewDockerInputProjection(idgen.New("sandbox-docker-input"), + idgen.New("sandbox-docker-attempt"), 1, plan.ID, observation.ID, run.ID, + observation.MissionID, observation.WorkspaceID, observation.InputArtifactDigest, + spec.SpecFingerprint, spec.AuthorityFingerprint, + sandbox.DockerInputArtifactMountTarget, entries, time.Now().UTC()) + if err != nil { + t.Fatal(err) + } + inserted, err := st.InsertDockerInputProjection(ctx, projection) + if err != nil || !inserted { + t.Fatalf("projection insert = %t, %v", inserted, err) + } + replayed, err := st.InsertDockerInputProjection(ctx, projection) + if err != nil || replayed { + t.Fatalf("projection replay = %t, %v", replayed, err) + } + }) + + t.Run("log capture receipt", func(t *testing.T) { + capturePlan, err := sandbox.NewDockerLogCapturePlan(idgen.New("sandbox-docker-attempt"), 1, + run.ID, storeTestDigest("container"), 1024, 64, 60) + if err != nil { + t.Fatal(err) + } + records := []sandbox.DockerLogStreamRecord{ + {Stream: "stdout", ByteCount: 6, LineCount: 1, ContentDigest: storeTestDigest("hello")}, + {Stream: "stderr", ContentDigest: storeTestDigest("")}, + } + receipt, err := sandbox.NewDockerLogCaptureReceipt(idgen.New("sandbox-docker-log"), + capturePlan.AttemptID, 1, capturePlan.RunID, capturePlan.ContainerIDFingerprint, + capturePlan, records, sandbox.DockerLogCaptureStatusCompleted, time.Now().UTC()) + if err != nil { + t.Fatal(err) + } + inserted, err := st.InsertDockerLogCaptureReceipt(ctx, receipt) + if err != nil || !inserted { + t.Fatalf("log receipt insert = %t, %v", inserted, err) + } + replayed, err := st.InsertDockerLogCaptureReceipt(ctx, receipt) + if err != nil || replayed { + t.Fatalf("log receipt replay = %t, %v", replayed, err) + } + }) + + t.Run("staging and atomic commit", func(t *testing.T) { + exportPlan, err := sandbox.NewDockerOutputExportPlan(idgen.New("sandbox-docker-attempt"), 1, + run.ID, storeTestDigest("container"), "/run/cyberagent/outputs", + sandbox.MaxDockerOutputFiles, sandbox.MaxDockerOutputFileBytes, + sandbox.MaxDockerOutputTotalBytes) + if err != nil { + t.Fatal(err) + } + entry := sandbox.DockerStagedOutputEntry{Path: "report/result.json", + SHA256: storeTestDigest("result"), SizeBytes: 6, MediaType: "application/json"} + staging, err := sandbox.NewDockerOutputStagingReceipt(idgen.New("sandbox-docker-staging"), + exportPlan.AttemptID, 1, exportPlan.RunID, exportPlan.ContainerIDFingerprint, + exportPlan, []sandbox.DockerStagedOutputEntry{entry}, + sandbox.DockerOutputStagingStatusCompleted, time.Now().UTC()) + if err != nil { + t.Fatal(err) + } + inserted, err := st.InsertDockerOutputStagingReceipt(ctx, staging) + if err != nil || !inserted { + t.Fatalf("staging insert = %t, %v", inserted, err) + } + accepted := []sandbox.DockerOutputCommitEntry{{ + Path: entry.Path, SHA256: entry.SHA256, SizeBytes: entry.SizeBytes, + MediaType: entry.MediaType, + }} + request, err := sandbox.NewDockerOutputCommitRequest(staging.AttemptID, 1, staging.RunID, + observation.WorkspaceID, staging.ID, storeTestDigest("operation"), accepted) + if err != nil { + t.Fatal(err) + } + receipt, err := sandbox.NewDockerOutputCommitReceipt(idgen.New("sandbox-docker-commit"), + request.AttemptID, request.Generation, request.RunID, request.WorkspaceID, + request, time.Now().UTC()) + if err != nil { + t.Fatal(err) + } + committed, err := st.CommitDockerOutputs(ctx, request, receipt) + if err != nil || !committed { + t.Fatalf("commit = %t, %v", committed, err) + } + replayed, err := st.CommitDockerOutputs(ctx, request, receipt) + if err != nil || replayed { + t.Fatalf("commit replay = %t, %v", replayed, err) + } + }) +} + +func removeSchemaV98ForTestStatements() []string { + return []string{ + `DROP TABLE sandbox_docker_output_commit_entries`, + `DROP TABLE sandbox_docker_output_commit_receipts`, + `DROP TABLE sandbox_docker_output_staging_receipts`, + `DROP TABLE sandbox_docker_log_capture_receipts`, + `DROP TABLE sandbox_docker_input_projections`, + `DELETE FROM schema_migrations WHERE version = 98`, + } +} diff --git a/internal/store/migration_v98.go b/internal/store/migration_v98.go new file mode 100644 index 00000000..fc7859ad --- /dev/null +++ b/internal/store/migration_v98.go @@ -0,0 +1,196 @@ +package store + +var sandboxDockerContainerIOStatements = []string{ + `CREATE TABLE sandbox_docker_input_projections ( + id TEXT PRIMARY KEY, + attempt_id TEXT NOT NULL UNIQUE, + generation INTEGER NOT NULL, + plan_id TEXT NOT NULL, + observation_id TEXT NOT NULL, + run_id TEXT NOT NULL, + mission_id TEXT NOT NULL, + workspace_id TEXT NOT NULL, + protocol_version TEXT NOT NULL, + input_artifact_digest TEXT NOT NULL, + spec_fingerprint TEXT NOT NULL, + authority_fingerprint TEXT NOT NULL, + mount_target TEXT NOT NULL, + mount_read_only INTEGER NOT NULL, + entry_count INTEGER NOT NULL, + total_bytes INTEGER NOT NULL, + projection_fingerprint TEXT NOT NULL UNIQUE, + created_at TEXT NOT NULL, + FOREIGN KEY(plan_id) REFERENCES sandbox_docker_container_plans(id) ON DELETE RESTRICT, + FOREIGN KEY(run_id) REFERENCES runs(id) ON DELETE RESTRICT, + FOREIGN KEY(mission_id) REFERENCES missions(id) ON DELETE RESTRICT, + FOREIGN KEY(workspace_id) REFERENCES workspaces(id) ON DELETE RESTRICT, + CHECK(protocol_version = 'sandbox_docker_input_projection.v1'), + CHECK(generation = 1), + CHECK(mount_target = '/run/cyberagent/inputs'), + CHECK(mount_read_only = 1), + CHECK(entry_count BETWEEN 1 AND 16), + CHECK(total_bytes BETWEEN 1 AND 16777216), + CHECK(length(input_artifact_digest) = 64 + AND input_artifact_digest = lower(input_artifact_digest) + AND input_artifact_digest NOT GLOB '*[^0-9a-f]*'), + CHECK(length(spec_fingerprint) = 64 + AND spec_fingerprint = lower(spec_fingerprint) + AND spec_fingerprint NOT GLOB '*[^0-9a-f]*'), + CHECK(length(authority_fingerprint) = 64 + AND authority_fingerprint = lower(authority_fingerprint) + AND authority_fingerprint NOT GLOB '*[^0-9a-f]*'), + CHECK(length(projection_fingerprint) = 64 + AND projection_fingerprint = lower(projection_fingerprint) + AND projection_fingerprint NOT GLOB '*[^0-9a-f]*') + )`, + `CREATE TABLE sandbox_docker_log_capture_receipts ( + id TEXT PRIMARY KEY, + attempt_id TEXT NOT NULL, + generation INTEGER NOT NULL, + run_id TEXT NOT NULL, + container_id_fingerprint TEXT NOT NULL, + protocol_version TEXT NOT NULL, + status TEXT NOT NULL, + stdout_bytes INTEGER NOT NULL, + stdout_lines INTEGER NOT NULL, + stdout_truncated_bytes INTEGER NOT NULL, + stdout_truncated_lines INTEGER NOT NULL, + stdout_truncated_deadline INTEGER NOT NULL, + stdout_utf8_violations INTEGER NOT NULL, + stdout_redacted_segments INTEGER NOT NULL, + stdout_content_digest TEXT NOT NULL, + stderr_bytes INTEGER NOT NULL, + stderr_lines INTEGER NOT NULL, + stderr_truncated_bytes INTEGER NOT NULL, + stderr_truncated_lines INTEGER NOT NULL, + stderr_truncated_deadline INTEGER NOT NULL, + stderr_utf8_violations INTEGER NOT NULL, + stderr_redacted_segments INTEGER NOT NULL, + stderr_content_digest TEXT NOT NULL, + capture_max_bytes INTEGER NOT NULL, + capture_max_lines INTEGER NOT NULL, + capture_fingerprint TEXT NOT NULL, + receipt_fingerprint TEXT NOT NULL UNIQUE, + created_at TEXT NOT NULL, + FOREIGN KEY(run_id) REFERENCES runs(id) ON DELETE RESTRICT, + CHECK(protocol_version = 'sandbox_docker_log_capture.v1'), + CHECK(generation = 1), + CHECK(status IN ('completed', 'truncated_bytes', 'truncated_lines', + 'truncated_deadline', 'invalid_stream')), + CHECK(capture_max_bytes BETWEEN 1 AND 262144), + CHECK(capture_max_lines BETWEEN 1 AND 4096), + CHECK(stdout_bytes BETWEEN 0 AND capture_max_bytes), + CHECK(stderr_bytes BETWEEN 0 AND capture_max_bytes), + CHECK(stdout_lines BETWEEN 0 AND capture_max_lines), + CHECK(stderr_lines BETWEEN 0 AND capture_max_lines), + CHECK(stdout_truncated_bytes IN (0, 1) AND stdout_truncated_lines IN (0, 1) + AND stdout_truncated_deadline IN (0, 1)), + CHECK(stderr_truncated_bytes IN (0, 1) AND stderr_truncated_lines IN (0, 1) + AND stderr_truncated_deadline IN (0, 1)), + CHECK(stdout_utf8_violations >= 0 AND stderr_utf8_violations >= 0), + CHECK(stdout_redacted_segments >= 0 AND stderr_redacted_segments >= 0), + CHECK(length(container_id_fingerprint) = 64 + AND container_id_fingerprint = lower(container_id_fingerprint) + AND container_id_fingerprint NOT GLOB '*[^0-9a-f]*'), + CHECK(length(stdout_content_digest) = 64 + AND stdout_content_digest = lower(stdout_content_digest) + AND stdout_content_digest NOT GLOB '*[^0-9a-f]*'), + CHECK(length(stderr_content_digest) = 64 + AND stderr_content_digest = lower(stderr_content_digest) + AND stderr_content_digest NOT GLOB '*[^0-9a-f]*'), + CHECK(length(capture_fingerprint) = 64 + AND capture_fingerprint = lower(capture_fingerprint) + AND capture_fingerprint NOT GLOB '*[^0-9a-f]*'), + CHECK(length(receipt_fingerprint) = 64 + AND receipt_fingerprint = lower(receipt_fingerprint) + AND receipt_fingerprint NOT GLOB '*[^0-9a-f]*') + )`, + `CREATE TABLE sandbox_docker_output_staging_receipts ( + id TEXT PRIMARY KEY, + attempt_id TEXT NOT NULL UNIQUE, + generation INTEGER NOT NULL, + run_id TEXT NOT NULL, + container_id_fingerprint TEXT NOT NULL, + protocol_version TEXT NOT NULL, + status TEXT NOT NULL, + file_count INTEGER NOT NULL, + total_bytes INTEGER NOT NULL, + redacted_count INTEGER NOT NULL, + entry_digest_set TEXT NOT NULL, + export_fingerprint TEXT NOT NULL, + receipt_fingerprint TEXT NOT NULL UNIQUE, + created_at TEXT NOT NULL, + FOREIGN KEY(run_id) REFERENCES runs(id) ON DELETE RESTRICT, + CHECK(protocol_version = 'sandbox_docker_output_staging.v1'), + CHECK(generation = 1), + CHECK(status IN ('completed', 'truncated_bytes', 'invalid_archive', + 'rejected_path', 'rejected_link', 'rejected_duplicate')), + CHECK(file_count BETWEEN 0 AND 64), + CHECK(total_bytes BETWEEN 0 AND 16777216), + CHECK(redacted_count BETWEEN 0 AND file_count), + CHECK(length(container_id_fingerprint) = 64 + AND container_id_fingerprint = lower(container_id_fingerprint) + AND container_id_fingerprint NOT GLOB '*[^0-9a-f]*'), + CHECK(length(entry_digest_set) = 64 + AND entry_digest_set = lower(entry_digest_set) + AND entry_digest_set NOT GLOB '*[^0-9a-f]*'), + CHECK(length(export_fingerprint) = 64 + AND export_fingerprint = lower(export_fingerprint) + AND export_fingerprint NOT GLOB '*[^0-9a-f]*'), + CHECK(length(receipt_fingerprint) = 64 + AND receipt_fingerprint = lower(receipt_fingerprint) + AND receipt_fingerprint NOT GLOB '*[^0-9a-f]*') + )`, + `CREATE TABLE sandbox_docker_output_commit_receipts ( + id TEXT PRIMARY KEY, + attempt_id TEXT NOT NULL, + generation INTEGER NOT NULL, + run_id TEXT NOT NULL, + workspace_id TEXT NOT NULL, + operation_key_digest TEXT NOT NULL UNIQUE, + request_fingerprint TEXT NOT NULL, + protocol_version TEXT NOT NULL, + status TEXT NOT NULL, + committed_count INTEGER NOT NULL, + committed_digest_set TEXT NOT NULL, + receipt_fingerprint TEXT NOT NULL UNIQUE, + created_at TEXT NOT NULL, + FOREIGN KEY(run_id) REFERENCES runs(id) ON DELETE RESTRICT, + FOREIGN KEY(workspace_id) REFERENCES workspaces(id) ON DELETE RESTRICT, + CHECK(protocol_version = 'sandbox_docker_output_commit.v1'), + CHECK(generation = 1), + CHECK(status = 'committed'), + CHECK(committed_count BETWEEN 1 AND 64), + CHECK(length(operation_key_digest) = 64 + AND operation_key_digest = lower(operation_key_digest) + AND operation_key_digest NOT GLOB '*[^0-9a-f]*'), + CHECK(length(request_fingerprint) = 64 + AND request_fingerprint = lower(request_fingerprint) + AND request_fingerprint NOT GLOB '*[^0-9a-f]*'), + CHECK(length(committed_digest_set) = 64 + AND committed_digest_set = lower(committed_digest_set) + AND committed_digest_set NOT GLOB '*[^0-9a-f]*'), + CHECK(length(receipt_fingerprint) = 64 + AND receipt_fingerprint = lower(receipt_fingerprint) + AND receipt_fingerprint NOT GLOB '*[^0-9a-f]*') + )`, + `CREATE TABLE sandbox_docker_output_commit_entries ( + receipt_id TEXT NOT NULL, + ordinal INTEGER NOT NULL, + path TEXT NOT NULL, + sha256 TEXT NOT NULL, + size_bytes INTEGER NOT NULL, + media_type TEXT NOT NULL, + PRIMARY KEY (receipt_id, ordinal), + FOREIGN KEY(receipt_id) REFERENCES sandbox_docker_output_commit_receipts(id) + ON DELETE RESTRICT, + CHECK(ordinal BETWEEN 1 AND 64), + CHECK(length(path) BETWEEN 1 AND 1024), + CHECK(path NOT GLOB '/*' AND path NOT GLOB '*..*'), + CHECK(size_bytes BETWEEN 1 AND 4194304), + CHECK(length(sha256) = 64 + AND sha256 = lower(sha256) + AND sha256 NOT GLOB '*[^0-9a-f]*'), + CHECK(length(media_type) BETWEEN 1 AND 256) + )`, +} diff --git a/internal/store/migrations.go b/internal/store/migrations.go index b8475264..70d6cd02 100644 --- a/internal/store/migrations.go +++ b/internal/store/migrations.go @@ -11,7 +11,7 @@ import ( "time" ) -const LatestSchemaVersion = 97 +const LatestSchemaVersion = 98 type migration struct { Version int diff --git a/internal/store/sqlite.go b/internal/store/sqlite.go index 260f6b2d..bff64309 100644 --- a/internal/store/sqlite.go +++ b/internal/store/sqlite.go @@ -324,6 +324,7 @@ func migrationPlan() []migration { {Version: 95, Name: "atomic embedded analyzer execution artifacts", Statements: analyzerExecutionCommitStatements}, {Version: 96, Name: "review-gated exact host command proposals", Statements: hostCommandProposalStatements}, {Version: 97, Name: "durable Docker lifecycle ownership and recovery", Statements: sandboxDockerLifecycleStatements}, + {Version: 98, Name: "bounded Docker container I/O contract and atomic output commit", Statements: sandboxDockerContainerIOStatements}, } } From 30e861122c1ce0ee3ed314866044107f570a91b0 Mon Sep 17 00:00:00 2001 From: gaoming <2168502171@qq.com> Date: Fri, 14 Aug 2026 15:28:57 +0800 Subject: [PATCH 2/2] docs: record the bounded Docker container I/O contract (ADR 0098) Update the project memory, status, progress, and task ledgers with the schema-v98 slice for issue #39 and add ADR 0098 documenting the sealed read-only input projection, bounded log capture, strict output staging, atomic commit, and the closed two-endpoint daemon surface. --- docs/PROGRESS_BOOK.md | 2 ++ docs/PROJECT_MEMORY.md | 41 ++++++++++++++++++++++++++++++++++++++--- docs/PROJECT_STATUS.md | 16 +++++++++++++++- docs/TASK_BOOK.md | 2 ++ 4 files changed, 57 insertions(+), 4 deletions(-) diff --git a/docs/PROGRESS_BOOK.md b/docs/PROGRESS_BOOK.md index c68868e2..3d97f9f0 100644 --- a/docs/PROGRESS_BOOK.md +++ b/docs/PROGRESS_BOOK.md @@ -6,6 +6,8 @@ ## 一、当前阶段 +2026-08-14 单切片 `Docker 容器 I/O 合同（新增标签，对应 Issue #39）` 推进到 SQLite v98，仍未接入任何产品入口或执行授权。只读输入投影 `sandbox_docker_input_projection.v1` 把规范化相对路径、SHA-256、大小与媒体类型封印到生命周期 attempt/generation、Plan、Observation、Run/Mission/Workspace 与 Spec 指纹上，输入挂载固定 `/run/cyberagent/inputs` 只读，并新增真实 inspect Mounts 隔离校验（唯一可写挂载必须是专用输出挂载，输入与工作区树必须只读）。日志捕获 `sandbox_docker_log_capture.v1` 只走 `POST containers/{id}/attach?logs=1&stderr=1&stdout=1&stream=0`：解复用器按流限制 256 KiB/4096 行与墙钟时限，拒绝畸形/超大帧，替换非法 UTF-8 并计数，脱敏 Secret，只落库元数据与摘要回执。输出暂存 `sandbox_docker_output_staging.v1` 只导出专用输出挂载的 `GET containers/{id}/archive`，tar 走查在 Windows/Linux 同一路径规则下拒绝绝对/穿越/反斜杠/盘符路径、symlink/hardlink/设备节点、重复项，并限制 64 文件/单文件 4 MiB/总量 16 MiB，文本文件按媒体类型识别与脱敏后写入进程内暂存目录；被拒归档不带任何受信清单。原子提交 `sandbox_docker_output_commit.v1` 要求接受清单与已完成暂存回执逐项精确一致，从暂存目录重读并复哈希后在同一 SQLite 事务中写入回执与全部条目，operation key 幂等，失败不残留半提交行。HTTP 白名单只开放上述两个端点，应用服务不接 CLI/HTTP/Desktop，新增 15 个事件类型记录 prepared/acquired/taken-over/failed/completed。域/黄金向量/对抗路径/传输/Store/应用定向测试在 macOS 通过；Windows 与 Linux CI 将跑同一路径矩阵。边界见 ADR 0098 与 Issue #39；产品准入（Policy/Approval/Budget/Network 与 CLI/HTTP/Desktop 接线）仍是下一个 Docker 切片。 + 2026-08-14 单切片 `D0-Mac（macOS Desktop 便携构建，新增标签）` 保持 SQLite v97 与全部权限边界不变，为 Desktop 增加 macOS 构建：`cmd/cyberagent-desktop` 按构建标签拆分为共享 `desktop`、Windows `windows && desktop && wv2runtime.error` 与新增 darwin 实现；macOS 使用系统自带 WKWebView 且无运行时预检，启动失败只写 stderr 并显示有界严格转义的 osascript 对话框，工作区启动器只通过固定 `/usr/bin/open` 打开已验证 .app（Finder、Terminal、Antigravity、PyCharm、WebStorm、Visual Studio Code）。Windows 专属的 Credential Manager、ConPTY 用户终端、Safe Web/WFP 与受控宿主执行在 macOS 保持失败关闭。`scripts/build-desktop-darwin.sh` 产出 ad-hoc 签名、未公证的 `build/desktop/Prayu.app`，复用连续双构建 SHA-256 校验并写入 `portable_release_metadata.v1`；`scripts/check-macos-compat.sh` 检查 Mach-O/架构/codesign/元数据与 operator-preview 包边界；CI 新增 `desktop-macos` 任务；React 标题栏为原生红绿灯预留空间。desktop-tag Go 测试在 macOS 通过，App/desktop-bridge React 测试通过；签名、公证与人工 macOS 矩阵未做，`release_ready` 保持 false。边界见 ADR 0097 与 `docs/DESKTOP_PLAN.md`；本轮没有重跑 WFP、Docker、付费 Provider 或全仓重型门，下次上下文压缩后不得重复。 项目正在从可运行的 v0.1 CLI/TUI 骨架迁移到 V2 Run-centric Runtime。当前只推进通用 Agent Harness 与 Code 工作流；CTF 专用求解和攻防自动化已经移出活跃路线图，仅保留通用扩展接口供未来独立附加包接入。 diff --git a/docs/PROJECT_MEMORY.md b/docs/PROJECT_MEMORY.md index e947e5a7..93955d4f 100644 --- a/docs/PROJECT_MEMORY.md +++ b/docs/PROJECT_MEMORY.md @@ -4,7 +4,41 @@ Last updated: 2026-08-14 -## Current Single-Slice Checkpoint: macOS Desktop Portable Build / Schema v97 Unchanged +## Current Single-Slice Checkpoint: Bounded Docker Container I/O Contract / Schema v98 + +On 2026-08-14 schema v98 added the bounded Docker container I/O contract for +issue #39, still without any product entry or authority grant. A sealed +read-only input projection (sandbox_docker_input_projection.v1) binds +canonical relative paths, SHA-256, size, and media type to the lifecycle +attempt/generation, plan, observation, Run, Mission, Workspace, and spec; the +input mount stays fixed at /run/cyberagent/inputs read-only and a mount +isolation verifier checks real inspection mounts (exactly one writable mount, +owned by the dedicated output target; input and workspace trees read-only). +Log capture (sandbox_docker_log_capture.v1) goes only through POST +containers/{id}/attach with logs=1&stderr=1&stdout=1&stream=0; a demuxer +bounds bytes (256 KiB) and lines (4096) per stream, enforces a deadline, +rejects malformed/oversized frames, counts and replaces invalid UTF-8, +redacts secrets, and persists only digest-only receipts. Output staging +(sandbox_docker_output_staging.v1) exports only GET containers/{id}/archive +of the dedicated output mount and walks the tar under one Windows/Linux path +rule set: no absolute/traversal/backslash/drive-letter paths, no +symlink/hardlink/device entries, no duplicates, caps of 64 files / 4 MiB per +file / 16 MiB total, media-type detection, and text redaction into a +process-local staging directory; rejected archives carry no trusted manifest. +Atomic output commit (sandbox_docker_output_commit.v1) requires an accepted +manifest that exactly matches completed staging entries, re-reads and +re-hashes the staged files, and inserts the receipt plus every entry in one +replay-safe SQLite transaction; a failure leaves no partial rows. The HTTP +transport allowlist admits exactly the two endpoints, the application service +stays unwired from CLI/HTTP/Desktop, and 15 new event kinds record the +prepared/acquired/taken-over/failed/completed transitions. Focused domain, +golden-vector, adversarial-path, transport, store, and application tests pass +on macOS; Windows/Linux CI will exercise the same path matrix. See ADR 0098 +and issue #39. Do not repeat this slice or the v97 lifecycle review after +compaction; product admission (Policy/Approval/Budget/Network and +CLI/HTTP/Desktop wiring) remains the next Docker slice. + +### Previous Checkpoint: macOS Desktop Portable Build / Schema v97 Unchanged On 2026-08-14 a new single-slice delivery added a macOS portable Desktop build without changing SQLite schema or any authority boundary. cmd/cyberagent-desktop @@ -565,6 +599,7 @@ Read in this order after a long context break: 102. `docs/DESKTOP_PLAN.md` 103. `docs/SKILL_PACKAGE_PLAN.md` 104. `docs/adr/0097-macos-desktop-portable-build.md` +105. `docs/adr/0098-bounded-docker-container-io-contract.md` ## Current Baseline @@ -573,8 +608,8 @@ Read in this order after a long context break: - Generic coding-agent workflow usability: about 98%. - Cyber autonomous-workflow usability: about 20%. - These are engineering estimates based on tested roadmap slices, not performance benchmarks. Do not reuse the retired single-axis "overall product vision" percentage. -- Database schema: v97. -- `README.md` carries the canonical bilingual schema timeline in strict `v1 -> v97` order. `internal/store/readme_history_test.go` binds its row count and ordering to `LatestSchemaVersion`, so a future migration cannot silently leave the public history missing or out of sequence. +- Database schema: v98. +- `README.md` carries the canonical bilingual schema timeline in strict `v1 -> v98` order. `internal/store/readme_history_test.go` binds its row count and ordering to `LatestSchemaVersion`, so a future migration cannot silently leave the public history missing or out of sequence. - Main languages: Go control plane, TypeScript React/Vite local console, and deterministic Rust 1.97.1 digest/ZIP protocol functions. Rust has no Agent, LLM, config, key, persistence, network, filesystem, subprocess, or product-lifecycle ownership. - Analyzer status: P10-A through P10-K define and validate the Go/Rust protocol and embedded-WASI boundary. P10-L/schema v94-v95 adds real fixed-module execution, one-shot exact-bound authorization, atomic consumption, redacted execution, metadata-only Artifact content, and Run events. P10-M exposes only this embedded module through CLI/control-token HTTP/Desktop/React; callers cannot provide WebAssembly, imports, mount, network, command, argv, environment, or native process. See ADR 0062, ADR 0063, ADR 0090, ADR 0091, and `analyzers/README.md`. - Model Harness status: non-schema A1/A2/A3 adds `model_harness.v1` exact transport/tool/JSON/streaming profiles, Go preflight for Root/Specialist/read-only Fan-out, and `model_harness_qualification.v1` at-most-two-call synthetic qualification. Mock is trusted offline; Anthropic-compatible models require explicit qualification. Qualification stores only exact binding digest, capability booleans, and seven-day expiry in existing Provider settings; the synthetic Tool is never executed, availability remains no-probe, and qualification grants no Tool/Shell/file/browser/Docker authority. P13-A adds the durable public-activity read projection; P13-B adds a separate process-local safe public assistant stream with exact cancellation and no raw Provider persistence; P13-C adds the Root-only, independently reviewed `approval` host-command proposal Tool without granting the model execution authority. P10-M2 proves the production Anthropic-compatible route and durable chat path against deterministic local SSE, while P13-B3 verifies one configured real DeepSeek path. Current OpenAPI is 88 paths / 96 operations / 212 schemas. See ADR 0074, ADR 0080, ADR 0091, ADR 0092, and ADR 0093. diff --git a/docs/PROJECT_STATUS.md b/docs/PROJECT_STATUS.md index 3551e28d..be83af84 100644 --- a/docs/PROJECT_STATUS.md +++ b/docs/PROJECT_STATUS.md @@ -6,7 +6,21 @@ Last updated: 2026-08-14 ## Resume Context -The current single-slice checkpoint is the macOS Desktop portable build at +The current single-slice checkpoint is the bounded Docker container I/O +contract at schema v98 for issue #39, still without product entry or +authority. A sealed read-only input projection binds canonical paths, +digests, sizes, and media types to the lifecycle attempt/generation and plan +identity, and a mount isolation verifier checks real inspection mounts. Log +capture is limited to the fixed stream=0 attach surface with byte/line/ +deadline caps, UTF-8 normalization, and secret redaction, persisting only +digest receipts. Output staging exports only the dedicated output mount and +rejects traversal, links, duplicates, and size violations under one +Windows/Linux path rule set; atomic commit requires an exact manifest match, +re-hashes staged files, and writes receipt plus entries in one replay-safe +transaction. The HTTP allowlist admits exactly two daemon endpoints and the +service is unwired from CLI/HTTP/Desktop. See ADR 0098. + +The previous single-slice checkpoint was the macOS Desktop portable build at unchanged schema v97. cmd/cyberagent-desktop is now split by build tag: shared shell code under "desktop", Windows-only WebView2/Acrylic/registry launcher code under "windows && desktop && wv2runtime.error", and a new diff --git a/docs/TASK_BOOK.md b/docs/TASK_BOOK.md index b26f9665..15ad5975 100644 --- a/docs/TASK_BOOK.md +++ b/docs/TASK_BOOK.md @@ -10,6 +10,8 @@ ## 当前基线 +Docker 容器 I/O 合同（Issue #39）已随 schema v98 落地（ADR 0098）：只读输入投影、有界日志回执、严格输出暂存与原子 Artifact 提交全部就位，但仍无产品入口；产品准入（Policy/Approval/Budget/Network 与 CLI/HTTP/Desktop 接线）是下一个 Docker 切片。 + D0-Mac 已把 Desktop 构建扩展到 macOS（ADR 0097）：darwin 构建标签复用同一 Go 控制平面与 WKWebView，产出未签名/未公证的本地 Prayu.app；Windows 专属的凭证库、ConPTY 终端、Safe Web/WFP 与受控宿主执行在 macOS 失败关闭。macOS 签名、公证与人工矩阵仍未排入当前切片。 - 架构完成度：约 99%；其中 V2 Run-centric 控制平面约 99%。