From 8d4cc39a3b22d7a0c3ed177b8f1220a65c70ac9e Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Mon, 10 Aug 2026 18:41:42 +0800 Subject: [PATCH] Add controlled-session framed protocol Define the versioned typed frame codec and shared lifecycle result types used between the host supervisor and controller. Give every V1 frame and event an explicit stable wire assignment while preserving terminal bytes exactly and validating structured payloads, direction, dimensions, endpoint identifiers, protocol version, duplicate JSON fields, and bounded payload length. Treat controller completion as finalization rather than a termination cause, and model bounded workload-output finalization explicitly in the opened contract, events, and final result. Align the controlled-session design and follow-up networking backlog with the temporary coarse lease-network grant and the planned gateway-based direct path. Add stable-wire, round-trip, malformed-frame, direction-confusion, duplicate-field, binary-data, output-finalization, and fuzz-seed coverage. --- docs/BACKLOG.md | 36 +- docs/CONTROLLED_SESSION_DESIGN.md | 242 ++++---- internal/controlledsession/authorization.go | 15 +- .../controlledsession/authorization_test.go | 3 +- internal/controlledsession/model.go | 197 +++++++ internal/controlledsession/protocol.go | 557 ++++++++++++++++++ internal/controlledsession/protocol_test.go | 313 ++++++++++ 7 files changed, 1210 insertions(+), 153 deletions(-) create mode 100644 internal/controlledsession/model.go create mode 100644 internal/controlledsession/protocol.go create mode 100644 internal/controlledsession/protocol_test.go diff --git a/docs/BACKLOG.md b/docs/BACKLOG.md index b5d84930..dcff50ac 100644 --- a/docs/BACKLOG.md +++ b/docs/BACKLOG.md @@ -183,10 +183,9 @@ This file is the day-to-day queue for design and implementation gaps. that do not belong in blueprints, staging overrides, or installation state. Do not add a general configuration file until the accumulated use cases justify its scope, precedence, user/system ownership, validation, - and portability. Initial potential use case: overriding the otherwise - fixed host-owned limits for controlled-session endpoint streams and - connection-open rates. Also record host-owned DNS resolver configuration - used to provide DNS under the coarse application network grants. The + and portability. Initial potential use case: host-owned DNS resolver + configuration used to provide DNS under the coarse application network + grants. The default local-capable path should use the host's configured resolver so VPN and split-DNS behavior remains available; the public-only path should use the built-in Google Public DNS profile (`8.8.8.8`, `8.8.4.4`). Allow @@ -195,17 +194,28 @@ This file is the day-to-day queue for design and implementation gaps. - [ ] `P2` Design and implement a Reploy userland L3 policy gateway. Keep this separate from the initial public/local kill switches and controlled - sessions. Define a capability-free application network namespace, a - one-shot route initializer, an isolated data path whose only peer is the - gateway, private gateway control, root-resistant route invariants, - IPv4/IPv6 and DNS policy, destination and port grants, auditing, resource + sessions. Define separate controller and workload network identities, a + capability-free application network namespace, a one-shot route + initializer, an isolated data path whose only peer is the gateway, + private gateway control, root-resistant route invariants, IPv4/IPv6 and + DNS policy, directional destination and port grants, auditing, resource limits, failure behavior, reconciliation, and Docker/Podman plus Desktop integration. Treat native engine primitives as fast paths rather than - exposing backend network modes as product policy. Replace or mediate the - initial controlled-session host-loopback endpoint publication so only the - lease-owned Host Reploy operation can reach the recorded application; - include multi-user-host tests proving unrelated local processes cannot - bypass the session endpoint grant. Replace the temporary, discouraged + exposing backend network modes as product policy. + + Make the gateway the target controlled-session endpoint policy. Permit + native TCP from the controller only to declared workload addresses and + ports; deny workload-initiated access to the controller, undeclared + workload ports, unrelated containers, and ungranted networks. Preserve + native application traffic while replacing the initial coarse + two-container shared-network policy after parity is proven. Make every + gateway rule, address, and network resource lease-owned and reconcile it + during teardown. Include Docker, Podman, Desktop, hostile-root, + multi-user-host, concurrent-connection, interruption, and cleanup tests + proving the workload cannot reverse the route or bypass the endpoint + grant and unrelated local processes cannot reach the workload endpoint. + + Replace the temporary, discouraged `environment.runtime.network.ambiguous: allow` escape hatch with precise translated-destination policy and deprecate that coarse override. diff --git a/docs/CONTROLLED_SESSION_DESIGN.md b/docs/CONTROLLED_SESSION_DESIGN.md index d9c0a4d9..17400093 100644 --- a/docs/CONTROLLED_SESSION_DESIGN.md +++ b/docs/CONTROLLED_SESSION_DESIGN.md @@ -10,9 +10,9 @@ summary: Capability-scoped execution sessions that inherit Reploy's global conta - Decision state: Focused review complete; high-level decisions approved - Implementation state: Initial global sandbox prerequisites, trusted - application-startup verification, and controlled-session authorization are - implemented; protocol, lifecycle, and Docker orchestration remain later - slices + application-startup verification, controlled-session authorization, and the + initial framed protocol are implemented; lifecycle, controlled networking, + and Docker orchestration remain later slices - Initial runtime: Linux containers under Docker - Motivating clients: OmegaFlow recording, sandboxed AI agents, security inspection, and untrusted-code execution @@ -215,9 +215,9 @@ The initial controlled-session work does not: **Session channel** : A private channel between the controller and the attached Host Reploy operation. It is created for one already planned lease and carries PTY data, - bounded controller requests, declared endpoint streams, and host-observed - lifecycle events. It grants no session-creation or general host-runtime - authority and is never mounted into the workload container. + bounded controller requests, and host-observed lifecycle events. It grants no + session-creation or general host-runtime authority and is never mounted into + the workload container. ## Architecture @@ -228,7 +228,7 @@ Host attached Reploy operation ├── records their exact identities and starts the session watchdog ├── starts the controller and workload containers ├── owns the Docker TTY attachment and framed session protocol -├── forwards only predeclared workload endpoints +├── creates the lease-owned controller/workload network ├── monitors and cleans every leased runtime resource └── independently observes lifecycle completion ⇅ @@ -237,7 +237,7 @@ Host attached Reploy operation Controller container ├── controller orchestration ├── optional policy, inspection, or capture components -└── optional controller-local endpoint adapter +└── native clients for granted workload endpoints Workload container ├── untrusted workload shell on the Docker-managed PTY @@ -382,7 +382,7 @@ record containing: - the effective runtime identity inherited from the execution scope; - the network and endpoint grants; - the mount and source grants; -- the permitted session and endpoint operations; +- the permitted session operations; - the admitted lease identity and its owner-connection policy. The authorization record is portable immutable data; it does not serialize a @@ -431,9 +431,6 @@ bytes are never parsed as protocol messages. that the controller has finalized its client-owned results. It does not stop an active workload and is rejected before workload output reaches a terminal state. -- `open_endpoint(request_id, endpoint_id)`: request one byte stream to a logical - endpoint fixed in the immutable session plan. `request_id` correlates the - result with this request; it is not the stream identity. - `acknowledge_terminated`: confirm receipt of the authoritative `terminated` event. This payload-free protocol handshake is mandatory housekeeping, not a granted capability, and is accepted only after Host Reploy has successfully @@ -459,28 +456,15 @@ cleanup and no canceled request is replayed. - `terminating(cause)`: reports that the host-owned terminal transition began. - `diagnostic(code, message)`: reports protocol, runtime, or cleanup failure without embedding secret values. -- `endpoint_opened(request_id, endpoint_id, stream_id)`: reports a successful - endpoint open and assigns its host-generated session-unique stream identity. -- `endpoint_open_failed(request_id, endpoint_id, code, message)`: reports an - unsuccessful endpoint open without assigning a stream identity. -- `endpoint_closed(stream_id, endpoint_id, reason)`: reports one forwarded - endpoint stream ending. - `workload_outputs_finalized(status, reason)`: establishes that no further workload output can arrive. `status` is `drained` when every byte was delivered and `failed` when bounded finalization had to close an incomplete output surface. -Every well-formed `open_endpoint` request receives exactly one correlated -`endpoint_opened` or `endpoint_open_failed` outcome. Request IDs are nonzero -controller-generated 64-bit values and must be unique while a request is -pending. Stream IDs are nonzero host-generated 64-bit values and are unique for -the session. Protocol-invalid frames are handled as protocol errors rather than -endpoint-open outcomes. - Host Reploy emits the authoritative lease lifecycle result: -- `terminated(cause, workload_status, controller_finalization_status, - cleanup_status, recovery_action)`. +- `terminated(cause, workload_status, workload_output_finalization_status, + controller_finalization_status, cleanup_status, recovery_action)`. `controller_finalization_status` reports the controller protocol outcome: `completed`, `lost`, `finalization-timeout`, `not-completed`, or @@ -503,18 +487,16 @@ uses bounded buffers. A slow controller applies backpressure; Reploy does not silently drop or reorder terminal bytes. Limits and timeout diagnostics are explicit. -PTY, endpoint, and lifecycle streams use independent bounded flow-control -windows. A stalled browser transfer cannot indefinitely block terminal output, -termination, or the authoritative lifecycle result. +PTY and lifecycle streams use independent bounded flow-control windows. Host Reploy owns workload-output finalization; it never waits indefinitely for workload cooperation. Once termination begins, it rejects new output surfaces, performs bounded graceful shutdown followed by forced container stop, and -continues draining the PTY plus every existing endpoint stream. The immutable -session plan carries a finite output-finalization deadline. The initial -implementation may use a fixed host-owned value, but the effective value is -reported by `opened` and applies to workload shutdown, final buffered-byte -delivery, and controller backpressure. +continues draining the PTY. The immutable session plan carries a finite +output-finalization deadline. The initial implementation may use a fixed +host-owned value, but the effective value is reported by `opened` and applies +to workload shutdown, final buffered-byte delivery, and controller +backpressure. If every final byte is delivered and every output surface reaches EOF before the deadline, Host Reploy emits `workload_outputs_finalized(drained)` only after @@ -526,30 +508,11 @@ that no output frame can follow either outcome. Failure is explicit and cannot be converted into successful completion by `complete` or terminal acknowledgement. -The barrier covers every workload-originated output surface declared by the -session plan. Initially these are the PTY and forwarded endpoint streams. A -future workload output-file or output-directory contract joins the same -barrier after its files are closed, validated, and published or have recorded -an explicit failure; protocol v1 does not otherwise speculate about file -payloads. - -Every endpoint byte frame in either direction carries its `stream_id`. Host -Reploy assigns that ID only after the endpoint connection succeeds, never -reuses it within the session, and rejects data or closure operations for an -unknown or already closed ID. The logical `endpoint_id` remains the capability -being exercised; it is not sufficient to distinguish concurrent connections -to that endpoint. - -The initial implementation also enforces fixed Host Reploy maxima of 32 active -streams per logical endpoint, 64 active endpoint streams per session, and 64 -new endpoint streams per second per session with a burst of 128. Reploy reserves -capacity before dialing or allocating stream state and releases it when the -stream closes. An excess `open_endpoint` request is not queued or dialed; it is -rejected immediately with a correlated `endpoint_open_failed` event whose code -is `resource_exhausted`. -Blueprints and controllers cannot raise these host-owned limits. A future -general Reploy configuration surface may make them operator-configurable after -concrete use cases justify that surface. +The barrier initially covers the PTY. A future workload output-file or +output-directory contract joins the same barrier after its files are closed, +validated, and published or have recorded an explicit failure; protocol v1 +does not otherwise speculate about file payloads. Native network traffic is not +session output and does not pass through this barrier. A PTY merges standard output and standard error. The contract does not pretend to recover separate streams. @@ -608,9 +571,8 @@ sequences, fake exit messages, and protocol-looking output cannot create control or lifecycle events. Root workload code has more authority inside its own container, but it still -cannot reach the host-controlled session channel. A forged terminal message, -workload exit, or closed endpoint stream cannot become an authoritative -successful termination. +cannot reach the host-controlled session channel. A forged terminal message or +workload exit cannot become an authoritative successful termination. ## Runtime Identity @@ -921,33 +883,51 @@ claim domain-, URL-, DNS-content-, general outbound destination-port-, or audit-level policy. A controller may receive an explicit session-local grant to a declared -workload endpoint. That grant is not treated as general local-network -access. +workload endpoint. Endpoint declarations remain the stable intent and the +future enforcement unit, but the initial direct-network backend does not claim +to enforce each declaration as a precise capability boundary. The first OmegaFlow prototype needs only: ```text -controller browser -> Host Reploy -> one declared workload HTTP endpoint +controller browser -> lease-private Docker network -> workload HTTP endpoint ``` -The controller and workload do not share a Docker network. Docker publishes -only the declared workload port to an ephemeral host-loopback port. A -controller-local adapter accepts Chromium connections on controller -loopback and multiplexes their byte streams over the private session channel. -Host Reploy maps the fixed logical endpoint identity to the loopback-published -port. The protocol accepts no raw host, IP address, port, or URL destination. -The workload cannot use this one-way forwarding path to reach the controller. - -The loopback-published port is never disclosed to the controller, but any local -host process that discovers it may connect while the session is active. The -initial implementation accepts this host-local exposure; its session grant -constrains the controller, not unrelated host processes. It must not claim -per-lease endpoint privacy on a multi-user host. The port, adapter streams, and -associated Docker state remain lease-owned and are removed with the session. -The future L3 policy gateway must eliminate this direct host reachability or -enforce equivalent per-lease access control. General public/local network denial -remains a separate prerequisite; this endpoint forwarding path is not a general -router, HTTP policy engine, or domain-aware firewall. +The initial implementation attaches only the controller and workload to one +fresh lease-owned, engine-internal Docker network. The immutable plans carry a +separate session-network grant that admits that lease network inside both +containers; it does not set or imply general `local: allow`. The initial backend +admits the complete lease network in both directions. The controller uses +ordinary native TCP to the workload's session-local network identity and +declared port. Endpoint coordinates are resolved before startup as part of the +immutable controller and workload plans; endpoint traffic never enters the +private session channel and no workload port is published on the host. + +This is intentionally a coarse pre-gateway boundary. Membership in the private +network gives the controller reachability to workload ports beyond the declared +endpoint and gives the workload network reachability toward the controller. +The containers still receive no route to unrelated containers, the host local +network, or the public Internet unless separately granted, and the controller +must not expose sensitive listeners on its session-network interface. Reploy +reports this limitation rather than describing endpoint declarations as fully +enforced capabilities. + +The target L3 policy gateway hardens this direct native transport. It gives the +controller and workload separate session network identities and permits TCP +only from the controller to exact declared workload addresses and ports. It +denies workload-initiated connections to the controller, undeclared workload +ports, unrelated containers, and ungranted networks. + +Gateway policy, addresses, and network resources are lease-owned. Host Reploy +installs them before either application can use the route, verifies the +root-resistant policy, and removes or reconciles them during session teardown. +After gateway parity is proven on the supported Docker, Podman, and Desktop +backends, it replaces the coarse shared-network policy without changing how +applications use native TCP. PTY, lifecycle, termination, and diagnostic +traffic remain on the private session channel. + +General public/local network denial remains a separate prerequisite; neither +endpoint backend is a general HTTP policy engine or domain-aware firewall. General network isolation and auditability are a separate design surface. Future work may include an HTTP/HTTPS proxy, destination and DNS-content @@ -970,8 +950,9 @@ For the OmegaFlow profile: - Host Reploy owns the Docker TTY attachment and external session supervision; - the shell runs on the Docker-managed PTY in the workload container; - the demonstrated web service runs in the workload environment; -- Chromium reaches that service only through its controller-local adapter and - the one granted Host Reploy endpoint stream. +- Chromium reaches that service over the lease-private native network; the + initial backend has the documented coarse shared-network gap and the target + gateway backend enforces the declared endpoint direction and port. Terminal-to-browser handoff is an OmegaFlow orchestration concern inside the controller. Reploy does not model beats, handoffs, browser actions, or capture @@ -1021,21 +1002,20 @@ preparing -> active -> terminating -> terminated The first accepted termination cause is latched and never rewritten. Causes include controller-requested termination, workload exit, host cancellation, controller loss, Docker-observation loss, and startup failure. Later events -remain diagnostic observations. Workload status, controller finalization -status, and pre-delivery cleanup success are reported separately in the session -result, so a cleanup failure can fail the operation without hiding its original -cause. Controller exit and delivery-tail cleanup are reported separately by the -invoking host operation after teardown. +remain diagnostic observations. Workload status, workload-output-finalization +status, controller finalization status, and pre-delivery cleanup success are +reported separately in the session result, so a cleanup failure can fail the +operation without hiding its original cause. Controller exit and delivery-tail +cleanup are reported separately by the invoking host operation after teardown. Channel closure is never successful completion. The controller must explicitly send `complete` after receiving `workload_outputs_finalized` and finalizing its client-owned results; for OmegaFlow these include the recording artifacts. -Repeated terminate or host cancel operations are idempotent. Input, resize, and -new endpoint streams are rejected after `terminating` begins. A single -`complete` remains valid during termination while Host Reploy is waiting for -controller finalization. A `failed` workload-output result makes the session -fail regardless of whether the controller preserves and finalizes partial -artifacts. +Repeated terminate or host cancel operations are idempotent. Input and resize +are rejected after `terminating` begins. A single `complete` remains valid +during termination while Host Reploy is waiting for controller finalization. A +`failed` workload-output result makes the session fail regardless of whether +the controller preserves and finalizes partial artifacts. Normal completion is: @@ -1051,13 +1031,13 @@ Normal completion is: 6. Host Reploy gives the live controller a bounded finalization period in which to close its client-owned output and send `complete`. A failed output outcome remains a session failure even when partial client artifacts are finalized. -7. Host Reploy removes the workload container, endpoint publication, temporary - mounts, networks, and every other lease resource not required to deliver the - final result. It keeps the controller and private session channel alive. -8. Host Reploy records the original cause, controller-finalization result, - workload status, controller protocol status, and pre-delivery cleanup result, - then emits the one authoritative `terminated` event. Only successful event - delivery arms the acknowledgement wait. +7. Host Reploy removes the workload container, temporary mounts, networks, and + every other lease resource not required to deliver the final result. It + keeps the controller and private session channel alive. +8. Host Reploy records the original cause, workload status, + workload-output-finalization status, controller-finalization status, and + pre-delivery cleanup result, then emits the one authoritative `terminated` + event. Only successful event delivery arms the acknowledgement wait. 9. Host Reploy waits for a bounded `acknowledge_terminated` response. Channel closure is not an acknowledgement. Timeout or disconnect does not block teardown. @@ -1070,15 +1050,16 @@ A controller disconnect latches `controller_lost` and starts the same teardown. A workload exit is reported to the still-live controller so it can finalize its client-owned results; the session cannot succeed if the controller disappears before that finalization. Neither terminal output nor an -endpoint-stream close can substitute for host-observed Docker state. +application-level connection close can substitute for host-observed Docker +state. ### Session Watchdog Host Reploy starts one short-lived watchdog for each live controlled session. It first creates inert Docker resources and durably records their exact identities. Before starting either container, it passes the watchdog an -immutable cleanup manifest containing the exact lease, container, endpoint, -network, volume, and host boot identities. The attached operation retains one +immutable cleanup manifest containing the exact lease, container, network, +volume, and host boot identities. The attached operation retains one end of a private parent pipe. A crash during inert resource creation leaves no untrusted code running and is handled by ordinary next-operation reconciliation. @@ -1106,10 +1087,10 @@ networking are unavailable. Loss of authoritative Docker observation therefore immediately latches `runtime_observation_lost`, fails the recording, and closes its session and -endpoint streams. The session is never resumed or accepted as valid after -observation returns. Host Reploy and the watchdog retry Docker access and -forcibly remove any survivors when control returns. Immediate termination while -Docker itself is unreachable is not promised. +network. The session is never resumed or accepted as valid after observation +returns. Host Reploy and the watchdog retry Docker access and forcibly remove +any survivors when control returns. Immediate termination while Docker itself +is unreachable is not promised. A real host reboot ends the processes. The no-restart policy prevents their automatic return, and prior-boot queue entries are discarded under Reploy's @@ -1166,11 +1147,9 @@ Controlled sessions have explicit limits for: - termination grace; - buffered terminal output; - controller request size; -- active endpoint streams and endpoint-open rate; - process, memory, CPU, and temporary-disk resources where supported. -Limit failures produce a structured diagnostic. Endpoint-admission limits -reject only the excess request as specified above; a timeout or session-wide +Limit failures produce a structured diagnostic. A timeout or session-wide resource failure that makes safe continuation impossible produces bounded teardown. A timeout never converts into successful completion. @@ -1214,7 +1193,7 @@ while preserving exact declared inbound endpoints. Live Docker coverage exercises all four public/local combinations over IPv4 and IPv6, strict and escaped ambiguous-range handling, a globally reachable exception nested inside a reserved parent range, a root default-denial case, non-root authority removal, -guarded exec, and host-loopback endpoint publication. Resource limits remain a +guarded exec, and host-loopback denial. Resource limits remain a separate prerequisite slice. Root host authority is now enforced at runtime: host sources are classified as input, shared state, or explicit output; UID 0 is rejected for all three before container creation; and root output options are @@ -1247,13 +1226,16 @@ death, Host Reploy `SIGKILL`, watchdog interruption, Docker daemon restart with live-restore, Docker unavailability, and host-reboot recovery without touching unrelated resources. -### Slice 4: Declared Endpoint Forwarding +### Slice 4: Direct Session Networking -Add the controller-local adapter and Host Reploy forwarding to one exact -host-loopback-published workload endpoint. Prove Chromium can use HTTP and -WebSocket streams while the workload cannot reach the controller and neither -container receives unrelated local or public access. Keep general proxy, DNS, -domain, redirect, QUIC, and audit policy outside this slice. +Create one lease-private Docker network containing only the controller and +workload. Resolve the declared workload endpoint into the immutable session +plans and prove Chromium can use ordinary HTTP and WebSocket connections +without host publication or access to unrelated containers, host-local +networks, or the public Internet. Test and document the initial broader mutual +reachability inside that two-container network. Keep precise directional and +per-port enforcement, general proxy, DNS, domain, redirect, QUIC, and audit +policy in the deferred L3 gateway slice. ### Slice 5: OmegaFlow Proof @@ -1298,14 +1280,18 @@ lease protocol. After the implemented coarse public/local kill switches, define a separate Reploy userland L3 policy gateway for finer network control. Its design should -cover a capability-free application network namespace, one-shot route -initialization, an isolated data path whose only peer is the gateway, private -gateway control, root-resistant route invariants, direct-egress prevention, -destination and port grants, DNS and IPv6 policy, metadata protection, -auditing, resource limits, failure behavior, reconciliation, and portable -Docker/Podman integration. The one-way, exact endpoint forwarding used by the -initial controlled session remains intentionally narrower and does not depend -on this later gateway. +cover separate controller and workload network identities, a capability-free +application network namespace, one-shot route initialization, an isolated data +path whose only peer is the gateway, private gateway control, root-resistant +route invariants, direct-egress prevention, directional destination and port +grants, DNS and IPv6 policy, metadata protection, auditing, resource limits, +failure behavior, reconciliation, and portable Docker/Podman integration. + +The gateway becomes the target controlled-session endpoint backend. It permits +native controller-to-declared-workload TCP while denying the reverse direction +and every undeclared destination. Migration requires functional parity on every +supported backend and replaces the initial coarse shared-network policy without +changing application traffic from native TCP. ### Disposable Writable Workspaces diff --git a/internal/controlledsession/authorization.go b/internal/controlledsession/authorization.go index 2eaa4761..da396c68 100644 --- a/internal/controlledsession/authorization.go +++ b/internal/controlledsession/authorization.go @@ -11,7 +11,6 @@ import ( "fmt" "io" "regexp" - "slices" "strings" "unicode" "unicode/utf8" @@ -27,11 +26,10 @@ const AuthorizationSchemaV1 = "controlled-session-authorization-v1" type OperationV1 string const ( - OperationInputV1 OperationV1 = "input" - OperationResizeV1 OperationV1 = "resize" - OperationTerminateV1 OperationV1 = "terminate" - OperationCompleteV1 OperationV1 = "complete" - OperationOpenEndpointV1 OperationV1 = "open-endpoint" + OperationInputV1 OperationV1 = "input" + OperationResizeV1 OperationV1 = "resize" + OperationTerminateV1 OperationV1 = "terminate" + OperationCompleteV1 OperationV1 = "complete" ) // RuntimeIdentityV1 records the exact container-local identity selected before @@ -117,7 +115,7 @@ func ValidateAuthorizationV1(authorization AuthorizationV1) error { } for index, operation := range authorization.Operations { switch operation { - case OperationInputV1, OperationResizeV1, OperationTerminateV1, OperationCompleteV1, OperationOpenEndpointV1: + case OperationInputV1, OperationResizeV1, OperationTerminateV1, OperationCompleteV1: default: return fmt.Errorf("controlled-session operation %q is unsupported", operation) } @@ -133,9 +131,6 @@ func ValidateAuthorizationV1(authorization AuthorizationV1) error { return fmt.Errorf("controlled-session endpoint IDs must be unique and sorted") } } - if len(authorization.EndpointIDs) != 0 && !slices.Contains(authorization.Operations, OperationOpenEndpointV1) { - return fmt.Errorf("controlled-session endpoint grants require the open-endpoint operation") - } return nil } diff --git a/internal/controlledsession/authorization_test.go b/internal/controlledsession/authorization_test.go index 3873b1a7..5e4b41d8 100644 --- a/internal/controlledsession/authorization_test.go +++ b/internal/controlledsession/authorization_test.go @@ -15,7 +15,7 @@ func testAuthorizationV1() AuthorizationV1 { DeploymentID: "demo", GenerationReference: "reploy/env/demo:g-current", BuildIdentity: digest, LiveRunID: "run-0000000000000001", WorkloadPlan: digest, ControllerPlan: digest, RuntimeIdentity: RuntimeIdentityV1{Username: "reploy", UID: "1000", GID: "1000", SupplementaryGIDs: []string{"10", "100"}}, - Operations: []OperationV1{OperationCompleteV1, OperationInputV1, OperationOpenEndpointV1, OperationResizeV1, OperationTerminateV1}, + Operations: []OperationV1{OperationCompleteV1, OperationInputV1, OperationResizeV1, OperationTerminateV1}, EndpointIDs: []string{"browser", "terminal"}, } } @@ -62,7 +62,6 @@ func TestValidateAuthorizationV1RejectsOpenOrAmbiguousRecords(t *testing.T) { value.EndpointIDs[0], value.EndpointIDs[1] = value.EndpointIDs[1], value.EndpointIDs[0] }, want: "unique and sorted"}, {name: "invalid endpoint", mutate: func(value *AuthorizationV1) { value.EndpointIDs = []string{"API"} }, want: "Docker-style"}, - {name: "endpoint without capability", mutate: func(value *AuthorizationV1) { value.Operations = []OperationV1{OperationCompleteV1} }, want: "require the open-endpoint"}, {name: "nil collection", mutate: func(value *AuthorizationV1) { value.EndpointIDs = nil }, want: "must use arrays"}, {name: "unsafe generation", mutate: func(value *AuthorizationV1) { value.GenerationReference = "bad\nreference" }, want: "safe text"}, {name: "formatted generation", mutate: func(value *AuthorizationV1) { value.GenerationReference = "bad\u202ereference" }, want: "safe text"}, diff --git a/internal/controlledsession/model.go b/internal/controlledsession/model.go new file mode 100644 index 00000000..4ac26a48 --- /dev/null +++ b/internal/controlledsession/model.go @@ -0,0 +1,197 @@ +package controlledsession + +import "fmt" + +type TerminationCauseV1 string + +const ( + CauseControllerTerminateV1 TerminationCauseV1 = "controller-terminate" + CauseWorkloadExitV1 TerminationCauseV1 = "workload-exit" + CauseHostCancelV1 TerminationCauseV1 = "host-cancel" + CauseControllerLostV1 TerminationCauseV1 = "controller-lost" + CauseRuntimeObservationLostV1 TerminationCauseV1 = "runtime-observation-lost" + CauseStartupFailureV1 TerminationCauseV1 = "startup-failure" +) + +type ProcessStatusKindV1 string + +const ( + ProcessStatusUnknownV1 ProcessStatusKindV1 = "unknown" + ProcessStatusExitedV1 ProcessStatusKindV1 = "exited" + ProcessStatusTerminatedV1 ProcessStatusKindV1 = "terminated" + ProcessStatusUnavailableV1 ProcessStatusKindV1 = "unavailable" +) + +type ProcessStatusV1 struct { + Kind ProcessStatusKindV1 `json:"kind"` + Code *int `json:"code,omitempty"` + Reason string `json:"reason,omitempty"` +} + +type ControllerFinalizationStatusKindV1 string + +const ( + ControllerFinalizationUnknownV1 ControllerFinalizationStatusKindV1 = "unknown" + ControllerFinalizationActiveV1 ControllerFinalizationStatusKindV1 = "active" + ControllerFinalizationCompletedV1 ControllerFinalizationStatusKindV1 = "completed" + ControllerFinalizationLostV1 ControllerFinalizationStatusKindV1 = "lost" + ControllerFinalizationTimeoutV1 ControllerFinalizationStatusKindV1 = "finalization-timeout" + ControllerFinalizationNotCompletedV1 ControllerFinalizationStatusKindV1 = "not-completed" + ControllerFinalizationStartupFailedV1 ControllerFinalizationStatusKindV1 = "startup-failed" +) + +type ControllerFinalizationStatusV1 struct { + Kind ControllerFinalizationStatusKindV1 `json:"kind"` + Reason string `json:"reason,omitempty"` +} + +type WorkloadOutputFinalizationStatusKindV1 string + +const ( + WorkloadOutputFinalizationDrainedV1 WorkloadOutputFinalizationStatusKindV1 = "drained" + WorkloadOutputFinalizationFailedV1 WorkloadOutputFinalizationStatusKindV1 = "failed" +) + +type WorkloadOutputFinalizationStatusV1 struct { + Kind WorkloadOutputFinalizationStatusKindV1 `json:"kind"` + Reason string `json:"reason,omitempty"` +} + +type CleanupStatusKindV1 string + +const ( + CleanupStatusSucceededV1 CleanupStatusKindV1 = "succeeded" + CleanupStatusFailedV1 CleanupStatusKindV1 = "failed" +) + +type CleanupStatusV1 struct { + Kind CleanupStatusKindV1 `json:"kind"` + Message string `json:"message,omitempty"` +} + +type RecoveryActionV1 string + +const ( + RecoveryNoneV1 RecoveryActionV1 = "none" + RecoveryRetryCleanupV1 RecoveryActionV1 = "retry-cleanup" + RecoveryReconcileNextOperationV1 RecoveryActionV1 = "reconcile-next-operation" +) + +type ResultV1 struct { + Cause TerminationCauseV1 `json:"cause"` + WorkloadStatus ProcessStatusV1 `json:"workload_status"` + WorkloadOutputFinalizationStatus WorkloadOutputFinalizationStatusV1 `json:"workload_output_finalization_status"` + ControllerFinalizationStatus ControllerFinalizationStatusV1 `json:"controller_finalization_status"` + CleanupStatus CleanupStatusV1 `json:"cleanup_status"` + RecoveryAction RecoveryActionV1 `json:"recovery_action"` +} + +func ValidateResultV1(result ResultV1) error { + if !validTerminationCauseV1(result.Cause) { + return fmt.Errorf("controlled-session termination cause %q is invalid", result.Cause) + } + if err := validateProcessStatusV1(result.WorkloadStatus, true); err != nil { + return err + } + if result.Cause == CauseWorkloadExitV1 && result.WorkloadStatus.Kind == ProcessStatusUnknownV1 { + return fmt.Errorf("workload-exit termination requires a known workload status") + } + if err := validateWorkloadOutputFinalizationStatusV1(result.WorkloadOutputFinalizationStatus); err != nil { + return err + } + if err := validateTerminalControllerFinalizationStatusV1(result.ControllerFinalizationStatus); err != nil { + return err + } + return validateCleanupResultV1(result.CleanupStatus, result.RecoveryAction) +} + +func validateWorkloadOutputFinalizationStatusV1(status WorkloadOutputFinalizationStatusV1) error { + switch status.Kind { + case WorkloadOutputFinalizationDrainedV1: + if status.Reason != "" { + return fmt.Errorf("drained workload output finalization must not contain a reason") + } + case WorkloadOutputFinalizationFailedV1: + if err := validateRequiredSafeTextV1("workload output finalization failure reason", status.Reason); err != nil { + return err + } + default: + return fmt.Errorf("workload output finalization status %q is invalid", status.Kind) + } + return nil +} + +func validateProcessStatusV1(status ProcessStatusV1, allowUnknown bool) error { + switch status.Kind { + case ProcessStatusUnknownV1: + if !allowUnknown || status.Code != nil || status.Reason != "" { + return fmt.Errorf("unknown workload status must have no code or reason") + } + case ProcessStatusExitedV1: + if status.Code == nil { + return fmt.Errorf("exited workload status requires an exit code") + } + case ProcessStatusTerminatedV1, ProcessStatusUnavailableV1: + if status.Code != nil { + return fmt.Errorf("workload status %q must not contain an exit code", status.Kind) + } + default: + return fmt.Errorf("workload status %q is invalid", status.Kind) + } + return validateOptionalSafeTextV1("workload status reason", status.Reason) +} + +func validateTerminalControllerFinalizationStatusV1(status ControllerFinalizationStatusV1) error { + switch status.Kind { + case ControllerFinalizationCompletedV1, ControllerFinalizationLostV1, ControllerFinalizationTimeoutV1, ControllerFinalizationNotCompletedV1, ControllerFinalizationStartupFailedV1: + default: + return fmt.Errorf("terminal controller finalization status %q is invalid", status.Kind) + } + return validateOptionalSafeTextV1("controller finalization status reason", status.Reason) +} + +func validateCleanupResultV1(status CleanupStatusV1, recovery RecoveryActionV1) error { + switch status.Kind { + case CleanupStatusSucceededV1: + if status.Message != "" { + return fmt.Errorf("successful cleanup status must not contain an error message") + } + if recovery != RecoveryNoneV1 { + return fmt.Errorf("successful cleanup must not require recovery") + } + case CleanupStatusFailedV1: + if err := validateRequiredSafeTextV1("cleanup failure message", status.Message); err != nil { + return err + } + if recovery != RecoveryRetryCleanupV1 && recovery != RecoveryReconcileNextOperationV1 { + return fmt.Errorf("failed cleanup requires a recovery action") + } + default: + return fmt.Errorf("cleanup status %q is invalid", status.Kind) + } + return nil +} + +func validateOptionalSafeTextV1(field string, value string) error { + if value == "" { + return nil + } + return validateSafeTextV1(field, value) +} + +func validateRequiredSafeTextV1(field string, value string) error { + if value == "" { + return fmt.Errorf("controlled-session %s is required", field) + } + return validateSafeTextV1(field, value) +} + +func validTerminationCauseV1(cause TerminationCauseV1) bool { + switch cause { + case CauseControllerTerminateV1, CauseWorkloadExitV1, CauseHostCancelV1, + CauseControllerLostV1, CauseRuntimeObservationLostV1, CauseStartupFailureV1: + return true + default: + return false + } +} diff --git a/internal/controlledsession/protocol.go b/internal/controlledsession/protocol.go new file mode 100644 index 00000000..79193578 --- /dev/null +++ b/internal/controlledsession/protocol.go @@ -0,0 +1,557 @@ +package controlledsession + +import ( + "bytes" + "encoding/binary" + "encoding/json" + "fmt" + "io" + "regexp" + "unicode/utf8" +) + +const ( + ProtocolVersionV1 = 1 + MaxFramePayloadV1 = 1 << 20 + DefaultOutputFinalizationTimeoutMillisecondsV1 uint32 = 30_000 + frameHeaderSizeV1 = 10 + maxProtocolCodeLengthV1 = 63 +) + +var frameMagicV1 = [4]byte{'R', 'P', 'S', 'N'} + +var protocolCodePatternV1 = regexp.MustCompile(`^[a-z][a-z0-9]*(?:_[a-z0-9]+)*$`) + +type RequestKindV1 string + +const ( + RequestInputV1 RequestKindV1 = "input" + RequestResizeV1 RequestKindV1 = "resize" + RequestTerminateV1 RequestKindV1 = "terminate" + RequestCompleteV1 RequestKindV1 = "complete" + RequestAcknowledgeTerminatedV1 RequestKindV1 = "acknowledge-terminated" +) + +type RequestV1 struct { + Kind RequestKindV1 + Bytes []byte + Columns uint32 + Rows uint32 +} + +type EventKindV1 string + +const ( + EventOpenedV1 EventKindV1 = "opened" + EventOutputV1 EventKindV1 = "output" + EventWorkloadExitV1 EventKindV1 = "workload-exit" + EventTerminatingV1 EventKindV1 = "terminating" + EventDiagnosticV1 EventKindV1 = "diagnostic" + EventWorkloadOutputsFinalizedV1 EventKindV1 = "workload-outputs-finalized" + EventTerminatedV1 EventKindV1 = "terminated" +) + +type OpenedV1 struct { + Authorization AuthorizationV1 `json:"authorization"` + Columns uint32 `json:"columns"` + Rows uint32 `json:"rows"` + OutputFinalizationTimeoutMilliseconds uint32 `json:"output_finalization_timeout_milliseconds"` +} + +type WorkloadExitV1 struct { + Status ProcessStatusV1 `json:"status"` +} + +type TerminatingV1 struct { + Cause TerminationCauseV1 `json:"cause"` +} + +type DiagnosticV1 struct { + Code string `json:"code"` + Message string `json:"message"` +} + +type WorkloadOutputsFinalizedV1 struct { + Status WorkloadOutputFinalizationStatusKindV1 `json:"status"` + Reason string `json:"reason,omitempty"` +} + +type EventV1 struct { + Kind EventKindV1 + Bytes []byte + Opened *OpenedV1 + WorkloadExit *WorkloadExitV1 + Terminating *TerminatingV1 + Diagnostic *DiagnosticV1 + WorkloadOutputsFinalized *WorkloadOutputsFinalizedV1 + Terminated *ResultV1 +} + +type wireKindV1 byte + +const ( + wireRequestInputV1 wireKindV1 = 0x01 + wireRequestResizeV1 wireKindV1 = 0x02 + wireRequestTerminateV1 wireKindV1 = 0x03 + wireRequestCompleteV1 wireKindV1 = 0x04 + wireRequestAcknowledgeTerminatedV1 wireKindV1 = 0x05 +) + +const ( + wireEventOpenedV1 wireKindV1 = 0x81 + wireEventOutputV1 wireKindV1 = 0x82 + wireEventWorkloadExitV1 wireKindV1 = 0x83 + wireEventTerminatingV1 wireKindV1 = 0x84 + wireEventDiagnosticV1 wireKindV1 = 0x85 + wireEventTerminatedV1 wireKindV1 = 0x86 + wireEventWorkloadOutputsFinalizedV1 wireKindV1 = 0x87 +) + +func ValidateRequestV1(request RequestV1) error { + switch request.Kind { + case RequestInputV1: + if request.Bytes == nil || request.Columns != 0 || request.Rows != 0 { + return fmt.Errorf("controlled-session input request must contain only a byte sequence") + } + case RequestResizeV1: + if request.Bytes != nil || !validDimensionsV1(request.Columns, request.Rows) { + return fmt.Errorf("controlled-session resize request requires dimensions between 1 and 65535") + } + case RequestTerminateV1, RequestCompleteV1, RequestAcknowledgeTerminatedV1: + if request.Bytes != nil || request.Columns != 0 || request.Rows != 0 { + return fmt.Errorf("controlled-session %s request must not contain a payload", request.Kind) + } + default: + return fmt.Errorf("controlled-session request kind %q is unsupported", request.Kind) + } + return nil +} + +func ValidateEventV1(event EventV1) error { + switch event.Kind { + case EventOpenedV1: + if event.Opened == nil || eventPayloadCountV1(event) != 1 { + return fmt.Errorf("controlled-session opened event requires exactly one opened payload") + } + if err := ValidateAuthorizationV1(event.Opened.Authorization); err != nil { + return fmt.Errorf("controlled-session opened event: %w", err) + } + if !validDimensionsV1(event.Opened.Columns, event.Opened.Rows) { + return fmt.Errorf("controlled-session opened dimensions must be between 1 and 65535") + } + if event.Opened.OutputFinalizationTimeoutMilliseconds == 0 { + return fmt.Errorf("controlled-session opened event requires a finite output-finalization timeout") + } + case EventOutputV1: + if event.Bytes == nil || eventPayloadCountV1(event) != 1 { + return fmt.Errorf("controlled-session output event must contain only a byte sequence") + } + case EventWorkloadExitV1: + if event.WorkloadExit == nil || eventPayloadCountV1(event) != 1 { + return fmt.Errorf("controlled-session workload-exit event requires exactly one status payload") + } + if err := validateProcessStatusV1(event.WorkloadExit.Status, false); err != nil { + return fmt.Errorf("controlled-session workload-exit event: %w", err) + } + case EventTerminatingV1: + if event.Terminating == nil || eventPayloadCountV1(event) != 1 { + return fmt.Errorf("controlled-session terminating event requires exactly one cause payload") + } + if !validTerminationCauseV1(event.Terminating.Cause) { + return fmt.Errorf("controlled-session terminating cause %q is invalid", event.Terminating.Cause) + } + case EventDiagnosticV1: + if event.Diagnostic == nil || eventPayloadCountV1(event) != 1 { + return fmt.Errorf("controlled-session diagnostic event requires exactly one diagnostic payload") + } + if err := validateProtocolCodeV1("diagnostic code", event.Diagnostic.Code); err != nil { + return err + } + if err := validateRequiredSafeTextV1("diagnostic message", event.Diagnostic.Message); err != nil { + return err + } + case EventWorkloadOutputsFinalizedV1: + if event.WorkloadOutputsFinalized == nil || eventPayloadCountV1(event) != 1 { + return fmt.Errorf("controlled-session workload-outputs-finalized event requires exactly one finalization payload") + } + if err := validateWorkloadOutputFinalizationStatusV1(WorkloadOutputFinalizationStatusV1{ + Kind: event.WorkloadOutputsFinalized.Status, + Reason: event.WorkloadOutputsFinalized.Reason, + }); err != nil { + return fmt.Errorf("controlled-session workload-outputs-finalized event: %w", err) + } + case EventTerminatedV1: + if event.Terminated == nil || eventPayloadCountV1(event) != 1 { + return fmt.Errorf("controlled-session terminated event requires exactly one result payload") + } + if err := ValidateResultV1(*event.Terminated); err != nil { + return fmt.Errorf("controlled-session terminated event: %w", err) + } + default: + return fmt.Errorf("controlled-session event kind %q is unsupported", event.Kind) + } + return nil +} + +func WriteRequestV1(writer io.Writer, request RequestV1) error { + if err := ValidateRequestV1(request); err != nil { + return err + } + kind, payload := encodeRequestPayloadV1(request) + return writeFrameV1(writer, kind, payload) +} + +func ReadRequestV1(reader io.Reader) (RequestV1, error) { + kind, payload, err := readFrameV1(reader) + if err != nil { + return RequestV1{}, err + } + request, err := decodeRequestPayloadV1(kind, payload) + if err != nil { + return RequestV1{}, err + } + if err := ValidateRequestV1(request); err != nil { + return RequestV1{}, err + } + return request, nil +} + +func WriteEventV1(writer io.Writer, event EventV1) error { + if err := ValidateEventV1(event); err != nil { + return err + } + kind, payload, err := encodeEventPayloadV1(event) + if err != nil { + return err + } + return writeFrameV1(writer, kind, payload) +} + +func ReadEventV1(reader io.Reader) (EventV1, error) { + kind, payload, err := readFrameV1(reader) + if err != nil { + return EventV1{}, err + } + event, err := decodeEventPayloadV1(kind, payload) + if err != nil { + return EventV1{}, err + } + if err := ValidateEventV1(event); err != nil { + return EventV1{}, err + } + return event, nil +} + +func encodeRequestPayloadV1(request RequestV1) (wireKindV1, []byte) { + switch request.Kind { + case RequestInputV1: + return wireRequestInputV1, request.Bytes + case RequestResizeV1: + payload := make([]byte, 8) + binary.BigEndian.PutUint32(payload[0:4], request.Columns) + binary.BigEndian.PutUint32(payload[4:8], request.Rows) + return wireRequestResizeV1, payload + case RequestTerminateV1: + return wireRequestTerminateV1, nil + case RequestCompleteV1: + return wireRequestCompleteV1, nil + case RequestAcknowledgeTerminatedV1: + return wireRequestAcknowledgeTerminatedV1, nil + default: + panic("validated request has unsupported kind") + } +} + +func decodeRequestPayloadV1(kind wireKindV1, payload []byte) (RequestV1, error) { + switch kind { + case wireRequestInputV1: + return RequestV1{Kind: RequestInputV1, Bytes: payload}, nil + case wireRequestResizeV1: + if len(payload) != 8 { + return RequestV1{}, fmt.Errorf("controlled-session resize frame payload must contain 8 bytes") + } + return RequestV1{Kind: RequestResizeV1, Columns: binary.BigEndian.Uint32(payload[:4]), Rows: binary.BigEndian.Uint32(payload[4:])}, nil + case wireRequestTerminateV1: + if len(payload) != 0 { + return RequestV1{}, fmt.Errorf("controlled-session terminate frame must not contain a payload") + } + return RequestV1{Kind: RequestTerminateV1}, nil + case wireRequestCompleteV1: + if len(payload) != 0 { + return RequestV1{}, fmt.Errorf("controlled-session complete frame must not contain a payload") + } + return RequestV1{Kind: RequestCompleteV1}, nil + case wireRequestAcknowledgeTerminatedV1: + if len(payload) != 0 { + return RequestV1{}, fmt.Errorf("controlled-session acknowledge-terminated frame must not contain a payload") + } + return RequestV1{Kind: RequestAcknowledgeTerminatedV1}, nil + default: + return RequestV1{}, fmt.Errorf("controlled-session frame kind 0x%02x is not a controller request", byte(kind)) + } +} + +func encodeEventPayloadV1(event EventV1) (wireKindV1, []byte, error) { + switch event.Kind { + case EventOpenedV1: + return marshalEventPayloadV1(wireEventOpenedV1, event.Opened) + case EventOutputV1: + return wireEventOutputV1, event.Bytes, nil + case EventWorkloadExitV1: + return marshalEventPayloadV1(wireEventWorkloadExitV1, event.WorkloadExit) + case EventTerminatingV1: + return marshalEventPayloadV1(wireEventTerminatingV1, event.Terminating) + case EventDiagnosticV1: + return marshalEventPayloadV1(wireEventDiagnosticV1, event.Diagnostic) + case EventWorkloadOutputsFinalizedV1: + return marshalEventPayloadV1(wireEventWorkloadOutputsFinalizedV1, event.WorkloadOutputsFinalized) + case EventTerminatedV1: + return marshalEventPayloadV1(wireEventTerminatedV1, event.Terminated) + default: + panic("validated event has unsupported kind") + } +} + +func marshalEventPayloadV1(kind wireKindV1, value any) (wireKindV1, []byte, error) { + payload, err := json.Marshal(value) + if err != nil { + return 0, nil, fmt.Errorf("encode controlled-session event: %w", err) + } + return kind, payload, nil +} + +func decodeEventPayloadV1(kind wireKindV1, payload []byte) (EventV1, error) { + switch kind { + case wireEventOpenedV1: + value := new(OpenedV1) + return EventV1{Kind: EventOpenedV1, Opened: value}, decodeStrictJSONV1("opened event", payload, value) + case wireEventOutputV1: + return EventV1{Kind: EventOutputV1, Bytes: payload}, nil + case wireEventWorkloadExitV1: + value := new(WorkloadExitV1) + return EventV1{Kind: EventWorkloadExitV1, WorkloadExit: value}, decodeStrictJSONV1("workload-exit event", payload, value) + case wireEventTerminatingV1: + value := new(TerminatingV1) + return EventV1{Kind: EventTerminatingV1, Terminating: value}, decodeStrictJSONV1("terminating event", payload, value) + case wireEventDiagnosticV1: + value := new(DiagnosticV1) + return EventV1{Kind: EventDiagnosticV1, Diagnostic: value}, decodeStrictJSONV1("diagnostic event", payload, value) + case wireEventWorkloadOutputsFinalizedV1: + value := new(WorkloadOutputsFinalizedV1) + return EventV1{Kind: EventWorkloadOutputsFinalizedV1, WorkloadOutputsFinalized: value}, decodeStrictJSONV1("workload-outputs-finalized event", payload, value) + case wireEventTerminatedV1: + value := new(ResultV1) + return EventV1{Kind: EventTerminatedV1, Terminated: value}, decodeStrictJSONV1("terminated event", payload, value) + default: + return EventV1{}, fmt.Errorf("controlled-session frame kind 0x%02x is not a session event", byte(kind)) + } +} + +func eventPayloadCountV1(event EventV1) int { + count := 0 + for _, present := range []bool{ + event.Bytes != nil, + event.Opened != nil, + event.WorkloadExit != nil, + event.Terminating != nil, + event.Diagnostic != nil, + event.WorkloadOutputsFinalized != nil, + event.Terminated != nil, + } { + if present { + count++ + } + } + return count +} + +func writeFrameV1(writer io.Writer, kind wireKindV1, payload []byte) error { + if len(payload) > MaxFramePayloadV1 { + return fmt.Errorf("controlled-session frame payload exceeds %d bytes", MaxFramePayloadV1) + } + header := make([]byte, frameHeaderSizeV1) + copy(header[:4], frameMagicV1[:]) + header[4] = ProtocolVersionV1 + header[5] = byte(kind) + binary.BigEndian.PutUint32(header[6:], uint32(len(payload))) + if err := writeAllV1(writer, header); err != nil { + return fmt.Errorf("write controlled-session frame header: %w", err) + } + if err := writeAllV1(writer, payload); err != nil { + return fmt.Errorf("write controlled-session frame payload: %w", err) + } + return nil +} + +func readFrameV1(reader io.Reader) (wireKindV1, []byte, error) { + header := make([]byte, frameHeaderSizeV1) + if _, err := io.ReadFull(reader, header); err != nil { + return 0, nil, fmt.Errorf("read controlled-session frame header: %w", err) + } + if !bytes.Equal(header[:4], frameMagicV1[:]) { + return 0, nil, fmt.Errorf("controlled-session frame magic is invalid") + } + if header[4] != ProtocolVersionV1 { + return 0, nil, fmt.Errorf("controlled-session protocol version %d is unsupported", header[4]) + } + length := binary.BigEndian.Uint32(header[6:]) + if length > MaxFramePayloadV1 { + return 0, nil, fmt.Errorf("controlled-session frame payload length %d exceeds %d bytes", length, MaxFramePayloadV1) + } + payload := make([]byte, int(length)) + if _, err := io.ReadFull(reader, payload); err != nil { + return 0, nil, fmt.Errorf("read controlled-session frame payload: %w", err) + } + return wireKindV1(header[5]), payload, nil +} + +func decodeStrictJSONV1(subject string, payload []byte, value any) error { + if !utf8.Valid(payload) { + return fmt.Errorf("controlled-session %s is not valid UTF-8 JSON", subject) + } + if err := rejectDuplicateJSONFieldsV1(payload); err != nil { + return fmt.Errorf("decode controlled-session %s: %w", subject, err) + } + decoder := json.NewDecoder(bytes.NewReader(payload)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(value); err != nil { + return fmt.Errorf("decode controlled-session %s: %w", subject, err) + } + var trailer any + if err := decoder.Decode(&trailer); err != io.EOF { + if err == nil { + return fmt.Errorf("controlled-session %s contains trailing JSON", subject) + } + return fmt.Errorf("decode controlled-session %s trailer: %w", subject, err) + } + return nil +} + +func rejectDuplicateJSONFieldsV1(payload []byte) error { + decoder := json.NewDecoder(bytes.NewReader(payload)) + if err := scanJSONValueV1(decoder); err != nil { + return err + } + if token, err := decoder.Token(); err != io.EOF { + if err == nil { + return fmt.Errorf("JSON contains trailing token %v", token) + } + return err + } + return nil +} + +func scanJSONValueV1(decoder *json.Decoder) error { + token, err := decoder.Token() + if err != nil { + return err + } + delimiter, isDelimiter := token.(json.Delim) + if !isDelimiter { + return nil + } + switch delimiter { + case '{': + seen := map[string]bool{} + for decoder.More() { + keyToken, err := decoder.Token() + if err != nil { + return err + } + key, ok := keyToken.(string) + if !ok { + return fmt.Errorf("JSON object key is not a string") + } + if !validWireJSONFieldNameV1(key) { + return fmt.Errorf("JSON object field %q is not lowercase ASCII snake_case", key) + } + if seen[key] { + return fmt.Errorf("JSON object repeats field %q", key) + } + seen[key] = true + if err := scanJSONValueV1(decoder); err != nil { + return err + } + } + closing, err := decoder.Token() + if err != nil { + return err + } + if closing != json.Delim('}') { + return fmt.Errorf("JSON object is not closed") + } + case '[': + for decoder.More() { + if err := scanJSONValueV1(decoder); err != nil { + return err + } + } + closing, err := decoder.Token() + if err != nil { + return err + } + if closing != json.Delim(']') { + return fmt.Errorf("JSON array is not closed") + } + default: + return fmt.Errorf("unexpected JSON delimiter %q", delimiter) + } + return nil +} + +func validWireJSONFieldNameV1(value string) bool { + if value == "" { + return false + } + for index := 0; index < len(value); index++ { + character := value[index] + if character >= 'a' && character <= 'z' { + continue + } + if index != 0 && ((character >= '0' && character <= '9') || character == '_') { + continue + } + return false + } + return true +} + +func writeAllV1(writer io.Writer, content []byte) error { + for len(content) != 0 { + written, err := writer.Write(content) + if err != nil { + return err + } + if written <= 0 || written > len(content) { + return io.ErrShortWrite + } + content = content[written:] + } + return nil +} + +func operationForRequestV1(kind RequestKindV1) OperationV1 { + switch kind { + case RequestInputV1: + return OperationInputV1 + case RequestResizeV1: + return OperationResizeV1 + case RequestTerminateV1: + return OperationTerminateV1 + case RequestCompleteV1: + return OperationCompleteV1 + default: + return "" + } +} + +func validDimensionsV1(columns uint32, rows uint32) bool { + return columns >= 1 && columns <= 65535 && rows >= 1 && rows <= 65535 +} + +func validateProtocolCodeV1(subject string, value string) error { + if len(value) == 0 || len(value) > maxProtocolCodeLengthV1 || !protocolCodePatternV1.MatchString(value) { + return fmt.Errorf("controlled-session %s %q must be a lowercase ASCII snake_case identifier no longer than %d bytes", subject, value, maxProtocolCodeLengthV1) + } + return nil +} diff --git a/internal/controlledsession/protocol_test.go b/internal/controlledsession/protocol_test.go new file mode 100644 index 00000000..ecfe13b4 --- /dev/null +++ b/internal/controlledsession/protocol_test.go @@ -0,0 +1,313 @@ +package controlledsession + +import ( + "bytes" + "encoding/binary" + "encoding/json" + "reflect" + "strings" + "testing" +) + +func TestRequestV1RoundTripsBinarySafeFrames(t *testing.T) { + requests := []RequestV1{ + {Kind: RequestInputV1, Bytes: []byte{0, 3, '\n', 0xff}}, + {Kind: RequestResizeV1, Columns: 120, Rows: 40}, + {Kind: RequestTerminateV1}, + {Kind: RequestCompleteV1}, + {Kind: RequestAcknowledgeTerminatedV1}, + } + var stream bytes.Buffer + for _, request := range requests { + if err := WriteRequestV1(&stream, request); err != nil { + t.Fatalf("WriteRequestV1(%s) error = %v", request.Kind, err) + } + } + for _, want := range requests { + got, err := ReadRequestV1(&stream) + if err != nil { + t.Fatalf("ReadRequestV1(%s) error = %v", want.Kind, err) + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("ReadRequestV1() = %#v, want %#v", got, want) + } + } +} + +func TestEventV1RoundTripsStrictTypedFrames(t *testing.T) { + code := 0 + authorization := testAuthorizationV1() + events := []EventV1{ + {Kind: EventOpenedV1, Opened: &OpenedV1{ + Authorization: authorization, Columns: 80, Rows: 24, + OutputFinalizationTimeoutMilliseconds: DefaultOutputFinalizationTimeoutMillisecondsV1, + }}, + {Kind: EventOutputV1, Bytes: []byte{0, '\n', 0xff}}, + {Kind: EventWorkloadExitV1, WorkloadExit: &WorkloadExitV1{Status: ProcessStatusV1{Kind: ProcessStatusExitedV1, Code: &code}}}, + {Kind: EventTerminatingV1, Terminating: &TerminatingV1{Cause: CauseWorkloadExitV1}}, + {Kind: EventDiagnosticV1, Diagnostic: &DiagnosticV1{Code: "cleanup_failed", Message: "container removal failed"}}, + {Kind: EventWorkloadOutputsFinalizedV1, WorkloadOutputsFinalized: &WorkloadOutputsFinalizedV1{Status: WorkloadOutputFinalizationDrainedV1}}, + {Kind: EventWorkloadOutputsFinalizedV1, WorkloadOutputsFinalized: &WorkloadOutputsFinalizedV1{Status: WorkloadOutputFinalizationFailedV1, Reason: "output deadline expired"}}, + {Kind: EventTerminatedV1, Terminated: &ResultV1{ + Cause: CauseWorkloadExitV1, WorkloadStatus: ProcessStatusV1{Kind: ProcessStatusExitedV1, Code: &code}, + WorkloadOutputFinalizationStatus: WorkloadOutputFinalizationStatusV1{Kind: WorkloadOutputFinalizationDrainedV1}, + ControllerFinalizationStatus: ControllerFinalizationStatusV1{Kind: ControllerFinalizationCompletedV1}, + CleanupStatus: CleanupStatusV1{Kind: CleanupStatusSucceededV1}, RecoveryAction: RecoveryNoneV1, + }}, + } + var stream bytes.Buffer + for _, event := range events { + if err := WriteEventV1(&stream, event); err != nil { + t.Fatalf("WriteEventV1(%s) error = %v", event.Kind, err) + } + } + for _, want := range events { + got, err := ReadEventV1(&stream) + if err != nil { + t.Fatalf("ReadEventV1(%s) error = %v", want.Kind, err) + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("ReadEventV1() = %#v, want %#v", got, want) + } + } +} + +func TestFrameV1RejectsWrongProtocolDirectionAndUnboundedInput(t *testing.T) { + var event bytes.Buffer + if err := WriteEventV1(&event, EventV1{Kind: EventOutputV1, Bytes: []byte("hello")}); err != nil { + t.Fatal(err) + } + if _, err := ReadRequestV1(&event); err == nil || !strings.Contains(err.Error(), "not a controller request") { + t.Fatalf("ReadRequestV1(event) error = %v", err) + } + + header := make([]byte, frameHeaderSizeV1) + copy(header, frameMagicV1[:]) + header[4] = ProtocolVersionV1 + header[5] = byte(wireRequestInputV1) + binary.BigEndian.PutUint32(header[6:], MaxFramePayloadV1+1) + if _, err := ReadRequestV1(bytes.NewReader(header)); err == nil || !strings.Contains(err.Error(), "exceeds") { + t.Fatalf("ReadRequestV1(oversized) error = %v", err) + } +} + +func TestFrameV1RejectsBadMagicVersionTruncationAndUnknownJSON(t *testing.T) { + validHeader := func() []byte { + header := make([]byte, frameHeaderSizeV1) + copy(header, frameMagicV1[:]) + header[4] = ProtocolVersionV1 + header[5] = byte(wireRequestCompleteV1) + return header + } + tests := []struct { + name string + data []byte + want string + }{ + {name: "bad magic", data: append([]byte("NOPE"), validHeader()[4:]...), want: "magic"}, + {name: "bad version", data: func() []byte { value := validHeader(); value[4] = 2; return value }(), want: "version 2"}, + {name: "short header", data: validHeader()[:5], want: "frame header"}, + {name: "short payload", data: func() []byte { + value := validHeader() + value[5] = byte(wireRequestInputV1) + binary.BigEndian.PutUint32(value[6:], 2) + return append(value, 1) + }(), want: "frame payload"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if _, err := ReadRequestV1(bytes.NewReader(test.data)); err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("ReadRequestV1() error = %v, want containing %q", err, test.want) + } + }) + } + + payload := []byte(`{"code":"bad","message":"failure","extra":true}`) + var framed bytes.Buffer + if err := writeFrameV1(&framed, wireEventDiagnosticV1, payload); err != nil { + t.Fatal(err) + } + if _, err := ReadEventV1(&framed); err == nil || !strings.Contains(err.Error(), "unknown field") { + t.Fatalf("ReadEventV1(unknown JSON) error = %v", err) + } + + duplicate := []byte(`{"code":"first","code":"second","message":"failure"}`) + framed.Reset() + if err := writeFrameV1(&framed, wireEventDiagnosticV1, duplicate); err != nil { + t.Fatal(err) + } + if _, err := ReadEventV1(&framed); err == nil || !strings.Contains(err.Error(), "repeats field") { + t.Fatalf("ReadEventV1(duplicate JSON) error = %v", err) + } + + caseVariant := []byte(`{"Code":"bad","message":"failure"}`) + framed.Reset() + if err := writeFrameV1(&framed, wireEventDiagnosticV1, caseVariant); err != nil { + t.Fatal(err) + } + if _, err := ReadEventV1(&framed); err == nil || !strings.Contains(err.Error(), "lowercase ASCII snake_case") { + t.Fatalf("ReadEventV1(case-variant JSON) error = %v", err) + } + + caseVariantDuplicate := []byte(`{"code":"first","Code":"second","message":"failure"}`) + framed.Reset() + if err := writeFrameV1(&framed, wireEventDiagnosticV1, caseVariantDuplicate); err != nil { + t.Fatal(err) + } + if _, err := ReadEventV1(&framed); err == nil || !strings.Contains(err.Error(), "lowercase ASCII snake_case") { + t.Fatalf("ReadEventV1(case-variant duplicate JSON) error = %v", err) + } + + nestedCaseVariant := []byte(`{"cause":"workload-exit","workload_status":{"Kind":"exited","code":0},"workload_output_finalization_status":{"kind":"drained"},"controller_finalization_status":{"kind":"completed"},"cleanup_status":{"kind":"succeeded"},"recovery_action":"none"}`) + framed.Reset() + if err := writeFrameV1(&framed, wireEventTerminatedV1, nestedCaseVariant); err != nil { + t.Fatal(err) + } + if _, err := ReadEventV1(&framed); err == nil || !strings.Contains(err.Error(), "lowercase ASCII snake_case") { + t.Fatalf("ReadEventV1(nested case-variant JSON) error = %v", err) + } +} + +func TestWireKindAssignmentsV1AreStable(t *testing.T) { + tests := []struct { + name string + got wireKindV1 + want byte + }{ + {name: "request input", got: wireRequestInputV1, want: 0x01}, + {name: "request resize", got: wireRequestResizeV1, want: 0x02}, + {name: "request terminate", got: wireRequestTerminateV1, want: 0x03}, + {name: "request complete", got: wireRequestCompleteV1, want: 0x04}, + {name: "request acknowledge terminated", got: wireRequestAcknowledgeTerminatedV1, want: 0x05}, + {name: "event opened", got: wireEventOpenedV1, want: 0x81}, + {name: "event output", got: wireEventOutputV1, want: 0x82}, + {name: "event workload exit", got: wireEventWorkloadExitV1, want: 0x83}, + {name: "event terminating", got: wireEventTerminatingV1, want: 0x84}, + {name: "event diagnostic", got: wireEventDiagnosticV1, want: 0x85}, + {name: "event terminated", got: wireEventTerminatedV1, want: 0x86}, + {name: "event workload outputs finalized", got: wireEventWorkloadOutputsFinalizedV1, want: 0x87}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if byte(test.got) != test.want { + t.Fatalf("wire kind = 0x%02x, want 0x%02x", byte(test.got), test.want) + } + }) + } + + var framed bytes.Buffer + if err := WriteRequestV1(&framed, RequestV1{Kind: RequestCompleteV1}); err != nil { + t.Fatal(err) + } + want := []byte{'R', 'P', 'S', 'N', ProtocolVersionV1, 0x04, 0, 0, 0, 0} + if !bytes.Equal(framed.Bytes(), want) { + t.Fatalf("complete frame = %x, want %x", framed.Bytes(), want) + } +} + +func TestValidateEventV1RejectsInvalidOutputFinalizationOutcomes(t *testing.T) { + code := 0 + tests := []EventV1{ + {Kind: EventOpenedV1, Opened: &OpenedV1{Authorization: testAuthorizationV1(), Columns: 80, Rows: 24}}, + {Kind: EventWorkloadOutputsFinalizedV1}, + {Kind: EventWorkloadOutputsFinalizedV1, WorkloadOutputsFinalized: &WorkloadOutputsFinalizedV1{Status: WorkloadOutputFinalizationDrainedV1, Reason: "unexpected"}}, + {Kind: EventWorkloadOutputsFinalizedV1, WorkloadOutputsFinalized: &WorkloadOutputsFinalizedV1{Status: WorkloadOutputFinalizationFailedV1}}, + {Kind: EventTerminatedV1, Terminated: &ResultV1{ + Cause: CauseWorkloadExitV1, WorkloadStatus: ProcessStatusV1{Kind: ProcessStatusExitedV1, Code: &code}, + ControllerFinalizationStatus: ControllerFinalizationStatusV1{Kind: ControllerFinalizationCompletedV1}, + CleanupStatus: CleanupStatusV1{Kind: CleanupStatusSucceededV1}, RecoveryAction: RecoveryNoneV1, + }}, + } + for _, event := range tests { + if err := ValidateEventV1(event); err == nil { + t.Fatalf("ValidateEventV1(%#v) unexpectedly succeeded", event) + } + } +} + +func TestReadEventV1RejectsWorkloadExitWithoutStatus(t *testing.T) { + result := ResultV1{ + Cause: CauseWorkloadExitV1, + WorkloadStatus: ProcessStatusV1{Kind: ProcessStatusUnknownV1}, + WorkloadOutputFinalizationStatus: WorkloadOutputFinalizationStatusV1{Kind: WorkloadOutputFinalizationDrainedV1}, + ControllerFinalizationStatus: ControllerFinalizationStatusV1{Kind: ControllerFinalizationCompletedV1}, + CleanupStatus: CleanupStatusV1{Kind: CleanupStatusSucceededV1}, + RecoveryAction: RecoveryNoneV1, + } + payload, err := json.Marshal(result) + if err != nil { + t.Fatal(err) + } + var framed bytes.Buffer + if err := writeFrameV1(&framed, wireEventTerminatedV1, payload); err != nil { + t.Fatal(err) + } + if _, err := ReadEventV1(&framed); err == nil || !strings.Contains(err.Error(), "workload-exit termination requires a known workload status") { + t.Fatalf("ReadEventV1(workload exit without status) error = %v", err) + } +} + +func TestValidateRequestV1RejectsAmbiguousUnionPayloads(t *testing.T) { + tests := []RequestV1{ + {Kind: RequestInputV1}, + {Kind: RequestResizeV1, Columns: 80}, + {Kind: RequestCompleteV1, Bytes: []byte{}}, + {Kind: RequestAcknowledgeTerminatedV1, Columns: 1}, + } + for _, request := range tests { + if err := ValidateRequestV1(request); err == nil { + t.Fatalf("ValidateRequestV1(%#v) unexpectedly succeeded", request) + } + } +} + +func TestValidateEventV1RejectsInvalidProtocolCodes(t *testing.T) { + tests := []string{ + "", + "resource-exhausted", + "Resource_exhausted", + "resource__exhausted", + "resource_exhausted_", + strings.Repeat("a", maxProtocolCodeLengthV1+1), + } + for _, code := range tests { + event := EventV1{Kind: EventDiagnosticV1, Diagnostic: &DiagnosticV1{Code: code, Message: "failure"}} + if err := ValidateEventV1(event); err == nil { + t.Fatalf("ValidateEventV1(diagnostic code %q) unexpectedly succeeded", code) + } + } +} + +func TestReadRequestV1RejectsAcknowledgeTerminatedPayload(t *testing.T) { + var framed bytes.Buffer + if err := writeFrameV1(&framed, wireRequestAcknowledgeTerminatedV1, []byte{1}); err != nil { + t.Fatal(err) + } + if _, err := ReadRequestV1(&framed); err == nil || !strings.Contains(err.Error(), "must not contain a payload") { + t.Fatalf("ReadRequestV1(acknowledge payload) error = %v", err) + } +} + +func FuzzReadRequestV1(f *testing.F) { + f.Add([]byte{}) + var valid bytes.Buffer + if err := WriteRequestV1(&valid, RequestV1{Kind: RequestInputV1, Bytes: []byte{0, 3, 0xff}}); err != nil { + f.Fatal(err) + } + f.Add(valid.Bytes()) + f.Fuzz(func(t *testing.T, content []byte) { + _, _ = ReadRequestV1(bytes.NewReader(content)) + }) +} + +func FuzzReadEventV1(f *testing.F) { + f.Add([]byte{}) + var valid bytes.Buffer + if err := WriteEventV1(&valid, EventV1{Kind: EventDiagnosticV1, Diagnostic: &DiagnosticV1{Code: "test", Message: "seed"}}); err != nil { + f.Fatal(err) + } + f.Add(valid.Bytes()) + f.Fuzz(func(t *testing.T, content []byte) { + _, _ = ReadEventV1(bytes.NewReader(content)) + }) +}