Add per-command metadata propagation to interfaces

docs.openc3.com/docs/guides/scripting-api.md

@@ -840,6 +840,19 @@ cmd("<Target Name>", "<Command Name>", "Param #1 Name" => <Param #1 Value>, "Par
 | timeout        | Optional named parameter to change the default timeout value of 5 seconds                            |
 | log_message    | Optional named parameter to prevent logging of the command                                           |
 | validate       | Optional named parameter to enable/disable validation (default is True)                              |
+| extra          | Optional metadata Hash/dict carried with the command packet to the interface                         |
+
+The `extra` keys `cmd_string`, `username`, `interface_name`, `queue_username`, `approver`,
+`cmd_success`, and `cmd_reason` are reserved for COSMOS audit data. Caller-supplied values for
+these keys are discarded or replaced by authoritative values as the command is processed.
+
+Keys that a packet's accessor writes into `extra` are also reserved. HTTP targets are the
+common case: `HTTP_PATH`, `HTTP_METHOD`, `HTTP_STATUS`, `HTTP_PACKET`, `HTTP_ERROR_PACKET`,
+`HTTP_HEADERS` and `HTTP_QUERIES` come from the command definition and always win over
+caller-supplied values, so `extra` cannot redirect the request an interface makes.
+
+`extra` is written to the command log, the command topics and the queue exactly as given.
+It is not obfuscated, so don't put secrets in it.
 
 <Tabs groupId="script-language">
 <TabItem value="python" label="Python Example">
@@ -848,6 +861,7 @@ cmd("<Target Name>", "<Command Name>", "Param #1 Name" => <Param #1 Value>, "Par
 cmd("INST COLLECT with DURATION 10, TYPE NORMAL")
 cmd("INST", "COLLECT", { "DURATION": 10, "TYPE": "NORMAL" })
 cmd("INST ABORT", timeout=10, log_message=False, validate=False)
+cmd("INST ABORT", extra={ "request_id": "1234-5678" })
 ```
 
 </TabItem>
@@ -860,6 +874,7 @@ cmd("INST COLLECT with DURATION 10, TYPE NORMAL")
 cmd("INST", "COLLECT", "DURATION" => 10, "TYPE" => "NORMAL")
 cmd("INST", "COLLECT", { "DURATION" => 10, "TYPE" => "NORMAL" })
 cmd("INST ABORT", timeout: 10, log_message: false, validate: false)
+cmd("INST ABORT", extra: { "request_id" => "1234-5678" })
 ```
 
 </TabItem>

openc3-cosmos-cmd-tlm-api/app/controllers/queues_controller.rb

@@ -164,15 +164,18 @@ def insert_command
         end
       end
       id = params[:id]&.to_f
+      extra = extra_param()
       # If id is nil this means insert at the end
       if target_name && cmd_name
         model.insert_command(id: id, username: username(), target_name: target_name, cmd_name: cmd_name,
-                             cmd_params: params[:cmd_params], validate: params[:validate], timeout: params[:timeout])
+                             cmd_params: params[:cmd_params], extra: extra, validate: params[:validate], timeout: params[:timeout])
       else
-        model.insert_command(id: id, username: username(), command: command,
+        model.insert_command(id: id, username: username(), command: command, extra: extra,
                              validate: params[:validate], timeout: params[:timeout])
       end
       render json: { status: 'success', message: 'Command added to queue' }
+    rescue ArgumentError => e
+      render json: { status: 'error', message: e.message, type: e.class.to_s }, status: :bad_request
     rescue StandardError => e
       log_error(e)
       render json: { status: 'error', message: e.message, type: e.class.to_s, backtrace: e.backtrace }, status: :internal_server_error
@@ -237,13 +240,16 @@ def update_command
       # validate should be true or false, default to true if not given
       validate = params[:validate].nil? ? true : params[:validate]
       # timeout can be nil which means use system default timeout
+      extra = extra_param()
       if target_name && cmd_name
         model.update_command(id: id, username: username(), target_name: target_name, cmd_name: cmd_name,
-                             cmd_params: params[:cmd_params], validate: validate, timeout: params[:timeout])
+                             cmd_params: params[:cmd_params], extra: extra, validate: validate, timeout: params[:timeout])
       else
-        model.update_command(id: id, username: username(), command: command, validate: validate, timeout: params[:timeout])
+        model.update_command(id: id, username: username(), command: command, extra: extra, validate: validate, timeout: params[:timeout])
       end
       render json: { status: 'success', message: 'Command updated' }
+    rescue ArgumentError => e
+      render json: { status: 'error', message: e.message, type: e.class.to_s }, status: :bad_request
     rescue OpenC3::QueueError => e
       log_error(e)
       render json: { status: 'error', message: e.message, type: e.class.to_s }, status: :bad_request
@@ -270,6 +276,8 @@ def exec_command
           # Support both new format (target_name, cmd_name, cmd_params) and legacy format (value)
           validate = command_data.key?('validate') ? command_data['validate'] : true
           timeout = command_data['timeout'] # Default is nil which means use system default timeout
+          extra = command_data['extra']
+          extra = JSON.parse(extra, allow_nan: true, create_additions: true) if extra.is_a?(String)
           if command_data['target_name'] && command_data['cmd_name']
             # New format: use 3-parameter cmd() method
             if command_data['cmd_params']
@@ -278,16 +286,16 @@ def exec_command
               cmd_params = {}
             end
             if hazardous
-              cmd_no_hazardous_check(command_data['target_name'], command_data['cmd_name'], cmd_params, queue: false, validate: validate, timeout: timeout, queue_username: command_data['username'], scope: params[:scope], token: token)
+              cmd_no_hazardous_check(command_data['target_name'], command_data['cmd_name'], cmd_params, queue: false, validate: validate, timeout: timeout, queue_username: command_data['username'], extra: extra, scope: params[:scope], token: token)
             else
-              cmd(command_data['target_name'], command_data['cmd_name'], cmd_params, queue: false, validate: validate, timeout: timeout, queue_username: command_data['username'], scope: params[:scope], token: token)
+              cmd(command_data['target_name'], command_data['cmd_name'], cmd_params, queue: false, validate: validate, timeout: timeout, queue_username: command_data['username'], extra: extra, scope: params[:scope], token: token)
             end
           elsif command_data['value']
             # Legacy format: use single string parameter
             if hazardous
-              cmd_no_hazardous_check(command_data['value'], queue: false, validate: validate, timeout: timeout, queue_username: command_data['username'], scope: params[:scope], token: token)
+              cmd_no_hazardous_check(command_data['value'], queue: false, validate: validate, timeout: timeout, queue_username: command_data['username'], extra: extra, scope: params[:scope], token: token)
             else
-              cmd(command_data['value'], queue: false, validate: validate, timeout: timeout, queue_username: command_data['username'], scope: params[:scope], token: token)
+              cmd(command_data['value'], queue: false, validate: validate, timeout: timeout, queue_username: command_data['username'], extra: extra, scope: params[:scope], token: token)
             end
           else
             log_error("Invalid command format in queue: #{command_data}")
@@ -330,6 +338,19 @@ def destroy
 
   private
 
+  # Caller metadata attached to a queued command. Rails wraps nested params in
+  # ActionController::Parameters so convert back to a plain Hash before it is
+  # JSON encoded into the queue entry.
+  def extra_param
+    extra = params[:extra]
+    return nil if extra.nil?
+
+    extra = extra.to_unsafe_h if extra.respond_to?(:to_unsafe_h)
+    raise ArgumentError, "Invalid extra parameter: #{extra}. Must be a Hash." unless extra.is_a?(Hash)
+
+    extra
+  end
+
   def change_state(params, state)
     return unless authorization('cmd')
     begin

openc3-cosmos-cmd-tlm-api/spec/controllers/queues_controller_spec.rb

@@ -281,6 +281,7 @@ def generate_queue_hash
         id: nil,
         username: "anonymous",
         command: "TEST COMMAND",
+        extra: nil,
         validate: nil,
         timeout: nil
       )
