return Publication{}, fmt.Errorf("build address: %w", err)
}
- data, err := encodeEnvelope(metadata, payload)
+ normalized := normalizePayload(payload)
+ err = validatePayload(address, metadata, normalized)
if err != nil {
- return Publication{}, fmt.Errorf("build envelope: %w", err)
+ return Publication{}, fmt.Errorf("validate publication: %w", err)
}
- retain := address.Kind == Descriptor || address.Kind == Availability || address.Kind == Reported || address.Kind == HubStatus
- _, err = parseMessage[T](address.Kind, topic, data, retain)
+ data, err := encodeEnvelope(metadata, normalized)
if err != nil {
- return Publication{}, fmt.Errorf("validate publication: %w", err)
+ return Publication{}, fmt.Errorf("build envelope: %w", err)
}
+ retain := address.Kind == Descriptor || address.Kind == Availability || address.Kind == Reported || address.Kind == HubStatus
+
qos := byte(1)
if address.Kind == Telemetry {
qos = 0
return Message[T]{}, fmt.Errorf("message topic does not match parser")
}
+ message, err := parsePayload[T](address, data, retained)
+ if err != nil {
+ return Message[T]{}, fmt.Errorf("parse payload: %w", err)
+ }
+
+ return message, nil
+}
+
+func parsePayload[T any](address Address, data []byte, retained bool) (Message[T], error) {
+ kind := address.Kind
+
if retained && (kind == Telemetry || kind == Event || kind == Command || kind == CommandBatch || kind == Result) {
return Message[T]{}, fmt.Errorf("message must not be retained")
}
return Message[T]{}, fmt.Errorf("decode payload: %w", err)
}
- err = validatePayload(address, envelope, payload)
+ err = validatePayload(address, envelope.Metadata, payload)
if err != nil {
return Message[T]{}, fmt.Errorf("validate payload: %w", err)
}
+ if address.Kind == Descriptor {
+ err = validateDescriptorFlags(envelope.Payload)
+ if err != nil {
+ return Message[T]{}, fmt.Errorf("validate descriptor access flags: %w", err)
+ }
+ }
+
return Message[T]{
Address: address,
Metadata: envelope.Metadata,
}, nil
}
-func validatePayload(address Address, envelope wireEnvelope, payload any) error {
+// normalizePayload keeps nil raw values as JSON null without changing caller
+// data.
+func normalizePayload(payload any) any {
switch value := payload.(type) {
- case DescriptorPayload:
- err := validateDescriptor(value, envelope.Payload)
- if err != nil {
- return fmt.Errorf("validate descriptor: %w", err)
- }
- case AvailabilityPayload:
- if value.Status != "online" && value.Status != "offline" {
- return fmt.Errorf("availability must be online or offline")
- }
case ValuePayload:
- if !json.Valid(value.Value) {
- return fmt.Errorf("value is required and must be valid JSON")
+ if value.Value == nil {
+ value.Value = json.RawMessage(`null`)
}
- if address.Kind == Command && envelope.ExpiresAt == nil {
- return fmt.Errorf("command must have an expiry")
- }
- case EventPayload:
- // An event object has application-defined fields.
- case ResultPayload:
- if envelope.CorrelationID != address.Target {
- return fmt.Errorf("result correlation must match its command ID")
- }
+ return value
+ case BatchPayload:
+ commands := make([]BatchCommand, len(value.Commands))
+ copy(commands, value.Commands)
- switch value.Status {
- case "accepted", "completed":
- if value.Reason != "" {
- return fmt.Errorf("successful result must not have a reason")
- }
- case "rejected", "failed":
- if value.Reason == "" {
- return fmt.Errorf("unsuccessful result must have a reason")
+ for index := range commands {
+ if commands[index].Payload.Value == nil {
+ commands[index].Payload.Value = json.RawMessage(`null`)
}
- default:
- return fmt.Errorf("result status is invalid")
}
+
+ value.Commands = commands
+
+ return value
+ default:
+ return payload
+ }
+}
+
+func validatePayload(address Address, metadata Metadata, payload any) error {
+ var err error
+
+ switch value := payload.(type) {
+ case DescriptorPayload:
+ err = validateDescriptor(value)
+ case AvailabilityPayload:
+ err = validateAvailability(value)
+ case ValuePayload:
+ err = validateValue(value, address.Kind, metadata)
+ case EventPayload:
+ err = validateEvent(value)
+ case ResultPayload:
+ err = validateResult(value, address, metadata)
case BatchPayload:
- err := validateBatch(value, envelope)
- if err != nil {
- return fmt.Errorf("validate command batch: %w", err)
- }
+ err = validateBatch(value, metadata)
case HubStatusPayload:
- err := validateHub(value, address, envelope)
- if err != nil {
- return fmt.Errorf("validate hub status: %w", err)
- }
+ err = validateHub(value, address, metadata)
default:
return fmt.Errorf("unsupported payload type")
}
+ if err != nil {
+ return fmt.Errorf("validate %s payload: %w", address.Kind, err)
+ }
+
return nil
}
-func validateDescriptor(payload DescriptorPayload, raw json.RawMessage) error {
- if payload.Manufacturer == "" || payload.Model == "" || payload.Firmware == "" || len(payload.Attribution) > 2000 || len(payload.Capabilities) == 0 {
+func validateDescriptor(payload DescriptorPayload) error {
+ if payload.Manufacturer == "" || payload.Model == "" || payload.Firmware == "" || len(string([]rune(payload.Attribution))) > 2000 || len(payload.Capabilities) == 0 {
return fmt.Errorf("descriptor metadata or capabilities are invalid")
}
- var fields struct {
- Capabilities []map[string]json.RawMessage `json:"capabilities"`
- }
-
- err := json.Unmarshal(raw, &fields)
- if err != nil {
- return fmt.Errorf("read descriptor access flags: %w", err)
- }
-
seen := make(map[string]bool, len(payload.Capabilities))
- for index, capability := range payload.Capabilities {
+ for _, capability := range payload.Capabilities {
if !ValidIdentifier(capability.ID) || !ValidIdentifier(capability.Type) || seen[capability.ID] {
return fmt.Errorf("capability IDs and types must be valid and unique")
}
return fmt.Errorf("capability must have at least one access mode")
}
+ seen[capability.ID] = true
+ }
+
+ return nil
+}
+
+func validateDescriptorFlags(raw json.RawMessage) error {
+ var fields struct {
+ Capabilities []map[string]json.RawMessage `json:"capabilities"`
+ }
+
+ err := json.Unmarshal(raw, &fields)
+ if err != nil {
+ return fmt.Errorf("read descriptor access flags: %w", err)
+ }
+
+ for _, capability := range fields.Capabilities {
for _, name := range []string{"reports_state", "accepts_commands"} {
- field := fields.Capabilities[index][name]
+ field := capability[name]
if string(field) != "true" && string(field) != "false" {
return fmt.Errorf("capability access flags must be present booleans")
}
}
+ }
- seen[capability.ID] = true
+ return nil
+}
+
+func validateAvailability(payload AvailabilityPayload) error {
+ if payload.Status != "online" && payload.Status != "offline" {
+ return fmt.Errorf("availability must be online or offline")
}
return nil
}
-func validateBatch(payload BatchPayload, envelope wireEnvelope) error {
+func validateValue(payload ValuePayload, kind Kind, metadata Metadata) error {
+ if !json.Valid(payload.Value) {
+ return fmt.Errorf("value is required and must be valid JSON")
+ }
+
+ if kind == Command && metadata.ExpiresAt == nil {
+ return fmt.Errorf("command must have an expiry")
+ }
+
+ return nil
+}
+
+func validateEvent(payload EventPayload) error {
+ if payload == nil {
+ return fmt.Errorf("event must be a JSON object")
+ }
+
+ for _, value := range payload {
+ if value != nil && !json.Valid(value) {
+ return fmt.Errorf("event values must be valid JSON")
+ }
+ }
+
+ return nil
+}
+
+func validateResult(payload ResultPayload, address Address, metadata Metadata) error {
+ if metadata.CorrelationID != address.Target {
+ return fmt.Errorf("result correlation must match its command ID")
+ }
+
+ switch payload.Status {
+ case "accepted", "completed":
+ if payload.Reason != "" {
+ return fmt.Errorf("successful result must not have a reason")
+ }
+ case "rejected", "failed":
+ if payload.Reason == "" {
+ return fmt.Errorf("unsuccessful result must have a reason")
+ }
+ default:
+ return fmt.Errorf("result status is invalid")
+ }
+
+ return nil
+}
+
+func validateBatch(payload BatchPayload, metadata Metadata) error {
if len(payload.Commands) < 2 || len(payload.Commands) > MaximumCommandBatchSize {
return fmt.Errorf("batch size must be 2 through %d", MaximumCommandBatchSize)
}
- if envelope.MessageID != payload.Commands[0].MessageID {
+ if metadata.MessageID != payload.Commands[0].MessageID {
return fmt.Errorf("batch ID must equal its first command ID")
}
return fmt.Errorf("batch targets and IDs must be valid and distinct")
}
- _, err := encodeEnvelope(command.Metadata, command.Payload)
+ err := validateMetadata(command.Metadata)
if err != nil {
- return fmt.Errorf("validate batch command envelope: %w", err)
+ return fmt.Errorf("validate batch command metadata: %w", err)
}
- if command.ExpiresAt == nil || !json.Valid(command.Payload.Value) {
- return fmt.Errorf("batch commands require expiry and a JSON value")
+ err = validateValue(command.Payload, Command, command.Metadata)
+ if err != nil {
+ return fmt.Errorf("validate batch command value: %w", err)
}
ids[command.MessageID] = true
return nil
}
-func validateHub(payload HubStatusPayload, address Address, envelope wireEnvelope) error {
- if payload.ID != address.ID || envelope.ExpiresAt == nil {
+func validateHub(payload HubStatusPayload, address Address, metadata Metadata) error {
+ if payload.ID != address.ID || metadata.ExpiresAt == nil {
return fmt.Errorf("hub status requires a matching ID and an expiry")
}
_, err = BuildDescriptor(deviceID, metadata, descriptor)
require.Error(t, err)
}
+
+func TestBuildersKeepNullValuesWithoutChangingInputs(t *testing.T) {
+ metadata := Metadata{MessageID: messageID}
+ payload := ValuePayload{}
+ publication, err := BuildReported(deviceID, "relay", metadata, payload)
+ require.NoError(t, err)
+ require.Nil(t, payload.Value)
+
+ parsed, err := ParseReported(publication.Topic, publication.Data, false)
+ require.NoError(t, err)
+ require.JSONEq(t, `null`, string(parsed.Payload.Value))
+
+ deadline := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+ batch := testBatch(deadline)
+ batch.Commands[0].Payload.Value = nil
+ publication, err = BuildCommandBatch(deviceID, metadata, batch)
+ require.NoError(t, err)
+ require.Nil(t, batch.Commands[0].Payload.Value)
+
+ commands, err := ParseCommandBatch(publication.Topic, publication.Data, false)
+ require.NoError(t, err)
+ require.JSONEq(t, `null`, string(commands.Payload.Commands[0].Payload.Value))
+}
+
+func TestBuildEventRejectsNonObjectAndInvalidValues(t *testing.T) {
+ testCases := []struct {
+ name string
+ payload EventPayload
+ valid bool
+ }{
+ {
+ name: "rejects null instead of an object",
+ },
+ {
+ name: "rejects malformed event data",
+ payload: EventPayload{"action": json.RawMessage(`bad`)},
+ },
+ {
+ name: "rejects empty nonnull event data",
+ payload: EventPayload{"action": json.RawMessage{}},
+ },
+ {
+ name: "keeps null event members",
+ payload: EventPayload{"action": nil},
+ valid: true,
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ publication, err := BuildEvent(deviceID, "press", Metadata{MessageID: messageID}, tc.payload)
+ if !tc.valid {
+ require.Error(t, err)
+
+ return
+ }
+
+ require.NoError(t, err)
+
+ parsed, err := ParseEvent(publication.Topic, publication.Data, false)
+ require.NoError(t, err)
+ require.JSONEq(t, `null`, string(parsed.Payload["action"]))
+ })
+ }
+}
+
+func TestBuildRejectsInvalidMetadataBeforeReturningPublications(t *testing.T) {
+ deadline := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+ metadata := Metadata{MessageID: messageID}
+ testCases := []struct {
+ name string
+ mutate func(*Metadata)
+ }{
+ {
+ name: "rejects malformed message identity",
+ mutate: func(metadata *Metadata) {
+ metadata.MessageID = "bad"
+ },
+ },
+ {
+ name: "rejects malformed correlation identity",
+ mutate: func(metadata *Metadata) {
+ metadata.CorrelationID = "bad"
+ },
+ },
+ {
+ name: "rejects malformed causation identity",
+ mutate: func(metadata *Metadata) {
+ metadata.CausationID = "bad"
+ },
+ },
+ {
+ name: "rejects non UTC expiry",
+ mutate: func(metadata *Metadata) {
+ expiry := deadline.In(time.FixedZone("test", 3600))
+ metadata.ExpiresAt = &expiry
+ },
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ invalid := metadata
+ tc.mutate(&invalid)
+
+ _, err := BuildReported(deviceID, "relay", invalid, ValuePayload{Value: json.RawMessage(`true`)})
+ require.Error(t, err)
+
+ batch := testBatch(deadline)
+ tc.mutate(&batch.Commands[1].Metadata)
+
+ _, err = BuildCommandBatch(deviceID, metadata, batch)
+ require.Error(t, err)
+ })
+ }
+}
+
+func TestDescriptorAttributionLimitUsesEncodedText(t *testing.T) {
+ payload := testDescriptor()
+ payload.Attribution = strings.Repeat("\xff", 667)
+ _, err := BuildDescriptor(deviceID, Metadata{MessageID: messageID}, payload)
+ require.Error(t, err)
+
+ payload.Attribution = strings.Repeat("\xff", 666)
+ publication, err := BuildDescriptor(deviceID, Metadata{MessageID: messageID}, payload)
+ require.NoError(t, err)
+
+ parsed, err := ParseDescriptor(publication.Topic, publication.Data, true)
+ require.NoError(t, err)
+ require.Len(t, parsed.Payload.Attribution, 1998)
+}
projects / onehouse / commitdiff