diff --git a/config/buf-lint-baseline.txt b/config/buf-lint-baseline.txt index 2865ac7..6824fac 100644 --- a/config/buf-lint-baseline.txt +++ b/config/buf-lint-baseline.txt @@ -1,12 +1,12 @@ -protos/rvbox/v1/common.proto:262:5:Enum value name "INVALID_ARGUMENT" should be prefixed with "CODE_". -protos/rvbox/v1/common.proto:263:5:Enum value name "NOT_FOUND" should be prefixed with "CODE_". -protos/rvbox/v1/common.proto:264:5:Enum value name "OFFLINE" should be prefixed with "CODE_". -protos/rvbox/v1/common.proto:265:5:Enum value name "CAPACITY_EXHAUSTED" should be prefixed with "CODE_". -protos/rvbox/v1/common.proto:266:5:Enum value name "CONFLICT" should be prefixed with "CODE_". -protos/rvbox/v1/common.proto:267:5:Enum value name "UNSUPPORTED" should be prefixed with "CODE_". -protos/rvbox/v1/common.proto:268:5:Enum value name "PROTOCOL_ERROR" should be prefixed with "CODE_". -protos/rvbox/v1/common.proto:269:5:Enum value name "TRANSIENT" should be prefixed with "CODE_". -protos/rvbox/v1/common.proto:270:5:Enum value name "INTERNAL" should be prefixed with "CODE_". +protos/rvbox/v1/common.proto:265:5:Enum value name "INVALID_ARGUMENT" should be prefixed with "CODE_". +protos/rvbox/v1/common.proto:266:5:Enum value name "NOT_FOUND" should be prefixed with "CODE_". +protos/rvbox/v1/common.proto:267:5:Enum value name "OFFLINE" should be prefixed with "CODE_". +protos/rvbox/v1/common.proto:268:5:Enum value name "CAPACITY_EXHAUSTED" should be prefixed with "CODE_". +protos/rvbox/v1/common.proto:269:5:Enum value name "CONFLICT" should be prefixed with "CODE_". +protos/rvbox/v1/common.proto:270:5:Enum value name "UNSUPPORTED" should be prefixed with "CODE_". +protos/rvbox/v1/common.proto:271:5:Enum value name "PROTOCOL_ERROR" should be prefixed with "CODE_". +protos/rvbox/v1/common.proto:272:5:Enum value name "TRANSIENT" should be prefixed with "CODE_". +protos/rvbox/v1/common.proto:273:5:Enum value name "INTERNAL" should be prefixed with "CODE_". protos/rvbox/v1/common.proto:32:3:Enum value name "SHELL_SH" should be prefixed with "SHELL_TYPE_". protos/rvbox/v1/common.proto:33:3:Enum value name "SHELL_BASH" should be prefixed with "SHELL_TYPE_". protos/rvbox/v1/common.proto:34:3:Enum value name "SHELL_CMD" should be prefixed with "SHELL_TYPE_". diff --git a/gen/go/rvbox/v1/common.pb.go b/gen/go/rvbox/v1/common.pb.go index f109aa4..43f158c 100644 --- a/gen/go/rvbox/v1/common.pb.go +++ b/gen/go/rvbox/v1/common.pb.go @@ -1829,9 +1829,12 @@ type CommandEvent struct { // *CommandEvent_ScriptStatus // *CommandEvent_OutputTruncation // *CommandEvent_OutputIncomplete - Payload isCommandEvent_Payload `protobuf_oneof:"payload"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + Payload isCommandEvent_Payload `protobuf_oneof:"payload"` + // SHA-256 of this event with this field cleared. It makes retransmission + // conflicts detectable without trusting a reused event_seq. + ImmutableEventSha256 []byte `protobuf:"bytes,12,opt,name=immutable_event_sha256,json=immutableEventSha256,proto3" json:"immutable_event_sha256,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *CommandEvent) Reset() { @@ -1964,6 +1967,13 @@ func (x *CommandEvent) GetOutputIncomplete() *OutputIncomplete { return nil } +func (x *CommandEvent) GetImmutableEventSha256() []byte { + if x != nil { + return x.ImmutableEventSha256 + } + return nil +} + type isCommandEvent_Payload interface { isCommandEvent_Payload() } @@ -2211,7 +2221,7 @@ const file_rvbox_v1_common_proto_rawDesc = "" + "\x12ScriptUploadStatus\x12%\n" + "\x0ereceived_bytes\x18\x01 \x01(\x04R\rreceivedBytes\x12\x1a\n" + "\bcomplete\x18\x02 \x01(\bR\bcomplete\x12\x16\n" + - "\x06detail\x18\x03 \x01(\tR\x06detail\"\x91\x05\n" + + "\x06detail\x18\x03 \x01(\tR\x06detail\"\xc7\x05\n" + "\fCommandEvent\x12\x1d\n" + "\n" + "issue_uuid\x18\x01 \x01(\tR\tissueUuid\x12\x1b\n" + @@ -2226,7 +2236,8 @@ const file_rvbox_v1_common_proto_rawDesc = "" + "\rscript_status\x18\t \x01(\v2\x1c.rvbox.v1.ScriptUploadStatusH\x00R\fscriptStatus\x12I\n" + "\x11output_truncation\x18\n" + " \x01(\v2\x1a.rvbox.v1.OutputTruncationH\x00R\x10outputTruncation\x12I\n" + - "\x11output_incomplete\x18\v \x01(\v2\x1a.rvbox.v1.OutputIncompleteH\x00R\x10outputIncompleteB\t\n" + + "\x11output_incomplete\x18\v \x01(\v2\x1a.rvbox.v1.OutputIncompleteH\x00R\x10outputIncomplete\x124\n" + + "\x16immutable_event_sha256\x18\f \x01(\fR\x14immutableEventSha256B\t\n" + "\apayload\"\xb2\x03\n" + "\fControlError\x12/\n" + "\x04code\x18\x01 \x01(\x0e2\x1b.rvbox.v1.ControlError.CodeR\x04code\x12\x18\n" + diff --git a/internal/agentproto/validate.go b/internal/agentproto/validate.go index 92a25d4..6f7e796 100644 --- a/internal/agentproto/validate.go +++ b/internal/agentproto/validate.go @@ -220,12 +220,16 @@ func validateCommandAccepted(accepted *rvboxv1.CommandAccepted, limits Limits) e } func validateCommandEvent(event *rvboxv1.CommandEvent, limits Limits) error { - if event == nil || event.Payload == nil || event.ObservedAt == nil || event.ObservedAt.CheckValid() != nil { + if event == nil || event.Payload == nil || event.ObservedAt == nil || event.ObservedAt.CheckValid() != nil || len(event.ImmutableEventSha256) != sha256.Size { return fmt.Errorf("%w: invalid command event", ErrInvalidEnvelope) } if err := validateIssueAndSequence(event.IssueUuid, event.EventSeq); err != nil { return err } + digest, err := CommandEventDigest(event) + if err != nil || string(digest[:]) != string(event.ImmutableEventSha256) { + return fmt.Errorf("%w: command event immutable digest mismatch", ErrInvalidEnvelope) + } switch payload := event.Payload.(type) { case *rvboxv1.CommandEvent_Output: _, err := DecodeOutputChunk(payload.Output, limits.MaxRawChunkBytes) @@ -265,6 +269,22 @@ func validateCommandEvent(event *rvboxv1.CommandEvent, limits Limits) error { return nil } +// CommandEventDigest is the canonical retransmission digest: protobuf bytes of +// the event with its digest field cleared. Producers populate the field with +// this value; consumers verify it before using event_seq for idempotency. +func CommandEventDigest(event *rvboxv1.CommandEvent) ([sha256.Size]byte, error) { + if event == nil { + return [sha256.Size]byte{}, ErrInvalidEnvelope + } + copyEvent := proto.Clone(event).(*rvboxv1.CommandEvent) + copyEvent.ImmutableEventSha256 = nil + encoded, err := proto.MarshalOptions{Deterministic: true}.Marshal(copyEvent) + if err != nil { + return [sha256.Size]byte{}, err + } + return sha256.Sum256(encoded), nil +} + func validateScriptDescriptor(descriptor *rvboxv1.ScriptDescriptor, limits Limits) error { if descriptor == nil || descriptor.Filename == "" || len(descriptor.Filename) > 255 || descriptor.SizeBytes > limits.MaxScriptBytes || len(descriptor.Sha256) != sha256.Size || !utf8.ValidString(descriptor.Filename) { return fmt.Errorf("%w: invalid script descriptor", ErrInvalidExecutionSpec) diff --git a/internal/agentproto/validate_test.go b/internal/agentproto/validate_test.go index cc8cd9a..b3cbbbf 100644 --- a/internal/agentproto/validate_test.go +++ b/internal/agentproto/validate_test.go @@ -209,7 +209,12 @@ func TestOutputChunkBoundedDecompression_HP_PROTO_05(t *testing.T) { func TestCommandEventPayloadValidation_BH_OUTFLOW_01(t *testing.T) { t.Parallel() - event := &rvboxv1.CommandEvent{IssueUuid: testIssueUUID, EventSeq: 1, ObservedAt: fixedTimestamp(), Payload: &rvboxv1.CommandEvent_Lifecycle{Lifecycle: &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle_COMMAND_RUNNING, CommandRevision: 1}}} + event := &rvboxv1.CommandEvent{IssueUuid: testIssueUUID, EventSeq: 1, ObservedAt: fixedTimestamp(), Payload: &rvboxv1.CommandEvent_Lifecycle{Lifecycle: &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle_COMMAND_RUNNING, CommandRevision: 1}}, ImmutableEventSha256: make([]byte, sha256.Size)} + digest, err := CommandEventDigest(event) + if err != nil { + t.Fatal(err) + } + event.ImmutableEventSha256 = digest[:] envelope := &rvboxv1.AgentEnvelope{SessionId: "session", SessionGeneration: 1, Payload: &rvboxv1.AgentEnvelope_CommandEvent{CommandEvent: event}} if err := ValidateEnvelope(envelope, DefaultLimits(), rvboxv1.Platform_PLATFORM_WINDOWS); err != nil { t.Fatalf("valid event: %v", err) diff --git a/internal/server/session/agent_server.go b/internal/server/session/agent_server.go index 7957254..b0b78f1 100644 --- a/internal/server/session/agent_server.go +++ b/internal/server/session/agent_server.go @@ -241,7 +241,70 @@ func (server *AgentServer) serveConnection(parent context.Context, connection *w server.close(connection, websocket.StatusPolicyViolation, "invalid command acceptance") return } + continue } + if event := envelope.GetCommandEvent(); event != nil { + appendEvent, eventErr := eventAppendFromWire(event, hello.GetClientId(), server.now()) + if eventErr != nil { + server.close(connection, websocket.StatusPolicyViolation, "invalid command event") + return + } + appended, eventErr := server.Store.AppendCommandEvent(sessionContext, appendEvent) + if eventErr != nil { + server.close(connection, websocket.StatusPolicyViolation, "command event was not accepted") + return + } + ack, eventErr := proto.Marshal(&rvboxv1.AgentEnvelope{SessionId: encodedSessionID, SessionGeneration: registration.Generation, Payload: &rvboxv1.AgentEnvelope_EventAck{EventAck: &rvboxv1.EventAck{IssueUuid: event.GetIssueUuid(), ThroughEventSeq: appended.ThroughEventSeq}}}) + if eventErr != nil || queue.EnqueueControl(Frame{Kind: FrameControl, Payload: ack}) != nil { + server.close(connection, websocket.StatusInternalError, "could not acknowledge command event") + return + } + } + } +} + +func eventAppendFromWire(event *rvboxv1.CommandEvent, clientID string, receipt time.Time) (store.EventAppend, error) { + issue, err := domain.ParseUUIDv7(event.GetIssueUuid()) + if err != nil { + return store.EventAppend{}, err + } + payload, err := proto.MarshalOptions{Deterministic: true}.Marshal(event) + if err != nil { + return store.EventAppend{}, err + } + var immutable [32]byte + copy(immutable[:], event.GetImmutableEventSha256()) + result := store.EventAppend{IssueUUID: [16]byte(issue), ClientID: clientID, EventSeq: event.GetEventSeq(), ObservedUnixNano: event.GetObservedAt().AsTime().UnixNano(), ReceiptUnixNano: receipt.UnixNano(), EventType: eventType(event), Compression: 1, RawLength: uint64(len(payload)), Payload: payload, ImmutableSHA256: immutable} + if output := event.GetOutput(); output != nil { + result.Stream = uint16(output.GetStream()) + result.Output = true + } + if result.EventType == 0 { + return store.EventAppend{}, errors.New("unsupported command event") + } + return result, nil +} + +func eventType(event *rvboxv1.CommandEvent) uint16 { + switch event.Payload.(type) { + case *rvboxv1.CommandEvent_Lifecycle: + return 4 + case *rvboxv1.CommandEvent_Output: + return 5 + case *rvboxv1.CommandEvent_Resource: + return 6 + case *rvboxv1.CommandEvent_StdinAck: + return 7 + case *rvboxv1.CommandEvent_SignalResult: + return 8 + case *rvboxv1.CommandEvent_ScriptStatus: + return 9 + case *rvboxv1.CommandEvent_OutputTruncation: + return 10 + case *rvboxv1.CommandEvent_OutputIncomplete: + return 11 + default: + return 0 } } diff --git a/internal/server/session/agent_server_test.go b/internal/server/session/agent_server_test.go index 1269d67..050a170 100644 --- a/internal/server/session/agent_server_test.go +++ b/internal/server/session/agent_server_test.go @@ -13,6 +13,7 @@ import ( "github.com/coder/websocket" rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1" + "github.com/rvbox/rvbox/internal/agentproto" "github.com/rvbox/rvbox/internal/domain" "github.com/rvbox/rvbox/internal/server/store" "google.golang.org/protobuf/proto" @@ -108,6 +109,19 @@ func TestAgentServerRejectsHostileWireInputs_BH_SES_04(t *testing.T) { } } +func TestWireEventAppendCarriesClientBinding_HP_EVENT_01(t *testing.T) { + event := &rvboxv1.CommandEvent{IssueUuid: "019c46f1-1d02-7000-8000-000000000096", EventSeq: 1, ObservedAt: timestamppb.Now(), Payload: &rvboxv1.CommandEvent_Lifecycle{Lifecycle: &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle_COMMAND_RUNNING, CommandRevision: 1}}} + digest, err := agentproto.CommandEventDigest(event) + if err != nil { + t.Fatal(err) + } + event.ImmutableEventSha256 = digest[:] + appendEvent, err := eventAppendFromWire(event, "client-a", time.Now()) + if err != nil || appendEvent.ClientID != "client-a" || appendEvent.EventSeq != 1 || appendEvent.EventType != 4 || appendEvent.ImmutableSHA256 != digest { + t.Fatalf("wire event append = %#v, %v", appendEvent, err) + } +} + func newTestAgentServer(t *testing.T) (*httptest.Server, func()) { t.Helper() dataDirectory := filepath.Join(t.TempDir(), "store") diff --git a/internal/server/store/event.go b/internal/server/store/event.go index d34f74f..738a3b3 100644 --- a/internal/server/store/event.go +++ b/internal/server/store/event.go @@ -33,6 +33,7 @@ func (function FaultInjectorFunc) Checkpoint(name string) error { return functio type EventAppend struct { IssueUUID [16]byte + ClientID string EventSeq uint64 ObservedUnixNano int64 ReceiptUnixNano int64 @@ -80,11 +81,17 @@ func (store *Store) AppendCommandEvent(ctx context.Context, event EventAppend) ( } var lastSequence, commandOutputCharged, commandCharged, clientCharged, serverCharged, closeoutRemaining uint64 var clientID string - err = database.QueryRowContext(ctx, `SELECT commands.last_event_seq, commands.output_charged_bytes, + query := `SELECT commands.last_event_seq, commands.output_charged_bytes, commands.charged_bytes, commands.closeout_remaining_bytes, commands.client_id, clients.charged_bytes, storage_counters.command_charged_bytes FROM commands JOIN clients ON clients.client_id = commands.client_id -JOIN storage_counters ON storage_counters.singleton = 1 WHERE commands.issue_uuid = ?`, event.IssueUUID[:]).Scan( +JOIN storage_counters ON storage_counters.singleton = 1 WHERE commands.issue_uuid = ?` + arguments := []any{event.IssueUUID[:]} + if event.ClientID != "" { + query += ` AND commands.client_id = ?` + arguments = append(arguments, event.ClientID) + } + err = database.QueryRowContext(ctx, query, arguments...).Scan( &lastSequence, &commandOutputCharged, &commandCharged, &closeoutRemaining, &clientID, &clientCharged, &serverCharged) if errors.Is(err, sql.ErrNoRows) { return result, ErrCommandNotFound diff --git a/protos/rvbox/v1/common.proto b/protos/rvbox/v1/common.proto index 124b1eb..f8220e2 100644 --- a/protos/rvbox/v1/common.proto +++ b/protos/rvbox/v1/common.proto @@ -254,6 +254,9 @@ message CommandEvent { OutputTruncation output_truncation = 10; OutputIncomplete output_incomplete = 11; } + // SHA-256 of this event with this field cleared. It makes retransmission + // conflicts detectable without trusting a reused event_seq. + bytes immutable_event_sha256 = 12; } message ControlError { diff --git a/test/coverage.toml b/test/coverage.toml index cdbf5d8..4574d63 100644 --- a/test/coverage.toml +++ b/test/coverage.toml @@ -218,6 +218,12 @@ layer = "unit" status = "implemented" tests = ["internal/server/store/command_test.go:TestRecordCommandAcceptanceFencesGeneration_HP_DISPATCH_05"] +[[requirements]] +id = "HP-EVENT-01" +layer = "unit" +status = "implemented" +tests = ["internal/server/session/agent_server_test.go:TestWireEventAppendCarriesClientBinding_HP_EVENT_01"] + [[requirements]] id = "HP-SES-05" layer = "integration"