@@ -301,11 +302,42 @@ def generate_queue_hash
         id: id.to_f,
         username: "anonymous",
         command: "TEST COMMAND",
+        extra: nil,
         validate: nil,
         timeout: nil
       )
     end
 
+    it "passes extra metadata through to the model" do
+      queue_model = double("QueueModel")
+      allow(OpenC3::QueueModel).to receive(:get_model).and_return(queue_model)
+      allow(queue_model).to receive(:insert_command)
+
+      post :insert_command, params: {name: "QUEUE1", command: "TEST COMMAND", extra: {flow_uuid: "1234-5678"}, scope: "DEFAULT"}
+      expect(response).to have_http_status(:ok)
+      expect(queue_model).to have_received(:insert_command).with(
+        id: nil,
+        username: "anonymous",
+        command: "TEST COMMAND",
+        extra: {"flow_uuid" => "1234-5678"},
+        validate: nil,
+        timeout: nil
+      )
+    end
+
+    it "returns 400 when extra is not a Hash" do
+      queue_model = double("QueueModel")
+      allow(OpenC3::QueueModel).to receive(:get_model).and_return(queue_model)
+      allow(queue_model).to receive(:insert_command)
+
+      post :insert_command, params: {name: "QUEUE1", command: "TEST COMMAND", extra: "invalid", scope: "DEFAULT"}
+      expect(response).to have_http_status(400)
+      json = JSON.parse(response.body, allow_nan: true, create_additions: true)
+      expect(json["status"]).to eql("error")
+      expect(json["message"]).to eql("Invalid extra parameter: invalid. Must be a Hash.")
+      expect(queue_model).to_not have_received(:insert_command)
+    end
+
     it "returns 404 when the queue is not found" do
       allow(OpenC3::QueueModel).to receive(:get_model).and_return(nil)
 
