From 163fbda3949a74f452301a5dc1f5c7ef0f4e769a Mon Sep 17 00:00:00 2001 From: npc0-hue Date: Tue, 1 Sep 2026 17:15:09 +0800 Subject: [PATCH] Add runtime SQLite query targets --- AGENTS.md | 2 +- README.md | 6 +- protocol/job.md | 2 +- protocol/log-ingest.md | 11 +- protocol/protected-request.md | 4 +- runtime/data_targets.go | 45 +++++ runtime/data_targets_test.go | 82 ++++++++ runtime/lifecycle.go | 53 +++-- runtime/lifecycle_test.go | 52 ++++- runtime/log_sources.go | 42 ++-- runtime/process_supervisor.go | 25 ++- runtime/protected_request.go | 40 +--- runtime/protected_request_test.go | 19 +- runtime/sqlite_query.go | 312 ++++++++++++++++++++++++++++++ runtime/sqlite_query_test.go | 103 ++++++++++ runtime/worker.go | 22 ++- runtime/worker_test.go | 2 +- 17 files changed, 708 insertions(+), 114 deletions(-) create mode 100644 runtime/sqlite_query.go create mode 100644 runtime/sqlite_query_test.go diff --git a/AGENTS.md b/AGENTS.md index 654252a..533061e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,7 +20,7 @@ Protocol structs live in `protocol/`. Local runtime types live in `runtime/` or ## Safety Rules -Run must enforce scoped paths and never expose raw host paths, local secrets, or unrestricted command execution to platform_web or plugins. +Run must enforce scoped paths and never expose raw host paths, local secrets, or unrestricted command execution to platform_web or plugins. Do not apply game-specific redaction to plugin-declared game records or SQL query rows: return those bounded typed fields faithfully through Platform channels. Supervised stdout/stderr and plugin-declared file tails are opaque verbatim channels: do not inspect, parse, filter, redact, truncate by content, transform, or special-case their payloads. They cannot be used to derive player or plugin records. This pass-through rule does not grant plugins or the browser a direct host-path, credential, key, or socket API outside that log channel. ## Generic Executor Boundary diff --git a/README.md b/README.md index b43d958..57e1c7a 100644 --- a/README.md +++ b/README.md @@ -72,7 +72,7 @@ In Docker, `RUN_PLATFORM_URL` must be `http://platform:8080` because `platform` Current executable behavior includes smoke mode plus worker mode. Worker mode registers with platform, opens a signed persistent control event stream for lightweight wakeups, sends heartbeat metadata, claims lifecycle jobs, acknowledges leases, reports bounded progress, executes scoped `process.install`, `process.start`, and `process.stop` command templates inside per-server workspaces, polls cancellation, submits terminal results, and reconciles active jobs. -Lifecycle templates are JSON files addressed by logical keys under the server workspace. They resolve to direct executable/argument vectors, not shell strings. Absolute paths, parent traversal, raw credentials, direct sockets, shell launchers, unsafe environment keys, and unsafe output are rejected or redacted. Plugin-declared Windows `.cmd` and `.bat` assets are launched through a bounded `cmd.exe` adapter and remain under the same process supervisor. Process identity journals are namespaced by Run endpoint, server, plugin, and component profile, so multiple Run services cannot overwrite one another; a restart migrates matching legacy process state and resumes observing stdout/stderr from the current supervised process. Process stdout/stderr is pushed through best-effort live relay, and lifecycle result metadata is queued through artifact hooks so control heartbeat and job result submission stay independent from log and artifact work. +Lifecycle templates are JSON files addressed by logical keys under the server workspace. They resolve to direct executable/argument vectors, not shell strings. Absolute paths, parent traversal, raw credentials, direct sockets, shell launchers, and unsafe environment keys are rejected. Plugin-declared Windows `.cmd` and `.bat` assets are launched through a bounded `cmd.exe` adapter and remain under the same process supervisor. Process identity journals are namespaced by Run endpoint, server, plugin, and component profile, so multiple Run services cannot overwrite one another; a restart migrates matching legacy process state and resumes observing stdout/stderr from the current supervised process. Process stdout/stderr is an opaque verbatim stream pushed through best-effort live relay, and lifecycle result metadata is queued through artifact hooks so control heartbeat and job result submission stay independent from log and artifact work. The control event stream carries only small hints such as `control.ready`, `control.heartbeat`, and `job.changed`. It never carries assignments, logs, artifact chunks, file bodies, host paths, credentials, or direct sockets; Run still fetches work through the durable job claim channel after a wake event. @@ -94,6 +94,6 @@ Worker mode now dispatches distribution capabilities in addition to lifecycle wo - `dependencies.install`: executes only typed install plans addressed under `dependencies/install/...`; arbitrary shell snippets are rejected before execution. - `logs.backfill`: advances historical log cursors for declared sources and returns a cursor/result artifact ref instead of embedding large log bodies in job results. -Declared file log sources used by explicit maintenance jobs remain bounded and redacted, but current supervised process output uses live relay only. FTP/rsync, SQL read, RCON command, and file transfer adapters are represented as bounded envelopes with scoped input or artifact refs. Long transfers remain lower priority than heartbeat, job ack/result, cancellation polling, reconcile, and current live log relay. +Declared file log sources and current supervised process output use opaque verbatim relay entries; Run does not inspect, filter, redact, or content-truncate them. FTP/rsync, SQL read, RCON command, and file transfer adapters are represented as bounded envelopes with scoped input or artifact refs. Long transfers remain lower priority than heartbeat, job ack/result, cancellation polling, reconcile, and current live log relay. -Protected SQL, RCON, and management-program requests use a separate signed one-time input route after Run claims a single-attempt fenced job. Run rechecks approval, expiry, server/endpoint/fence, capability kind, and logical transport/target bindings before dispatching to a local handler. Request text, private connection configuration, and response bodies do not enter assignments, journals, or terminal results. Management-program stdout/stderr is redacted before live relay and must not be stored as a Run-owned durable log body. See [`protocol/protected-request.md`](protocol/protected-request.md). +Protected SQL, RCON, and management-program requests use a separate signed one-time input route after Run claims a single-attempt fenced job. Run rechecks approval, expiry, server/endpoint/fence, capability kind, and logical transport/target bindings before dispatching to a local handler. Request text, private connection configuration, and response bodies do not enter assignments, journals, or terminal results. Management-program stdout/stderr is opaque verbatim live-log output and must not be stored as a Run-owned durable log body. See [`protocol/protected-request.md`](protocol/protected-request.md). diff --git a/protocol/job.md b/protocol/job.md index d91befa..a488429 100644 --- a/protocol/job.md +++ b/protocol/job.md @@ -76,7 +76,7 @@ The executor resolves lifecycle action templates under the scoped server workspa - Remote database and RCON jobs must use scoped input/artifact refs rather than embedding query or command bodies in job results. - Run self-update, dependency, and log backfill jobs must use declared capabilities, logical target keys, scoped refs, and bounded result refs. - Job payloads must not include logs, artifact chunks, raw host paths, raw credentials, direct sockets, or large inline result bodies. -- Process stdout/stderr must be redacted and sent through the asynchronous live +- Process stdout/stderr must be relayed verbatim through the asynchronous live relay rather than embedded in progress/result bodies. Relay failure or drop must not block job execution or create a local log backlog. - Job ack/progress/result/cancel/reconcile calls are lightweight lifecycle metadata and must be able to complete while artifact/file transfer work is active or retrying. diff --git a/protocol/log-ingest.md b/protocol/log-ingest.md index 58da6e9..dd94cc9 100644 --- a/protocol/log-ingest.md +++ b/protocol/log-ingest.md @@ -16,8 +16,8 @@ channel. Game plugins own durable log storage and analysis. ## Payloads -- `LogBatchIngestRequest`: run ID, session token, server instance ID, stream ID, source, sequence range, compression metadata, checksum, and bounded entries. -- `LogEntry`: sequence, timestamp, level, line, parser metadata, and redaction state. +- `LogBatchIngestRequest`: run ID, session token, server instance ID, stream ID, source, sequence range, compression metadata, checksum, and opaque entries. +- `LogEntry`: sequence, timestamp, level, opaque line, and parser metadata. - `LogBatchIngestResponse`: accepted sequence range, latest stream metadata sequence, compatibility duplicate/retry fields, and server time. - `LogStreamCursorRequest`: stream ID, sequence cursor, and limit. - `LogStreamCursorResponse`: compatibility ordered entries, next cursor, and latest stored metadata sequence. @@ -35,9 +35,10 @@ progress. A full in-memory relay queue or a failed relay request drops the current event and never blocks, retries, or changes lifecycle state. Run does not create a log spool, resend backlog, or platform delivery watermark. -Log relay payloads carry bounded redacted entries only and must not include -artifact chunks, file bodies, host paths, raw credentials, or direct socket -details. +Log relay payloads preserve supervised stdout/stderr and plugin-declared file +tail text verbatim. Run does not inspect, filter, redact, or content-truncate +those payloads; the channel still must not carry artifact chunks or direct +socket APIs. ## Browser Channel diff --git a/protocol/protected-request.md b/protocol/protected-request.md index 6310f57..3b7326e 100644 --- a/protocol/protected-request.md +++ b/protocol/protected-request.md @@ -32,7 +32,7 @@ safe result `protected_request_unknown`; other transport failures use bounded diagnostics without forwarding handler errors or response bodies. These errors affect only the current request. -Management-program stdout and stderr are bounded, redacted, and sent to the -live log relay with source `management-program` and streams +Management-program stdout and stderr are sent verbatim to the live log relay +with source `management-program` and streams `management-program.stdout` / `management-program.stderr`. They are not file execution logs and are never embedded in job result content. diff --git a/runtime/data_targets.go b/runtime/data_targets.go index 077ed7c..a2bc3e8 100644 --- a/runtime/data_targets.go +++ b/runtime/data_targets.go @@ -88,6 +88,51 @@ func (worker *Worker) materializeSQLiteProbeDataTarget(ctx context.Context, assi return matched.WorkspaceKey, nil } +func (worker *Worker) materializeSQLiteQueryDataTarget(ctx context.Context, assignment protocol.RunJobAssignment) (string, error) { + plan, _, ok, err := LoadAutonomousLifecyclePlan(worker.cfg) + if err != nil { + return assignment.TargetKey, newDataTargetMaterializeError("data_target_plan_invalid", false) + } + if !ok || plan == nil || len(plan.DataTargets) == 0 { + return assignment.TargetKey, newDataTargetMaterializeError("data_target_not_declared", false) + } + if plan.ServerInstanceID != assignment.ServerInstanceID || plan.RunEndpointID != assignment.RunEndpointID || plan.PluginID != assignment.ExecutionInput.PluginID { + return assignment.TargetKey, newDataTargetMaterializeError("data_target_scope_mismatch", false) + } + if plan.ProfileKey != "" && assignment.ExecutionInput.WorkspaceScope != "" && plan.ProfileKey != assignment.ExecutionInput.WorkspaceScope { + return assignment.TargetKey, newDataTargetMaterializeError("data_target_scope_mismatch", false) + } + var matched *protocol.RunAutonomousDataTarget + for i := range plan.DataTargets { + target := &plan.DataTargets[i] + if dataTargetMatchesSQLiteProbeAssignment(*target, assignment.TargetKey) { + matched = target + break + } + } + if matched == nil { + return assignment.TargetKey, newDataTargetMaterializeError("data_target_not_declared", false) + } + if !dataTargetSupportsPlatform(matched.Platforms, runtime.GOOS) { + return assignment.TargetKey, newDataTargetMaterializeError("data_target_platform_unsupported", false) + } + manifest, err := materializeSQLiteSnapshotDataTarget(ctx, worker.cfg.WorkspaceRoot, assignment.ServerInstanceID, assignment.ExecutionInput.WorkspaceScope, *matched, plan.RuntimeBindings) + if err != nil { + log.Printf("RUN phase=data_target.snapshot status=failed target=%s code=%s", safeOptional(matched.Key), safeOptional(dataTargetErrorCode(err))) + return assignment.TargetKey, err + } + log.Printf("RUN phase=data_target.snapshot status=complete target=%s workspaceKey=%s bytes=%d checksum=%s", safeOptional(matched.Key), safeOptional(matched.WorkspaceKey), manifest.SnapshotSizeBytes, safeOptional(manifest.SnapshotChecksum)) + return matched.WorkspaceKey, nil +} + +func sqliteQueryFailureForDataTarget(err error) LifecycleExecutionResult { + var materializeErr dataTargetMaterializeError + if errors.As(err, &materializeErr) { + return lifecycleExecutionFailure(materializeErr.code, "SQLite query data target is unavailable", materializeErr.retryable) + } + return lifecycleExecutionFailure("data_target_unavailable", "SQLite query data target is unavailable", true) +} + func dataTargetMatchesSQLiteProbeAssignment(target protocol.RunAutonomousDataTarget, assignmentTargetKey string) bool { if target.Kind != "sqlite.snapshot" || assignmentTargetKey == "" { return false diff --git a/runtime/data_targets_test.go b/runtime/data_targets_test.go index e67f09f..30b7e7a 100644 --- a/runtime/data_targets_test.go +++ b/runtime/data_targets_test.go @@ -93,6 +93,88 @@ func TestWorkerMaterializesMatchingDataTargetBeforeSQLiteProbe(t *testing.T) { } } +func TestWorkerMaterializesMatchingDataTargetBeforeSQLiteQuery(t *testing.T) { + client := newFakeWorkerClient() + cfg := workerTestConfig(t) + cfg.ServerInstanceID = "server-data-query" + cfg.PluginID = "game.example" + cfg.ComponentKind = config.PackageComponentRun + cfg.ComponentKey = "run-local" + sourceRoot := t.TempDir() + createSQLiteProbeFixture(t, filepath.Join(sourceRoot, "Saved", "SaveFiles", "current.db")) + writeAutonomousPlanFixture(t, cfg.WorkspaceRoot, cfg.ServerInstanceID, cfg.ComponentKey, protocol.RunAutonomousLifecyclePlan{ + SchemaVersion: "1", + ServerInstanceID: cfg.ServerInstanceID, + PluginID: cfg.PluginID, + PluginVersion: "1.0.0", + RunEndpointID: cfg.RunEndpointID, + ProfileKey: cfg.ComponentKey, + TargetOS: runtime.GOOS, + TargetArch: runtime.GOARCH, + TargetRelease: "run-dist-test", + Bootstrap: &protocol.RunAutonomousLifecycleAction{Action: "start", Operation: "start", Capability: protocol.RunCapabilityProcessStart, TargetKey: "actions/start.json"}, + DataTargets: []protocol.RunAutonomousDataTarget{{Key: "current-db", Kind: "sqlite.snapshot", TransportKey: "current-db", SourceRootKey: "server-root", SourcePath: "Saved/SaveFiles/current.db", WorkspaceKey: "databases/current-db", RefreshPolicy: "on-demand-snapshot", MaxBytes: 128 * 1024 * 1024, Platforms: []string{runtime.GOOS}}}, + RuntimeBindings: map[string]string{"server-root": sourceRoot}, + }) + scope, err := NewWorkspaceResolver(cfg.WorkspaceRoot).Scope(cfg.ServerInstanceID, cfg.ComponentKey) + if err != nil { + t.Fatalf("create package scope: %v", err) + } + writeSQLiteQueryAsset(t, scope, "sql/members.sql", "SELECT CAST(id AS TEXT) AS memberId, name AS displayName FROM members ORDER BY id LIMIT :limit") + worker, err := NewWorker(cfg, client) + if err != nil { + t.Fatalf("new worker: %v", err) + } + assignment := lifecycleAssignment(protocol.RunCapabilityRemoteRunDBSQLiteQuery) + assignment.ServerInstanceID = cfg.ServerInstanceID + assignment.RunEndpointID = cfg.RunEndpointID + assignment.TargetKey = "current-db" + assignment.InputRef = "input://plugin-query-poll/server-data-query/members" + assignment.ExecutionInput.WorkspaceScope = cfg.ComponentKey + assignment.ExecutionInput.PluginID = cfg.PluginID + assignment.ExecutionInput.RemoteAdapterKey = "current-db" + assignment.ExecutionInput.RemoteAdapterKind = "database" + assignment.ExecutionInput.TimeoutSeconds = 10 + assignment.ExecutionInput.Inputs = map[string]string{"sqlRef": "sql/members.sql", "maxRows": "2"} + + result := worker.executeAssignment(context.Background(), assignment) + if result.State != lifecycleResultStateSucceeded || result.ExecutionResult.Kind != "sqlite.query" || !strings.Contains(result.ExecutionResult.Content, `"displayName":"alpha"`) { + serialized, _ := json.Marshal(result) + t.Fatalf("expected worker query to materialize and return rows, got %s", serialized) + } +} + +func TestWorkerSQLiteQueryRequiresDeclaredDataTarget(t *testing.T) { + client := newFakeWorkerClient() + cfg := workerTestConfig(t) + cfg.ServerInstanceID = "server-data-query-missing" + cfg.PluginID = "game.example" + cfg.ComponentKind = config.PackageComponentRun + cfg.ComponentKey = "run-local" + writeAutonomousPlanFixture(t, cfg.WorkspaceRoot, cfg.ServerInstanceID, cfg.ComponentKey, protocol.RunAutonomousLifecyclePlan{ + SchemaVersion: "1", ServerInstanceID: cfg.ServerInstanceID, PluginID: cfg.PluginID, PluginVersion: "1.0.0", RunEndpointID: cfg.RunEndpointID, ProfileKey: cfg.ComponentKey, + TargetOS: runtime.GOOS, TargetArch: runtime.GOARCH, TargetRelease: "run-dist-test", + }) + worker, err := NewWorker(cfg, client) + if err != nil { + t.Fatalf("new worker: %v", err) + } + assignment := lifecycleAssignment(protocol.RunCapabilityRemoteRunDBSQLiteQuery) + assignment.ServerInstanceID = cfg.ServerInstanceID + assignment.RunEndpointID = cfg.RunEndpointID + assignment.TargetKey = "current-db" + assignment.InputRef = "input://plugin-query-poll/server-data-query-missing/members" + assignment.ExecutionInput.WorkspaceScope = cfg.ComponentKey + assignment.ExecutionInput.PluginID = cfg.PluginID + assignment.ExecutionInput.RemoteAdapterKey = "current-db" + assignment.ExecutionInput.RemoteAdapterKind = "database" + assignment.ExecutionInput.Inputs = map[string]string{"sqlRef": "sql/members.sql", "maxRows": "2"} + + if result := worker.executeAssignment(context.Background(), assignment); result.ErrorCode != "data_target_not_declared" { + t.Fatalf("expected missing target failure, got %+v", result) + } +} + func TestMaterializeSQLiteSnapshotDataTargetRejectsUnboundSourceRoot(t *testing.T) { target := protocol.RunAutonomousDataTarget{Key: "current-db", Kind: "sqlite.snapshot", TransportKey: "current-db", SourceRootKey: "server-root", SourcePath: "Saved/SaveFiles/current.db", WorkspaceKey: "databases/current-db", RefreshPolicy: "on-demand-snapshot", MaxBytes: 128 * 1024 * 1024} _, err := materializeSQLiteSnapshotDataTarget(context.Background(), t.TempDir(), "server-data", "run-local", target, map[string]string{}) diff --git a/runtime/lifecycle.go b/runtime/lifecycle.go index cf92abc..2ca3194 100644 --- a/runtime/lifecycle.go +++ b/runtime/lifecycle.go @@ -67,6 +67,7 @@ type LifecycleExecutor struct { localStartupDiagnostics bool protectedRequests *ProtectedRequestRegistry sqliteSchemaProbe *SQLiteSchemaProbeExecutor + sqliteQuery *SQLiteQueryExecutor metricCollector MetricCollector } @@ -118,6 +119,7 @@ func NewLifecycleExecutor(options ...LifecycleExecutorOption) LifecycleExecutor executor.fileExecutor = files } executor.sqliteSchemaProbe = NewSQLiteSchemaProbeExecutor(executor.workspaceRoot) + executor.sqliteQuery = NewSQLiteQueryExecutor(executor.workspaceRoot) return executor } @@ -389,6 +391,9 @@ func (executor LifecycleExecutor) ExecuteContext(ctx context.Context, assignment return lifecycleFailure("unsafe_lifecycle_command", err.Error()) } log.Printf("RUN phase=lifecycle.command status=starting job=%s workdir=%s command=%s timeoutMs=%d envKeys=%s", assignment.JobID, safeOptional(command.WorkDir), redactedCommandLine(command.Args), command.Timeout.Milliseconds(), envKeysSummary(command.Env, nil)) + command.OutputLine = func(stream string, line string) { + _ = executor.logSink.Append(ctx, assignment, stream, line) + } result, err := executor.supervisor.Run(ctx, command) if err != nil && ctx.Err() != nil { log.Printf("RUN phase=lifecycle.command status=cancelled job=%s error=%s", assignment.JobID, RedactText(ctx.Err().Error())) @@ -399,7 +404,9 @@ func (executor LifecycleExecutor) ExecuteContext(ctx context.Context, assignment ErrorCode: "lifecycle_cancelled", } } - executor.writeProcessLogs(ctx, assignment, result) + if !result.OutputRelayed { + executor.writeProcessLogs(ctx, assignment, result) + } if err != nil { log.Printf("RUN phase=lifecycle.command status=failed job=%s exitCode=%d error=%s", assignment.JobID, result.ExitCode, RedactText(err.Error())) return lifecycleFailure("lifecycle_process_failed", RedactText(err.Error())) @@ -519,7 +526,7 @@ func (executor LifecycleExecutor) writeProcessLogs(ctx context.Context, assignme {stream: "stdout", body: result.Stdout}, {stream: "stderr", body: result.Stderr}, } { - for _, line := range splitBoundedLines(item.body) { + for _, line := range splitRawLogLines(item.body) { _ = executor.logSink.Append(ctx, assignment, item.stream, line) } } @@ -919,12 +926,14 @@ type ProcessCommand struct { JobID string Capability string Action string + OutputLine func(string, string) } type ProcessResult struct { - ExitCode int - Stdout string - Stderr string + ExitCode int + Stdout string + Stderr string + OutputRelayed bool } type ProcessSupervisor interface { @@ -969,7 +978,7 @@ func (supervisor OSProcessSupervisor) Run(ctx context.Context, command ProcessCo err := cmd.Wait() stdoutWriter.Flush() stderrWriter.Flush() - result := ProcessResult{Stdout: RedactText(stdout.String()), Stderr: RedactText(stderr.String())} + result := ProcessResult{Stdout: stdout.String(), Stderr: stderr.String(), OutputRelayed: command.OutputLine != nil} if cmd.ProcessState != nil { result.ExitCode = cmd.ProcessState.ExitCode() } @@ -1011,7 +1020,7 @@ func (writer *lifecycleOutputWriter) Write(p []byte) (int, error) { if index < 0 { break } - line := strings.TrimRight(writer.pending[:index], "\r") + line := writer.pending[:index] writer.pending = writer.pending[index+1:] writer.logLine(line) } @@ -1021,19 +1030,17 @@ func (writer *lifecycleOutputWriter) Write(p []byte) (int, error) { func (writer *lifecycleOutputWriter) Flush() { writer.mu.Lock() defer writer.mu.Unlock() - if strings.TrimSpace(writer.pending) == "" { - writer.pending = "" + if writer.pending == "" { return } - writer.logLine(strings.TrimRight(writer.pending, "\r")) + writer.logLine(writer.pending) writer.pending = "" } func (writer *lifecycleOutputWriter) logLine(line string) { - if strings.TrimSpace(line) == "" { - return + if writer.command.OutputLine != nil { + writer.command.OutputLine(writer.stream, line) } - log.Printf("RUN phase=process.command.output status=line job=%s capability=%s action=%s stream=%s line=%q", safeOptional(writer.command.JobID), safeOptional(writer.command.Capability), safeOptional(writer.command.Action), writer.stream, RedactText(line)) } type ioLimitWriter struct { @@ -1242,6 +1249,26 @@ func splitBoundedLines(value string) []string { return out } +// splitRawLogLines only removes the newline framing used by LogEntry. It +// deliberately preserves every other byte, including blank lines and spaces. +func splitRawLogLines(value string) []string { + if value == "" { + return nil + } + lines := make([]string, 0, strings.Count(value, "\n")+1) + for start := 0; start < len(value); { + end := strings.IndexByte(value[start:], '\n') + if end < 0 { + lines = append(lines, value[start:]) + break + } + end += start + lines = append(lines, value[start:end]) + start = end + 1 + } + return lines +} + func checksumForText(value string) string { sum := sha256.Sum256([]byte(value)) return "sha256:" + hex.EncodeToString(sum[:]) diff --git a/runtime/lifecycle_test.go b/runtime/lifecycle_test.go index af5acce..ebf3b5b 100644 --- a/runtime/lifecycle_test.go +++ b/runtime/lifecycle_test.go @@ -452,7 +452,7 @@ func TestResolveRuntimeProfilesSupportsDeclaredModesAndSafeMissingKeys(t *testin } } -func TestTailDeclaredFileLogSourceUsesCheckpointAndRedaction(t *testing.T) { +func TestTailDeclaredFileLogSourceUsesCheckpointAndVerbatimOutput(t *testing.T) { root := t.TempDir() assignment := lifecycleAssignment(protocol.RunCapabilityLogsRead) serverRoot := filepath.Join(root, assignment.ServerInstanceID, "logs") @@ -472,8 +472,8 @@ func TestTailDeclaredFileLogSourceUsesCheckpointAndRedaction(t *testing.T) { if result.State != "succeeded" || !strings.Contains(result.ResultRef, "live-log-checkpoint") { t.Fatalf("expected file tail success, got %+v", result) } - if len(sink.lines) != 2 || strings.Contains(strings.Join(sink.lines, "\n"), "password=hidden") { - t.Fatalf("expected redacted tailed lines, got %+v", sink.lines) + if len(sink.lines) != 2 || strings.Join(sink.lines, "\n") != "latest-log:first line\nlatest-log:password=hidden" { + t.Fatalf("expected verbatim tailed lines, got %+v", sink.lines) } checkpoint := store.GetLogCheckpoint("latest-log") if checkpoint.Offset == 0 || checkpoint.Sequence != 2 || strings.Contains(RedactedLogCheckpointSummary(checkpoint), "/Users/") { @@ -490,6 +490,36 @@ func TestTailDeclaredFileLogSourceUsesCheckpointAndRedaction(t *testing.T) { } } +func TestTailDeclaredFileLogSourceDoesNotLimitOrRewriteOutput(t *testing.T) { + root := t.TempDir() + assignment := lifecycleAssignment(protocol.RunCapabilityLogsRead) + serverRoot := filepath.Join(root, assignment.ServerInstanceID, "logs") + if err := os.MkdirAll(serverRoot, 0o755); err != nil { + t.Fatalf("create logs dir: %v", err) + } + longLine := "password=visible\r" + strings.Repeat("x", maxLifecycleOutputBytes+1) + logPath := filepath.Join(serverRoot, "current.log") + if err := os.WriteFile(logPath, []byte(longLine+"\n\nfinal"), 0o644); err != nil { + t.Fatalf("write log file: %v", err) + } + store := NewMemoryLogCheckpointStore() + sink := &recordingLogSink{} + source := RuntimeLogSource{Key: "current-log", Kind: "file.tail", TargetKey: "logs/current.log", StreamKey: "current-log", CursorKind: "offset"} + + result := TailDeclaredFileLogSource(context.Background(), root, assignment, source, sink, store) + + if result.State != lifecycleResultStateSucceeded || len(sink.lines) != 3 { + t.Fatalf("expected all unbounded log entries, result=%+v lines=%d", result, len(sink.lines)) + } + if sink.lines[0] != "current-log:"+longLine || sink.lines[1] != "current-log:" || sink.lines[2] != "current-log:final" { + t.Fatalf("expected byte-for-byte log payloads, got lengths=%d,%d,%d", len(sink.lines[0]), len(sink.lines[1]), len(sink.lines[2])) + } + checkpoint := store.GetLogCheckpoint(source.Key) + if checkpoint.Offset != int64(len(longLine+"\n\nfinal")) || checkpoint.Sequence != 3 { + t.Fatalf("expected complete checkpoint after unbounded tail, got %+v", checkpoint) + } +} + func TestLifecycleExecutorExecutesDeclaredLogBackfillTail(t *testing.T) { root := t.TempDir() assignment := lifecycleAssignment(protocol.RunCapabilityLogsBackfill) @@ -523,6 +553,22 @@ func (supervisor *recordingSupervisor) Run(_ context.Context, command ProcessCom return ProcessResult{ExitCode: 0, Stdout: "recorded\n"}, nil } +func TestOSProcessSupervisorRelaysRawCarriageReturn(t *testing.T) { + var output []string + result, err := (OSProcessSupervisor{}).Run(context.Background(), ProcessCommand{ + Args: []string{"sh", "-c", "printf 'raw\\r\\n'"}, + OutputLine: func(stream string, line string) { + output = append(output, stream+":"+line) + }, + }) + if err != nil || result.ExitCode != 0 { + t.Fatalf("run process: result=%+v err=%v", result, err) + } + if len(output) != 1 || output[0] != "stdout:raw\r" { + t.Fatalf("expected raw CR to be relayed, got %+v", output) + } +} + type recordingLogSink struct { mu sync.Mutex lines []string diff --git a/runtime/log_sources.go b/runtime/log_sources.go index f686681..0c114d8 100644 --- a/runtime/log_sources.go +++ b/runtime/log_sources.go @@ -1,8 +1,10 @@ package runtime import ( + "bufio" "context" "fmt" + "io" "net/url" "os" "strings" @@ -80,26 +82,32 @@ func TailDeclaredFileLogSource(ctx context.Context, workspaceRoot string, assign return lifecycleFailure("log_source_seek_failed", err.Error()) } } - body := make([]byte, maxLifecycleOutputBytes) - n, err := file.Read(body) - if err != nil && n == 0 { - return LifecycleExecutionResult{ - State: lifecycleResultStateSucceeded, - Progress: protocol.RunJobProgressReport{Percent: 100, Message: "live log checkpoint unchanged"}, - ResultRef: fmt.Sprintf("artifact://jobs/%s/live-log-checkpoint", url.PathEscape(assignment.JobID)), - Message: "live log source had no new lines", + reader := bufio.NewReader(file) + for { + // A newline only frames an entry. Every other byte, including CR, blank + // lines, and arbitrarily long output, remains untouched. + line, readErr := reader.ReadString('\n') + if len(line) > 0 { + checkpoint.Sequence++ + if err := sink.Append(ctx, assignment, source.StreamKey, strings.TrimSuffix(line, "\n")); err != nil { + return lifecycleFailure("log_source_sink_failed", err.Error()) + } + checkpoint.SourceKey = source.Key + checkpoint.Offset += int64(len(line)) + checkpoint.CursorRef = fmt.Sprintf("artifact://jobs/%s/live-log-checkpoint", url.PathEscape(assignment.JobID)) + store.PutLogCheckpoint(checkpoint) } - } - for _, line := range splitBoundedLines(string(body[:n])) { - checkpoint.Sequence++ - if err := sink.Append(ctx, assignment, source.StreamKey, line); err != nil { - return lifecycleFailure("log_source_sink_failed", err.Error()) + if readErr == nil { + continue } + if readErr == io.EOF { + break + } + return lifecycleFailure("log_source_read_failed", readErr.Error()) + } + if checkpoint.CursorRef == "" { + checkpoint.CursorRef = fmt.Sprintf("artifact://jobs/%s/live-log-checkpoint", url.PathEscape(assignment.JobID)) } - checkpoint.SourceKey = source.Key - checkpoint.Offset += int64(n) - checkpoint.CursorRef = fmt.Sprintf("artifact://jobs/%s/live-log-checkpoint", url.PathEscape(assignment.JobID)) - store.PutLogCheckpoint(checkpoint) return LifecycleExecutionResult{ State: lifecycleResultStateSucceeded, Progress: protocol.RunJobProgressReport{Percent: 100, Message: "live log checkpoint updated"}, diff --git a/runtime/process_supervisor.go b/runtime/process_supervisor.go index 5f6d2e8..409c9a6 100644 --- a/runtime/process_supervisor.go +++ b/runtime/process_supervisor.go @@ -535,20 +535,17 @@ func (supervisor *OSManagedProcessSupervisor) tailOutput(ctx context.Context, ta if len(line) > 0 { startOffset := offset endOffset := offset + int64(len(line)) - text := strings.TrimSpace(line) - if text != "" { - for { - if sinkErr := sink(identity, ManagedProcessLine{Text: text, StartOffset: startOffset, EndOffset: endOffset}); sinkErr == nil { - log.Printf("RUN phase=process.managed.output status=line job=%s pid=%d stream=%s line=%q", safeOptional(identity.JobID), identity.PID, stream, RedactText(text)) - break - } else { - log.Printf("RUN phase=process.managed.output status=sink_retry job=%s pid=%d stream=%s error=%s", safeOptional(identity.JobID), identity.PID, stream, RedactText(sinkErr.Error())) - } - select { - case <-ctx.Done(): - return - case <-time.After(managedProcessOutputRetryDelay): - } + text := strings.TrimSuffix(line, "\n") + for { + if sinkErr := sink(identity, ManagedProcessLine{Text: text, StartOffset: startOffset, EndOffset: endOffset}); sinkErr == nil { + break + } else { + log.Printf("RUN phase=process.managed.output status=sink_retry job=%s pid=%d stream=%s error=%s", safeOptional(identity.JobID), identity.PID, stream, RedactText(sinkErr.Error())) + } + select { + case <-ctx.Done(): + return + case <-time.After(managedProcessOutputRetryDelay): } } for { diff --git a/runtime/protected_request.go b/runtime/protected_request.go index 3c37efc..0b8d8bc 100644 --- a/runtime/protected_request.go +++ b/runtime/protected_request.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "strings" "sync" "time" @@ -154,48 +153,13 @@ func validProtectedRequestKind(kind string) bool { func (executor LifecycleExecutor) writeProtectedProgramLogs(ctx context.Context, assignment protocol.RunJobAssignment, outcome ProtectedRequestOutcome) { for _, item := range []struct{ stream, body string }{{"management-program.stdout", outcome.Stdout}, {"management-program.stderr", outcome.Stderr}} { for _, line := range splitProtectedProgramLines(item.body) { - _ = executor.logSink.Append(ctx, assignment, item.stream, sanitizeProtectedProgramLogLine(line)) + _ = executor.logSink.Append(ctx, assignment, item.stream, line) } } } func splitProtectedProgramLines(value string) []string { - value = strings.ToValidUTF8(strings.TrimSpace(value), "") - if len(value) > maxLifecycleOutputBytes { - value = strings.ToValidUTF8(value[:maxLifecycleOutputBytes], "") - } - lines := strings.Split(value, "\n") - bounded := make([]string, 0, len(lines)) - for _, line := range lines { - line = strings.TrimRight(line, "\r") - if strings.TrimSpace(line) != "" { - bounded = append(bounded, line) - } - } - return bounded -} - -func sanitizeProtectedProgramLogLine(line string) string { - if containsProtectedProgramPrivateText(line) { - return "[redacted protected management-program output]" - } - return RedactText(line) -} - -func containsProtectedProgramPrivateText(value string) bool { - lower := strings.ToLower(value) - for _, marker := range []string{"password=", "password:", "secret=", "secret:", "token=", "token:", "credential", "bearer ", "api_key", "apikey", "dsn=", "path=", "file=", "database=", "://", "mysql:", "postgres:", "sqlite:", "file:", "socket", "named pipe"} { - if strings.Contains(lower, marker) { - return true - } - } - for _, field := range strings.Fields(value) { - field = strings.Trim(field, "\"'()[]{}<>,;") - if strings.HasPrefix(field, "/") || strings.HasPrefix(field, "./") || strings.HasPrefix(field, "../") || strings.HasPrefix(field, `\\`) || len(field) >= 3 && ((field[0] >= 'a' && field[0] <= 'z') || (field[0] >= 'A' && field[0] <= 'Z')) && field[1] == ':' && (field[2] == '/' || field[2] == '\\') { - return true - } - } - return false + return splitRawLogLines(value) } func protectedRequestFailure(capability string, status string, code string) LifecycleExecutionResult { diff --git a/runtime/protected_request_test.go b/runtime/protected_request_test.go index 5567123..d185c2c 100644 --- a/runtime/protected_request_test.go +++ b/runtime/protected_request_test.go @@ -57,22 +57,23 @@ func TestWorkerExecutesFencedProtectedProgramAndSpoolsDedicatedLogs(t *testing.T if err != nil || entryCount != 5 { t.Fatalf("expected five program log lines: batches=%+v err=%v", batches, err) } - redactedEntries := 0 + logLines := []string{} for _, batch := range batches { if batch.Source != "management-program" || batch.Source == "file" || batch.Source == "process" { t.Fatalf("program output used wrong log source: %+v", batch) } for _, entry := range batch.Entries { if entry.Redacted { - redactedEntries++ - } - if strings.Contains(entry.Line, "password=hidden") || strings.Contains(entry.Line, "/Users/") || strings.Contains(entry.Line, "://") { - t.Fatalf("program log leaked protected output: %+v", entry) + t.Fatalf("program log must remain verbatim, got %+v", entry) } + logLines = append(logLines, entry.Line) } } - if redactedEntries != 3 { - t.Fatalf("expected three explicitly redacted private lines, got %d", redactedEntries) + joinedLogs := strings.Join(logLines, "\n") + for _, expected := range []string{"password=hidden", "/Users/private/scum.ini", "mysql://private"} { + if !strings.Contains(joinedLogs, expected) { + t.Fatalf("program log did not preserve %q: %q", expected, joinedLogs) + } } serialized, err := json.Marshal([]any{client.resultRequests, client.protectedRequests, worker.journal.ActiveJobs()}) if err != nil { @@ -166,8 +167,8 @@ func TestUnknownProtectedProgramKeepsSafeDiagnosticInProgramLogOnly(t *testing.T if result.ErrorCode != "protected_request_unknown" || result.State != lifecycleResultStateFailed { t.Fatalf("unexpected unknown program result: %+v", result) } - if len(sink.lines) != 1 || !strings.HasPrefix(sink.lines[0], "management-program.stderr:") || strings.Contains(sink.lines[0], "/private/") || !strings.Contains(sink.lines[0], "[redacted protected") { - t.Fatalf("unknown program diagnostic was not safely channelized: %+v", sink.lines) + if len(sink.lines) != 1 || sink.lines[0] != "management-program.stderr:unknown field database=/private/scum.db" { + t.Fatalf("unknown program diagnostic was not faithfully channelized: %+v", sink.lines) } } diff --git a/runtime/sqlite_query.go b/runtime/sqlite_query.go new file mode 100644 index 0000000..81d1ad5 --- /dev/null +++ b/runtime/sqlite_query.go @@ -0,0 +1,312 @@ +package runtime + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "os" + "regexp" + "strconv" + "strings" + "time" + + "browser.local/run/protocol" +) + +const ( + defaultSQLiteQueryMaxRows = 500 + maxSQLiteQueryRows = 500 + maxSQLiteQueryColumns = 128 + maxSQLiteQuerySQLBytes = 64 * 1024 + maxSQLiteQueryResultBytes = 64 * 1024 + maxSQLiteQueryTimeout = 60 * time.Second +) + +var sqliteNamedParameterPattern = regexp.MustCompile(`[:@$]([A-Za-z_][A-Za-z0-9_]*)`) + +// SQLiteQueryExecutor executes a package-provided, read-only SQL asset against +// a declared SQLite snapshot inside the server's scoped Run workspace. +type SQLiteQueryExecutor struct{ resolver WorkspaceResolver } + +func NewSQLiteQueryExecutor(workspaceRoot string) *SQLiteQueryExecutor { + return &SQLiteQueryExecutor{resolver: NewWorkspaceResolver(workspaceRoot)} +} + +func (executor *SQLiteQueryExecutor) Execute(ctx context.Context, assignment protocol.RunJobAssignment) LifecycleExecutionResult { + if err := validateSQLiteQueryAssignment(assignment); err != nil { + return sqliteQueryFailure("invalid_request", false) + } + queryAsset, maxRows, err := sqliteQueryRequest(assignment.ExecutionInput.Inputs) + if err != nil { + return sqliteQueryFailure("invalid_request", false) + } + scope, err := executor.resolver.Scope(assignment.ServerInstanceID, assignment.ExecutionInput.WorkspaceScope) + if err != nil { + return sqliteQueryFailure("target_unavailable", false) + } + databasePath, err := executor.resolver.ExistingTarget(scope, assignment.TargetKey) + if err != nil { + return sqliteQueryFailure("target_unavailable", false) + } + sqlPath, err := executor.resolver.ExistingTarget(scope, queryAsset) + if err != nil { + return sqliteQueryFailure("sql_asset_unavailable", false) + } + statement, err := readSQLiteQueryAsset(sqlPath) + if err != nil { + return sqliteQueryFailure(sqliteQueryAssetErrorCode(err), false) + } + if !isReadOnlySQLiteStatement(statement) { + return sqliteQueryFailure("sql_asset_not_read_only", false) + } + + queryCtx, cancel := sqliteQueryContext(ctx, assignment.ExecutionInput.TimeoutSeconds) + defer cancel() + database, err := sql.Open(sqliteSchemaProbeDriver, "file:"+databasePath+"?mode=ro") + if err != nil { + return sqliteQueryFailure("sqlite_open_failed", true) + } + defer database.Close() + database.SetMaxOpenConns(1) + if _, err := database.ExecContext(queryCtx, "PRAGMA query_only = ON"); err != nil { + return sqliteQueryFailure(sqliteQueryErrorCode(queryCtx, err), true) + } + rows, err := database.QueryContext(queryCtx, statement, sqliteQueryArguments(statement, maxRows)...) + if err != nil { + return sqliteQueryFailure(sqliteQueryErrorCode(queryCtx, err), true) + } + defer rows.Close() + content, err := collectSQLiteQueryRows(rows, maxRows) + if err != nil { + return sqliteQueryFailure(sqliteQueryErrorCode(queryCtx, err), false) + } + if err := rows.Err(); err != nil { + return sqliteQueryFailure(sqliteQueryErrorCode(queryCtx, err), true) + } + return LifecycleExecutionResult{ + State: lifecycleResultStateSucceeded, + Progress: protocol.RunJobProgressReport{Percent: 100, Message: "SQLite query completed"}, + Message: "SQLite query completed", + ExecutionResult: protocol.RunJobExecutionResult{ + Kind: "sqlite.query", + Checksum: checksumForText(content), + SizeBytes: int64(len([]byte(content))), + Summary: "bounded query-only SQLite rows", + Content: content, + }, + } +} + +func validateSQLiteQueryAssignment(assignment protocol.RunJobAssignment) error { + if assignment.Capability != protocol.RunCapabilityRemoteRunDBSQLiteQuery || assignment.ServerInstanceID == "" || assignment.ExecutionInput.WorkspaceScope == "" { + return fmt.Errorf("SQLite query requires a scoped database job") + } + if err := protocol.ValidateRunJobAssignment(assignment); err != nil { + return err + } + if assignment.ExecutionInput.RemoteAdapterKind != "" && assignment.ExecutionInput.RemoteAdapterKind != "database" { + return fmt.Errorf("SQLite query requires the database adapter") + } + if assignment.ExecutionInput.RemoteAdapterKey != "" && !protocol.ValidLogicalFileKey(assignment.ExecutionInput.RemoteAdapterKey) { + return fmt.Errorf("SQLite query adapter key is invalid") + } + return nil +} + +func sqliteQueryRequest(inputs map[string]string) (string, int, error) { + if len(inputs) == 0 { + return "", 0, fmt.Errorf("SQLite query inputs are missing") + } + queryAsset := strings.TrimSpace(inputs["sqlRef"]) + if queryAsset == "" || queryAsset != inputs["sqlRef"] || !protocol.ValidLogicalFileKey(queryAsset) || !strings.HasPrefix(queryAsset, "sql/") || !strings.HasSuffix(strings.ToLower(queryAsset), ".sql") { + return "", 0, fmt.Errorf("SQLite query SQL asset is invalid") + } + maxRows := defaultSQLiteQueryMaxRows + if rawMaxRows, exists := inputs["maxRows"]; exists { + parsed, err := strconv.Atoi(rawMaxRows) + if err != nil || parsed < 1 || parsed > maxSQLiteQueryRows { + return "", 0, fmt.Errorf("SQLite query row limit is invalid") + } + maxRows = parsed + } + return queryAsset, maxRows, nil +} + +func readSQLiteQueryAsset(path string) (string, error) { + info, err := os.Stat(path) + if err != nil || !info.Mode().IsRegular() || info.Size() < 1 || info.Size() > maxSQLiteQuerySQLBytes { + return "", errSQLiteQueryAssetInvalid + } + body, err := os.ReadFile(path) + if err != nil || len(body) == 0 || len(body) > maxSQLiteQuerySQLBytes { + return "", errSQLiteQueryAssetInvalid + } + statement := strings.TrimSpace(string(body)) + if statement == "" || strings.ContainsRune(statement, '\x00') { + return "", errSQLiteQueryAssetInvalid + } + return statement, nil +} + +var errSQLiteQueryAssetInvalid = errors.New("SQLite query asset is invalid") + +func sqliteQueryAssetErrorCode(err error) string { + if errors.Is(err, errSQLiteQueryAssetInvalid) { + return "sql_asset_invalid" + } + return "sql_asset_unavailable" +} + +func isReadOnlySQLiteStatement(statement string) bool { + trimmed := strings.TrimSpace(statement) + trimmed = strings.TrimSuffix(trimmed, ";") + if trimmed == "" || strings.Contains(trimmed, ";") { + return false + } + parts := strings.Fields(strings.ToUpper(strings.TrimSpace(stripSQLiteLeadingComments(trimmed)))) + if len(parts) == 0 { + return false + } + switch parts[0] { + case "SELECT", "WITH": + return true + case "EXPLAIN": + return len(parts) > 1 && (parts[1] == "SELECT" || len(parts) > 2 && parts[1] == "QUERY" && parts[2] == "PLAN") + default: + return false + } +} + +func stripSQLiteLeadingComments(statement string) string { + for { + statement = strings.TrimSpace(statement) + switch { + case strings.HasPrefix(statement, "--"): + if lineEnd := strings.IndexByte(statement, '\n'); lineEnd >= 0 { + statement = statement[lineEnd+1:] + continue + } + return "" + case strings.HasPrefix(statement, "/*"): + commentEnd := strings.Index(statement[2:], "*/") + if commentEnd < 0 { + return "" + } + statement = statement[commentEnd+4:] + continue + } + return statement + } +} + +func sqliteQueryContext(ctx context.Context, timeoutSeconds int) (context.Context, context.CancelFunc) { + timeout := maxSQLiteQueryTimeout + if timeoutSeconds > 0 && time.Duration(timeoutSeconds)*time.Second < timeout { + timeout = time.Duration(timeoutSeconds) * time.Second + } + return context.WithTimeout(ctx, timeout) +} + +func sqliteQueryArguments(statement string, maxRows int) []any { + seen := map[string]struct{}{} + arguments := make([]any, 0) + for _, match := range sqliteNamedParameterPattern.FindAllStringSubmatch(statement, -1) { + name := match[1] + if _, exists := seen[name]; exists { + continue + } + seen[name] = struct{}{} + value := any(nil) + if name == "limit" { + value = maxRows + } + arguments = append(arguments, sql.Named(name, value)) + } + return arguments +} + +func collectSQLiteQueryRows(rows *sql.Rows, maxRows int) (string, error) { + columns, err := rows.Columns() + if err != nil { + return "", err + } + if len(columns) == 0 || len(columns) > maxSQLiteQueryColumns || !uniqueSQLiteQueryColumns(columns) { + return "", errSQLiteQueryResultLimit + } + result := struct { + Rows []map[string]any `json:"rows"` + }{Rows: make([]map[string]any, 0, maxRows)} + for rows.Next() { + if len(result.Rows) >= maxRows { + return "", errSQLiteQueryResultLimit + } + values := make([]any, len(columns)) + pointers := make([]any, len(values)) + for index := range values { + pointers[index] = &values[index] + } + if err := rows.Scan(pointers...); err != nil { + return "", err + } + row := make(map[string]any, len(columns)) + for index, column := range columns { + row[column] = sqliteQueryValue(values[index]) + } + result.Rows = append(result.Rows, row) + payload, err := json.Marshal(result) + if err != nil || len(payload) > maxSQLiteQueryResultBytes { + return "", errSQLiteQueryResultLimit + } + } + payload, err := json.Marshal(result) + if err != nil || len(payload) > maxSQLiteQueryResultBytes { + return "", errSQLiteQueryResultLimit + } + return string(payload), nil +} + +var errSQLiteQueryResultLimit = errors.New("SQLite query result exceeds limit") + +func uniqueSQLiteQueryColumns(columns []string) bool { + seen := map[string]struct{}{} + for _, column := range columns { + if strings.TrimSpace(column) == "" { + return false + } + if _, exists := seen[column]; exists { + return false + } + seen[column] = struct{}{} + } + return true +} + +func sqliteQueryValue(value any) any { + if bytes, ok := value.([]byte); ok { + return append([]byte(nil), bytes...) + } + return value +} + +func sqliteQueryErrorCode(ctx context.Context, err error) string { + if errors.Is(err, errSQLiteQueryResultLimit) { + return "result_limit_exceeded" + } + if errors.Is(ctx.Err(), context.Canceled) { + return "cancelled" + } + if errors.Is(ctx.Err(), context.DeadlineExceeded) { + return "timeout" + } + lower := strings.ToLower(fmt.Sprint(err)) + if strings.Contains(lower, "locked") || strings.Contains(lower, "busy") { + return "database_busy" + } + return "sqlite_read_failed" +} + +func sqliteQueryFailure(code string, retryable bool) LifecycleExecutionResult { + return lifecycleExecutionFailure(code, "SQLite query failed", retryable) +} diff --git a/runtime/sqlite_query_test.go b/runtime/sqlite_query_test.go new file mode 100644 index 0000000..a31ead2 --- /dev/null +++ b/runtime/sqlite_query_test.go @@ -0,0 +1,103 @@ +package runtime + +import ( + "context" + "database/sql" + "encoding/json" + "os" + "path/filepath" + "testing" + + "browser.local/run/protocol" +) + +func TestSQLiteQueryReturnsDeclaredRowsVerbatim(t *testing.T) { + root := t.TempDir() + assignment := sqliteQueryAssignment() + scope, err := NewWorkspaceResolver(root).Scope(assignment.ServerInstanceID, assignment.ExecutionInput.WorkspaceScope) + if err != nil { + t.Fatalf("create scope: %v", err) + } + createSQLiteQueryFixture(t, filepath.Join(scope, assignment.TargetKey)) + writeSQLiteQueryAsset(t, scope, assignment.ExecutionInput.Inputs["sqlRef"], ` +SELECT id AS steamId, name AS displayName, note AS originalText +FROM users +WHERE (:steamId IS NULL OR id = :steamId) +ORDER BY id +LIMIT :limit; +`) + + result := NewSQLiteQueryExecutor(root).Execute(context.Background(), assignment) + if result.State != lifecycleResultStateSucceeded || result.ExecutionResult.Kind != "sqlite.query" || result.ExecutionResult.Checksum == "" || result.ExecutionResult.SizeBytes <= 0 { + t.Fatalf("expected successful SQLite rows, got %+v", result) + } + var payload struct { + Rows []map[string]any `json:"rows"` + } + if err := json.Unmarshal([]byte(result.ExecutionResult.Content), &payload); err != nil { + t.Fatalf("decode rows: %v", err) + } + if len(payload.Rows) != 2 || payload.Rows[0]["steamId"] != "steam-1" || payload.Rows[0]["originalText"] != "password=opaque C:/game/users.db" { + t.Fatalf("SQLite rows were not faithfully returned: %+v", payload.Rows) + } +} + +func TestSQLiteQueryRejectsUnsafeAssetsAndLimitOverrun(t *testing.T) { + root := t.TempDir() + assignment := sqliteQueryAssignment() + scope, err := NewWorkspaceResolver(root).Scope(assignment.ServerInstanceID, assignment.ExecutionInput.WorkspaceScope) + if err != nil { + t.Fatalf("create scope: %v", err) + } + createSQLiteQueryFixture(t, filepath.Join(scope, assignment.TargetKey)) + writeSQLiteQueryAsset(t, scope, assignment.ExecutionInput.Inputs["sqlRef"], "SELECT id AS steamId FROM users ORDER BY id") + + unsafe := assignment + unsafe.ExecutionInput.Inputs = map[string]string{"sqlRef": "../outside.sql", "maxRows": "2"} + if result := NewSQLiteQueryExecutor(root).Execute(context.Background(), unsafe); result.ErrorCode != "invalid_request" { + t.Fatalf("expected unsafe asset rejection, got %+v", result) + } + limited := assignment + limited.ExecutionInput.Inputs = map[string]string{"sqlRef": assignment.ExecutionInput.Inputs["sqlRef"], "maxRows": "1"} + if result := NewSQLiteQueryExecutor(root).Execute(context.Background(), limited); result.ErrorCode != "result_limit_exceeded" { + t.Fatalf("expected result limit failure instead of truncation, got %+v", result) + } +} + +func sqliteQueryAssignment() protocol.RunJobAssignment { + assignment := lifecycleAssignment(protocol.RunCapabilityRemoteRunDBSQLiteQuery) + assignment.TargetKey = "databases/current.db" + assignment.InputRef = "input://plugin-query-poll/server-1/users" + assignment.ExecutionInput.WorkspaceScope = "profile-default" + assignment.ExecutionInput.RemoteAdapterKey = "current-db" + assignment.ExecutionInput.RemoteAdapterKind = "database" + assignment.ExecutionInput.TimeoutSeconds = 10 + assignment.ExecutionInput.Inputs = map[string]string{"sqlRef": "sql/users.sql", "maxRows": "2"} + return assignment +} + +func createSQLiteQueryFixture(t *testing.T, path string) { + t.Helper() + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + t.Fatal(err) + } + database, err := sql.Open(sqliteSchemaProbeDriver, path) + if err != nil { + t.Fatal(err) + } + defer database.Close() + if _, err := database.Exec(`CREATE TABLE users (id TEXT PRIMARY KEY, name TEXT NOT NULL, note TEXT); INSERT INTO users(id, name, note) VALUES ('steam-1', 'Ada', 'password=opaque C:/game/users.db'), ('steam-2', 'Lin', 'unchanged');`); err != nil { + t.Fatal(err) + } +} + +func writeSQLiteQueryAsset(t *testing.T, scope string, asset string, body string) { + t.Helper() + path := filepath.Join(scope, filepath.FromSlash(asset)) + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatal(err) + } +} diff --git a/runtime/worker.go b/runtime/worker.go index 6035102..3f13e06 100644 --- a/runtime/worker.go +++ b/runtime/worker.go @@ -735,6 +735,18 @@ func (worker *Worker) executeAssignment(ctx context.Context, assignment protocol } return worker.executor.sqliteSchemaProbe.Execute(ctx, assignment) } + if assignment.Capability == protocol.RunCapabilityRemoteRunDBSQLiteQuery { + log.Printf("RUN phase=job.dispatch status=selected job=%s executor=sqlite_query", assignment.JobID) + if worker.executor.sqliteQuery == nil { + return lifecycleFailure("sqlite_query_unavailable", "SQLite query executor is unavailable") + } + targetKey, err := worker.materializeSQLiteQueryDataTarget(ctx, assignment) + if err != nil { + return sqliteQueryFailureForDataTarget(err) + } + assignment.TargetKey = targetKey + return worker.executor.sqliteQuery.Execute(ctx, assignment) + } if isSupportedLifecycleCapability(assignment.Capability) { log.Printf("RUN phase=job.dispatch status=selected job=%s executor=lifecycle", assignment.JobID) return worker.executor.ExecuteContext(ctx, assignment) @@ -1484,8 +1496,6 @@ func (sink *LiveLogSink) append(assignment protocol.RunJobAssignment, stream str if sink == nil { return nil } - redactedLine := RedactText(line) - redacted := line != redactedLine || strings.HasPrefix(redactedLine, "[redacted protected ") streamKey := declaredProcessStreamKey(assignment, stream) logStreamID := logStreamIDForAssignment(assignment, streamKey) sink.mu.Lock() @@ -1514,7 +1524,7 @@ func (sink *LiveLogSink) append(assignment protocol.RunJobAssignment, stream str FirstSeq: sequence, LastSeq: sequence, Compression: "none", - Entries: []protocol.LogEntry{{Seq: sequence, Timestamp: time.Now().UTC(), Level: "info", Line: redactedLine, Redacted: redacted}}, + Entries: []protocol.LogEntry{{Seq: sequence, Timestamp: time.Now().UTC(), Level: "info", Line: line, Redacted: false}}, } batch.Checksum, _ = checksumForLogEntries(batch.Entries) select { @@ -1564,11 +1574,9 @@ func (sink *SpoolLogSink) AppendWithCursor(ctx context.Context, assignment proto func (sink *SpoolLogSink) append(ctx context.Context, assignment protocol.RunJobAssignment, stream string, line string, cursor *spool.LogSourceCursor) error { sink.mu.Lock() defer sink.mu.Unlock() - redactedLine := RedactText(line) - redacted := line != redactedLine || strings.HasPrefix(redactedLine, "[redacted protected ") streamKey := declaredProcessStreamKey(assignment, stream) logStreamID := logStreamIDForAssignment(assignment, streamKey) - entry := protocol.LogEntry{Timestamp: time.Now().UTC(), Level: "info", Line: redactedLine, Redacted: redacted} + entry := protocol.LogEntry{Timestamp: time.Now().UTC(), Level: "info", Line: line, Redacted: false} source := "process" if strings.HasPrefix(stream, "management-program.") { source = "management-program" @@ -1640,7 +1648,7 @@ type QueueArtifactHook struct { func (hook QueueArtifactHook) QueueLifecycleResult(_ context.Context, assignment protocol.RunJobAssignment, result ProcessResult) (string, error) { ref := fmt.Sprintf("artifact://jobs/%s/lifecycle-result", assignment.JobID) - payload := []byte(RedactText(result.Stdout + result.Stderr)) + payload := []byte(result.Stdout + result.Stderr) if len(payload) == 0 { payload = []byte("lifecycle result metadata") } diff --git a/runtime/worker_test.go b/runtime/worker_test.go index 739ef03..20fccb6 100644 --- a/runtime/worker_test.go +++ b/runtime/worker_test.go @@ -392,7 +392,7 @@ func TestWorkerSpoolHooksUseRegisteredSession(t *testing.T) { if err != nil { t.Fatalf("pending logs: %v", err) } - if len(logs) != 1 || logs[0].RunEndpointID != "run-test" || logs[0].SessionToken != "session-token" || logs[0].StreamKey != "game.console.stdout" || containsText(logs[0].Entries[0].Line, "password=hidden") { + if len(logs) != 1 || logs[0].RunEndpointID != "run-test" || logs[0].SessionToken != "session-token" || logs[0].StreamKey != "game.console.stdout" || logs[0].Entries[0].Line != "started password=hidden" || logs[0].Entries[0].Redacted { t.Fatalf("unexpected spooled logs: %+v", logs) } chunks, err := artifactQueue.Pending()