fix: carry immutable hash in command dispatch
This commit is contained in:
@@ -534,8 +534,10 @@ type CommandDispatch struct {
|
|||||||
IssueTime *timestamppb.Timestamp `protobuf:"bytes,4,opt,name=issue_time,json=issueTime,proto3" json:"issue_time,omitempty"`
|
IssueTime *timestamppb.Timestamp `protobuf:"bytes,4,opt,name=issue_time,json=issueTime,proto3" json:"issue_time,omitempty"`
|
||||||
Spec *ExecutionSpec `protobuf:"bytes,5,opt,name=spec,proto3" json:"spec,omitempty"`
|
Spec *ExecutionSpec `protobuf:"bytes,5,opt,name=spec,proto3" json:"spec,omitempty"`
|
||||||
QueueExpiryTime *timestamppb.Timestamp `protobuf:"bytes,6,opt,name=queue_expiry_time,json=queueExpiryTime,proto3" json:"queue_expiry_time,omitempty"`
|
QueueExpiryTime *timestamppb.Timestamp `protobuf:"bytes,6,opt,name=queue_expiry_time,json=queueExpiryTime,proto3" json:"queue_expiry_time,omitempty"`
|
||||||
unknownFields protoimpl.UnknownFields
|
// Required immutable digest for client-side UUID/hash idempotency checks.
|
||||||
sizeCache protoimpl.SizeCache
|
ImmutableRequestSha256 []byte `protobuf:"bytes,7,opt,name=immutable_request_sha256,json=immutableRequestSha256,proto3" json:"immutable_request_sha256,omitempty"`
|
||||||
|
unknownFields protoimpl.UnknownFields
|
||||||
|
sizeCache protoimpl.SizeCache
|
||||||
}
|
}
|
||||||
|
|
||||||
func (x *CommandDispatch) Reset() {
|
func (x *CommandDispatch) Reset() {
|
||||||
@@ -610,6 +612,13 @@ func (x *CommandDispatch) GetQueueExpiryTime() *timestamppb.Timestamp {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (x *CommandDispatch) GetImmutableRequestSha256() []byte {
|
||||||
|
if x != nil {
|
||||||
|
return x.ImmutableRequestSha256
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
type CommandAccepted struct {
|
type CommandAccepted struct {
|
||||||
state protoimpl.MessageState `protogen:"open.v1"`
|
state protoimpl.MessageState `protogen:"open.v1"`
|
||||||
IssueUuid string `protobuf:"bytes,1,opt,name=issue_uuid,json=issueUuid,proto3" json:"issue_uuid,omitempty"`
|
IssueUuid string `protobuf:"bytes,1,opt,name=issue_uuid,json=issueUuid,proto3" json:"issue_uuid,omitempty"`
|
||||||
@@ -1502,7 +1511,7 @@ const file_rvbox_v1_agent_proto_rawDesc = "" +
|
|||||||
"\rServerWelcome\x12F\n" +
|
"\rServerWelcome\x12F\n" +
|
||||||
"\x11selected_protocol\x18\x01 \x01(\v2\x19.rvbox.v1.ProtocolVersionR\x10selectedProtocol\x12;\n" +
|
"\x11selected_protocol\x18\x01 \x01(\v2\x19.rvbox.v1.ProtocolVersionR\x10selectedProtocol\x12;\n" +
|
||||||
"\vserver_time\x18\x02 \x01(\v2\x1a.google.protobuf.TimestampR\n" +
|
"\vserver_time\x18\x02 \x01(\v2\x1a.google.protobuf.TimestampR\n" +
|
||||||
"serverTime\"\xc7\x02\n" +
|
"serverTime\"\x81\x03\n" +
|
||||||
"\x0fCommandDispatch\x12\x1d\n" +
|
"\x0fCommandDispatch\x12\x1d\n" +
|
||||||
"\n" +
|
"\n" +
|
||||||
"issue_uuid\x18\x01 \x01(\tR\tissueUuid\x12)\n" +
|
"issue_uuid\x18\x01 \x01(\tR\tissueUuid\x12)\n" +
|
||||||
@@ -1511,7 +1520,8 @@ const file_rvbox_v1_agent_proto_rawDesc = "" +
|
|||||||
"\n" +
|
"\n" +
|
||||||
"issue_time\x18\x04 \x01(\v2\x1a.google.protobuf.TimestampR\tissueTime\x12+\n" +
|
"issue_time\x18\x04 \x01(\v2\x1a.google.protobuf.TimestampR\tissueTime\x12+\n" +
|
||||||
"\x04spec\x18\x05 \x01(\v2\x17.rvbox.v1.ExecutionSpecR\x04spec\x12F\n" +
|
"\x04spec\x18\x05 \x01(\v2\x17.rvbox.v1.ExecutionSpecR\x04spec\x12F\n" +
|
||||||
"\x11queue_expiry_time\x18\x06 \x01(\v2\x1a.google.protobuf.TimestampR\x0fqueueExpiryTime\"\xad\x01\n" +
|
"\x11queue_expiry_time\x18\x06 \x01(\v2\x1a.google.protobuf.TimestampR\x0fqueueExpiryTime\x128\n" +
|
||||||
|
"\x18immutable_request_sha256\x18\a \x01(\fR\x16immutableRequestSha256\"\xad\x01\n" +
|
||||||
"\x0fCommandAccepted\x12\x1d\n" +
|
"\x0fCommandAccepted\x12\x1d\n" +
|
||||||
"\n" +
|
"\n" +
|
||||||
"issue_uuid\x18\x01 \x01(\tR\tissueUuid\x12)\n" +
|
"issue_uuid\x18\x01 \x01(\tR\tissueUuid\x12)\n" +
|
||||||
|
|||||||
@@ -188,7 +188,7 @@ func validateClientHello(hello *rvboxv1.ClientHello) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func validateCommandDispatch(dispatch *rvboxv1.CommandDispatch, limits Limits, peerPlatform rvboxv1.Platform) error {
|
func validateCommandDispatch(dispatch *rvboxv1.CommandDispatch, limits Limits, peerPlatform rvboxv1.Platform) error {
|
||||||
if dispatch == nil || dispatch.CommandRevision == 0 || dispatch.TargetSessionGeneration == 0 || dispatch.IssueTime == nil || dispatch.IssueTime.CheckValid() != nil {
|
if dispatch == nil || dispatch.CommandRevision == 0 || dispatch.TargetSessionGeneration == 0 || dispatch.IssueTime == nil || dispatch.IssueTime.CheckValid() != nil || len(dispatch.ImmutableRequestSha256) != sha256.Size {
|
||||||
return fmt.Errorf("%w: invalid command dispatch", ErrInvalidEnvelope)
|
return fmt.Errorf("%w: invalid command dispatch", ErrInvalidEnvelope)
|
||||||
}
|
}
|
||||||
if err := validateIssueUUID(dispatch.IssueUuid); err != nil {
|
if err := validateIssueUUID(dispatch.IssueUuid); err != nil {
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package agentproto
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
|
"crypto/sha256"
|
||||||
"errors"
|
"errors"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
@@ -140,7 +141,7 @@ func TestWindowsEnvironmentCaseCollision_BH_LAUNCH_01(t *testing.T) {
|
|||||||
func TestCommandDispatchUUIDRevisionAndTimestamp_BH_IDEM_01(t *testing.T) {
|
func TestCommandDispatchUUIDRevisionAndTimestamp_BH_IDEM_01(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
dispatch := &rvboxv1.CommandDispatch{IssueUuid: testIssueUUID, CommandRevision: 1, TargetSessionGeneration: 2, IssueTime: fixedTimestamp(), Spec: validCommandSpec()}
|
dispatch := &rvboxv1.CommandDispatch{IssueUuid: testIssueUUID, CommandRevision: 1, TargetSessionGeneration: 2, IssueTime: fixedTimestamp(), Spec: validCommandSpec(), ImmutableRequestSha256: make([]byte, sha256.Size)}
|
||||||
envelope := &rvboxv1.AgentEnvelope{SessionId: "session", SessionGeneration: 2, Payload: &rvboxv1.AgentEnvelope_CommandDispatch{CommandDispatch: dispatch}}
|
envelope := &rvboxv1.AgentEnvelope{SessionId: "session", SessionGeneration: 2, Payload: &rvboxv1.AgentEnvelope_CommandDispatch{CommandDispatch: dispatch}}
|
||||||
if err := ValidateEnvelope(envelope, DefaultLimits(), rvboxv1.Platform_PLATFORM_LINUX); err != nil {
|
if err := ValidateEnvelope(envelope, DefaultLimits(), rvboxv1.Platform_PLATFORM_LINUX); err != nil {
|
||||||
t.Fatalf("valid dispatch: %v", err)
|
t.Fatalf("valid dispatch: %v", err)
|
||||||
@@ -149,6 +150,7 @@ func TestCommandDispatchUUIDRevisionAndTimestamp_BH_IDEM_01(t *testing.T) {
|
|||||||
func(value *rvboxv1.CommandDispatch) { value.IssueUuid = "not-a-uuid" },
|
func(value *rvboxv1.CommandDispatch) { value.IssueUuid = "not-a-uuid" },
|
||||||
func(value *rvboxv1.CommandDispatch) { value.CommandRevision = 0 },
|
func(value *rvboxv1.CommandDispatch) { value.CommandRevision = 0 },
|
||||||
func(value *rvboxv1.CommandDispatch) { value.TargetSessionGeneration = 0 },
|
func(value *rvboxv1.CommandDispatch) { value.TargetSessionGeneration = 0 },
|
||||||
|
func(value *rvboxv1.CommandDispatch) { value.ImmutableRequestSha256 = nil },
|
||||||
func(value *rvboxv1.CommandDispatch) {
|
func(value *rvboxv1.CommandDispatch) {
|
||||||
value.IssueTime = timestamppb.New(time.Date(10000, 1, 1, 0, 0, 0, 0, time.UTC))
|
value.IssueTime = timestamppb.New(time.Date(10000, 1, 1, 0, 0, 0, 0, time.UTC))
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -257,7 +257,7 @@ func (server *AgentServer) enqueueNextDispatch(ctx context.Context, queue *Write
|
|||||||
}
|
}
|
||||||
dispatch := &rvboxv1.CommandDispatch{
|
dispatch := &rvboxv1.CommandDispatch{
|
||||||
IssueUuid: candidate.IssueUUID.String(), CommandRevision: candidate.Revision, TargetSessionGeneration: generation,
|
IssueUuid: candidate.IssueUUID.String(), CommandRevision: candidate.Revision, TargetSessionGeneration: generation,
|
||||||
IssueTime: timestamppb.New(candidate.IssueTime), Spec: spec,
|
IssueTime: timestamppb.New(candidate.IssueTime), Spec: spec, ImmutableRequestSha256: candidate.ImmutableSHA256[:],
|
||||||
}
|
}
|
||||||
if candidate.QueueExpiryTime != nil {
|
if candidate.QueueExpiryTime != nil {
|
||||||
dispatch.QueueExpiryTime = timestamppb.New(*candidate.QueueExpiryTime)
|
dispatch.QueueExpiryTime = timestamppb.New(*candidate.QueueExpiryTime)
|
||||||
|
|||||||
@@ -23,6 +23,7 @@ type DispatchCandidate struct {
|
|||||||
Revision uint64
|
Revision uint64
|
||||||
IssueTime time.Time
|
IssueTime time.Time
|
||||||
QueueExpiryTime *time.Time
|
QueueExpiryTime *time.Time
|
||||||
|
ImmutableSHA256 [32]byte
|
||||||
ExecutionSpec []byte
|
ExecutionSpec []byte
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -51,13 +52,13 @@ func (store *Store) ClaimNextDispatch(ctx context.Context, clientID string, gene
|
|||||||
WHERE client_id = ? AND lifecycle = 1 AND queue_expiry_time IS NOT NULL AND queue_expiry_time <= ?`, now.UTC().UnixNano(), clientID, now.UTC().UnixNano()); err != nil {
|
WHERE client_id = ? AND lifecycle = 1 AND queue_expiry_time IS NOT NULL AND queue_expiry_time <= ?`, now.UTC().UnixNano(), clientID, now.UTC().UnixNano()); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
var encodedIssue, stored []byte
|
var encodedIssue, stored, immutableHash []byte
|
||||||
var revision uint64
|
var revision uint64
|
||||||
var issueTime int64
|
var issueTime int64
|
||||||
var expiry sql.NullInt64
|
var expiry sql.NullInt64
|
||||||
var rawBytes uint64
|
var rawBytes uint64
|
||||||
err = tx.QueryRowContext(ctx, `SELECT issue_uuid, revision, issue_time, queue_expiry_time, execution_spec, execution_spec_raw_bytes
|
err = tx.QueryRowContext(ctx, `SELECT issue_uuid, revision, issue_time, queue_expiry_time, immutable_request_sha256, execution_spec, execution_spec_raw_bytes
|
||||||
FROM commands WHERE client_id = ? AND lifecycle = 1 ORDER BY issue_time, issue_uuid LIMIT 1`, clientID).Scan(&encodedIssue, &revision, &issueTime, &expiry, &stored, &rawBytes)
|
FROM commands WHERE client_id = ? AND lifecycle = 1 ORDER BY issue_time, issue_uuid LIMIT 1`, clientID).Scan(&encodedIssue, &revision, &issueTime, &expiry, &immutableHash, &stored, &rawBytes)
|
||||||
if errors.Is(err, sql.ErrNoRows) {
|
if errors.Is(err, sql.ErrNoRows) {
|
||||||
if err := tx.Commit(); err != nil {
|
if err := tx.Commit(); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@@ -67,11 +68,13 @@ FROM commands WHERE client_id = ? AND lifecycle = 1 ORDER BY issue_time, issue_u
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if len(encodedIssue) != 16 {
|
if len(encodedIssue) != 16 || len(immutableHash) != 32 {
|
||||||
return nil, ErrInvalidSegmentRecord
|
return nil, ErrInvalidSegmentRecord
|
||||||
}
|
}
|
||||||
var issue domain.UUID
|
var issue domain.UUID
|
||||||
copy(issue[:], encodedIssue)
|
copy(issue[:], encodedIssue)
|
||||||
|
var immutable [32]byte
|
||||||
|
copy(immutable[:], immutableHash)
|
||||||
spec, err := decompressCommandSpec(stored, rawBytes)
|
spec, err := decompressCommandSpec(stored, rawBytes)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@@ -90,7 +93,7 @@ FROM commands WHERE client_id = ? AND lifecycle = 1 ORDER BY issue_time, issue_u
|
|||||||
if err := tx.Commit(); err != nil {
|
if err := tx.Commit(); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
candidate := &DispatchCandidate{IssueUUID: issue, Revision: revision, IssueTime: time.Unix(0, issueTime).UTC(), ExecutionSpec: spec}
|
candidate := &DispatchCandidate{IssueUUID: issue, Revision: revision, IssueTime: time.Unix(0, issueTime).UTC(), ImmutableSHA256: immutable, ExecutionSpec: spec}
|
||||||
if expiry.Valid {
|
if expiry.Valid {
|
||||||
value := time.Unix(0, expiry.Int64).UTC()
|
value := time.Unix(0, expiry.Int64).UTC()
|
||||||
candidate.QueueExpiryTime = &value
|
candidate.QueueExpiryTime = &value
|
||||||
|
|||||||
@@ -63,6 +63,8 @@ message CommandDispatch {
|
|||||||
google.protobuf.Timestamp issue_time = 4;
|
google.protobuf.Timestamp issue_time = 4;
|
||||||
ExecutionSpec spec = 5;
|
ExecutionSpec spec = 5;
|
||||||
google.protobuf.Timestamp queue_expiry_time = 6;
|
google.protobuf.Timestamp queue_expiry_time = 6;
|
||||||
|
// Required immutable digest for client-side UUID/hash idempotency checks.
|
||||||
|
bytes immutable_request_sha256 = 7;
|
||||||
}
|
}
|
||||||
|
|
||||||
message CommandAccepted {
|
message CommandAccepted {
|
||||||
|
|||||||
@@ -106,7 +106,7 @@ func TestWebSocketDispatchAfterReconciliation_HP_DISPATCH_03(t *testing.T) {
|
|||||||
t.Fatalf("dispatch envelope = %#v, %v", envelope, err)
|
t.Fatalf("dispatch envelope = %#v, %v", envelope, err)
|
||||||
}
|
}
|
||||||
dispatch := envelope.GetCommandDispatch()
|
dispatch := envelope.GetCommandDispatch()
|
||||||
if dispatch == nil || dispatch.GetIssueUuid() != issue.String() || dispatch.GetCommandRevision() != 1 || dispatch.GetTargetSessionGeneration() != accepted.Generation || !proto.Equal(dispatch.GetSpec(), spec) {
|
if dispatch == nil || dispatch.GetIssueUuid() != issue.String() || dispatch.GetCommandRevision() != 1 || dispatch.GetTargetSessionGeneration() != accepted.Generation || !proto.Equal(dispatch.GetSpec(), spec) || string(dispatch.GetImmutableRequestSha256()) != string(requestHash[:]) {
|
||||||
t.Fatalf("dispatch = %#v", dispatch)
|
t.Fatalf("dispatch = %#v", dispatch)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user