@@ -440,11 +472,43 @@ def generate_queue_hash
         id: id,
         username: "anonymous",
         command: "UPDATED COMMAND",
+        extra: nil,
         validate: true, # Even when not provided, validate should default to true
         timeout: nil
       )
     end
 
+    it "passes extra metadata through to the model" do
+      queue_model = double("QueueModel")
+      allow(OpenC3::QueueModel).to receive(:get_model).and_return(queue_model)
+      allow(queue_model).to receive(:update_command)
+
+      post :update_command, params: {name: "QUEUE1", command: "UPDATED COMMAND", id: "1.0",
+                                     extra: {flow_uuid: "1234-5678"}, scope: "DEFAULT"}
+      expect(response).to have_http_status(:ok)
+      expect(queue_model).to have_received(:update_command).with(
+        id: "1.0",
+        username: "anonymous",
+        command: "UPDATED COMMAND",
+        extra: {"flow_uuid" => "1234-5678"},
+        validate: true,
+        timeout: nil
+      )
+    end
+
+    it "returns 400 when extra is not a Hash" do
+      queue_model = double("QueueModel")
+      allow(OpenC3::QueueModel).to receive(:get_model).and_return(queue_model)
+      allow(queue_model).to receive(:update_command)
+
+      post :update_command, params: {name: "QUEUE1", command: "UPDATED COMMAND", id: "1.0", extra: "invalid", scope: "DEFAULT"}
+      expect(response).to have_http_status(400)
+      json = JSON.parse(response.body, allow_nan: true, create_additions: true)
+      expect(json["status"]).to eql("error")
+      expect(json["message"]).to eql("Invalid extra parameter: invalid. Must be a Hash.")
+      expect(queue_model).to_not have_received(:update_command)
+    end
+
     it "returns 404 when the queue is not found" do
       allow(OpenC3::QueueModel).to receive(:get_model).and_return(nil)
 
