From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 21:46:59 +0000 (-0300) Subject: Validate typed publications before encoding X-Git-Url: https://git.giacomolli.uk/onehouse/commitdiff_plain/c376881c67694cd9931d80fb4a04ec69751d222b?ds=sidebyside;hp=4bba0e163f1a4dc07b34fd9e4dc403de1f9b4cb9 Validate typed publications before encoding Builders share payload validation with parsers without encoding and parsing a publication for validation. Batch members use the shared metadata and value checks without envelope encoding. Access flag presence is a JSON input check, separate from typed descriptor checks. Nil raw values still encode as null without changing caller data, and the attribution limit measures the text that JSON encodes. Issue: 8fa2f25 --- diff --git a/doc/mqtt.rst b/doc/mqtt.rst index 1f5fa4a..d13015e 100644 --- a/doc/mqtt.rst +++ b/doc/mqtt.rst @@ -30,9 +30,12 @@ run timers or implement publication suppression. Each operation has a ``Build`` function and a ``Parse`` function, such as ``BuildReported`` and ``ParseReported``. Builders accept an ID, an optional target, metadata, and a typed payload. They validate the same -contract as the parsers. ``Parse`` selects the typed payload and address -kind from an incoming topic. A caller does not split or join topics and -does not build an envelope. +contract as the parsers through shared typed validation. A builder +encodes its validated payload without parsing it. Batch members use the +same metadata and value validation without envelope encoding. Access +flag presence checks apply to parsed JSON only. ``Parse`` selects the +typed payload and address kind from an incoming topic. A caller does not +split or join topics and does not build an envelope. A ``Publication`` contains topic, JSON bytes, QoS, and retention. It also supplies the expiry deadline, response topic, and correlation data diff --git a/protocol/message.go b/protocol/message.go index 0fd4ec9..cb575d2 100644 --- a/protocol/message.go +++ b/protocol/message.go @@ -83,17 +83,19 @@ func buildMessage[T any](address Address, metadata Metadata, payload T) (Publica 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 @@ -146,6 +148,17 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M 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") } @@ -162,11 +175,18 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M 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, @@ -175,78 +195,71 @@ func parseMessage[T any](kind Kind, topic string, data []byte, retained bool) (M }, 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") } @@ -255,25 +268,95 @@ func validateDescriptor(payload DescriptorPayload, raw json.RawMessage) error { 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") } @@ -285,13 +368,14 @@ func validateBatch(payload BatchPayload, envelope wireEnvelope) error { 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 @@ -301,8 +385,8 @@ func validateBatch(payload BatchPayload, envelope wireEnvelope) error { 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") } diff --git a/protocol/message_test.go b/protocol/message_test.go index 79871dd..660885d 100644 --- a/protocol/message_test.go +++ b/protocol/message_test.go @@ -294,3 +294,134 @@ func TestBuildersApplyCommandAndResultRules(t *testing.T) { _, 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) +} diff --git a/protocol/wire_envelope.go b/protocol/wire_envelope.go index 5e8b778..c372346 100644 --- a/protocol/wire_envelope.go +++ b/protocol/wire_envelope.go @@ -77,19 +77,32 @@ func validateEnvelope(envelope wireEnvelope) error { return fmt.Errorf("unsupported schema %d", envelope.Schema) } - if !ValidID(envelope.MessageID) { + err := validateMetadata(envelope.Metadata) + if err != nil { + return fmt.Errorf("validate metadata: %w", err) + } + + if !isObject(envelope.Payload) { + return fmt.Errorf("payload must be a JSON object") + } + + return nil +} + +func validateMetadata(metadata Metadata) error { + if !ValidID(metadata.MessageID) { return fmt.Errorf("message ID must be a canonical ULID") } - if envelope.CorrelationID != "" && !ValidID(envelope.CorrelationID) { + if metadata.CorrelationID != "" && !ValidID(metadata.CorrelationID) { return fmt.Errorf("correlation ID must be a canonical ULID") } - if envelope.CausationID != "" && !ValidID(envelope.CausationID) { + if metadata.CausationID != "" && !ValidID(metadata.CausationID) { return fmt.Errorf("causation ID must be a canonical ULID") } - for _, timestamp := range []*time.Time{envelope.SentAt, envelope.ExpiresAt} { + for _, timestamp := range []*time.Time{metadata.SentAt, metadata.ExpiresAt} { if timestamp != nil { _, offset := timestamp.Zone() if offset != 0 { @@ -98,10 +111,6 @@ func validateEnvelope(envelope wireEnvelope) error { } } - if !isObject(envelope.Payload) { - return fmt.Errorf("payload must be a JSON object") - } - return nil }