@@ -550,6 +614,7 @@ def generate_queue_hash
       command_data = {
         "username" => "user1",
         "value" => "TEST COMMAND",
+        "extra" => JSON.generate({ "flow_uuid" => "1234-5678", "data" => "\xFF".b }.as_json, allow_nan: true),
         "timestamp" => 1000,
         "id" => 1.0
       }
@@ -561,6 +626,7 @@ def generate_queue_hash
           validate: true,
           timeout: nil,
           queue_username: "user1",
+          extra: { "flow_uuid" => "1234-5678", "data" => "\xFF".b },
           scope: "DEFAULT",
           token: anything
         }
@@ -591,6 +657,7 @@ def generate_queue_hash
           validate: true,
           timeout: nil,
           queue_username: "user2",
+          extra: nil,
           scope: "DEFAULT",
           token: anything
         }
@@ -658,6 +725,7 @@ def generate_queue_hash
           validate: true,
           timeout: nil,
           queue_username: "user3",
+          extra: nil,
           scope: "DEFAULT",
           token: anything
         }
@@ -687,6 +755,7 @@ def generate_queue_hash
           validate: true,
           timeout: 0,
           queue_username: "user4",
+          extra: nil,
           scope: "DEFAULT",
           token: anything
         }
@@ -779,4 +848,4 @@ def generate_queue_hash
       )
     end
   end
-end
+end

openc3/lib/openc3/api/cmd_api.rb

@@ -450,7 +450,7 @@ def _extract_target_command_parameter_names(method_name, *args)
 
     # NOTE: When adding new keywords to this method, make sure to update script/commands.rb
     def _cmd_implementation(method_name, *args, range_check:, hazardous_check:, raw:, timeout: nil, log_message: nil, manual: false, validate: true, queue: nil,
-                            queue_username: nil, scope: $openc3_scope, token: $openc3_token, **kwargs)
+                            queue_username: nil, extra: nil, scope: $openc3_scope, token: $openc3_token, **kwargs)
       extract_string_kwargs_to_args(args, kwargs)
       unless [nil, true, false].include?(log_message)
         raise "Invalid log_message parameter: #{log_message}. Must be true or false."
@@ -462,6 +462,9 @@ def _cmd_implementation(method_name, *args, range_check:, hazardous_check:, raw:
           raise "Invalid timeout parameter: #{timeout}. Must be numeric."
         end
       end
+      unless extra.nil? || extra.is_a?(Hash)
+        raise "Invalid extra parameter: #{extra}. Must be a Hash."
+      end
 
       case args.length
       when 1
@@ -535,6 +538,7 @@ def _cmd_implementation(method_name, *args, range_check:, hazardous_check:, raw:
         'log_message' => log_message.to_s,
         'obfuscated_items' => packet['obfuscated_items'].to_s
       }
+      command['extra'] = extra unless extra.nil?
       # Record the original queuing user (author) separately from 'username' (the
       # user or process that actually executed the command). Command History shows
       # 'username' as "Executed By" and queue_username as "Queued By".
@@ -550,6 +554,7 @@ def _cmd_implementation(method_name, *args, range_check:, hazardous_check:, raw:
           target_name: target_name,
           cmd_name: cmd_name,
           cmd_params: cmd_params,
+          extra: extra,
           validate: validate,
           timeout: timeout,
           username: username,

openc3/lib/openc3/interfaces/protocols/preidentified_protocol.rb

@@ -65,7 +65,9 @@ def write_packet(packet)
       @write_extra = nil
       if packet.extra
         @write_flags |= COSMOS4_EXTRA_FLAG_MASK
-        @write_extra = packet.extra.as_json().to_json(allow_nan: true)
+        # Force binary so appending it can't promote data_to_send to UTF-8, which
+        # then raises Encoding::CompatibilityError on the binary fields that follow
+        @write_extra = packet.extra.as_json().to_json(allow_nan: true).b
       end
       return packet
     end
@@ -76,7 +78,9 @@ def write_data(data, extra = nil)
       data_to_send << @sync_pattern if @sync_pattern
       data_to_send << @write_flags
       if @write_extra
-        data_to_send << [@write_extra.length].pack('N')
+        # Length field must count bytes, not characters, or a non-ASCII value in
+        # extra desyncs the receiver
+        data_to_send << [@write_extra.bytesize].pack('N')
         data_to_send << @write_extra
       end
       data_to_send << @write_time_seconds

openc3/lib/openc3/logs/packet_log_writer.rb

@@ -313,15 +313,20 @@ def write_entry(entry_type, cmd_or_tlm, target_name, packet_name, time_nsec_sinc
           if @data_format == :CBOR
             extra_encoded = extra.as_json.to_cbor
           else
-            extra_encoded = JSON.generate(extra.as_json, allow_nan: true)
+            # Force binary: JSON.generate returns UTF-8 and appending that to the
+            # binary entry raises Encoding::CompatibilityError once the entry holds
+            # a byte >= 0x80, which silently drops the packet from the log
+            extra_encoded = JSON.generate(extra.as_json, allow_nan: true).b
           end
-          length += extra_encoded.length
+          # Count bytes, not characters, so a non-ASCII value in extra can't produce
+          # a short entry length
+          length += extra_encoded.bytesize
         end
         length += OPENC3_PACKET_SECONDARY_FIXED_SIZE + data.length
         @entry.clear
         @entry << [length, flags, packet_index, time_nsec_since_epoch].pack(OPENC3_PACKET_PACK_DIRECTIVE)
         @entry << [received_time_nsec_since_epoch].pack(OPENC3_RECEIVED_TIME_PACK_DIRECTIVE) if received_time_nsec_since_epoch
-        @entry << [extra_encoded.length].pack(OPENC3_EXTRA_LENGTH_PACK_DIRECTIVE) << extra_encoded if extra_encoded
+        @entry << [extra_encoded.bytesize].pack(OPENC3_EXTRA_LENGTH_PACK_DIRECTIVE) << extra_encoded if extra_encoded
         @entry << data.force_encoding('ASCII-8BIT')
         @first_time = time_nsec_since_epoch if !@first_time or time_nsec_since_epoch < @first_time
         @last_time = time_nsec_since_epoch if !@last_time or time_nsec_since_epoch > @last_time

openc3/lib/openc3/microservices/interface_microservice.rb

@@ -284,6 +284,21 @@
             end
 
             command.extra ||= {}
+            if msg_hash['extra']
+              # Caller metadata is the base layer. Values already in command.extra were
+              # written by the packet's accessor during build_cmd (e.g. HttpAccessor sets
+              # HTTP_PATH, HTTP_METHOD, HTTP_HEADERS, HTTP_QUERIES from the command
+              # definition) and are authoritative, so they overwrite caller values rather
+              # than the other way around. Otherwise a caller could redirect the request
+              # an interface makes on their behalf.
+              caller_extra = JSON.parse(msg_hash['extra'], allow_nan: true, create_additions: true)
+              command.extra = caller_extra.merge(command.extra)
+            end
+            # These fields are populated from COSMOS workflow state below.
+            command.extra.delete('queue_username')
+            command.extra.delete('approver')
+            command.extra.delete('cmd_success')
+            command.extra.delete('cmd_reason')
             command.extra['cmd_string'] = msg_hash['cmd_string']
             command.extra['username'] = msg_hash['username']
             command.extra['interface_name'] = @interface.name
@@ -867,7 +882,7 @@
       # Skip reconnect if stop() has been called to avoid re-creating the status model
       if allow_reconnect and @interface.auto_reconnect and @interface.state != 'DISCONNECTED' and !@cancel_thread
         attempting()
         if !@cancel_thread
           # @logger.debug "reconnect delay: #{@interface.reconnect_delay}"
           @interface_thread_sleeper.sleep(@interface.reconnect_delay)
         end