diff --git a/docs.openc3.com/docs/guides/scripting-api.md b/docs.openc3.com/docs/guides/scripting-api.md index 66aa19839d..75092e39fd 100644 --- a/docs.openc3.com/docs/guides/scripting-api.md +++ b/docs.openc3.com/docs/guides/scripting-api.md @@ -846,6 +846,14 @@ cmd("", "", "Param #1 Name" => , "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 interface receives `extra` as `packet.extra['metadata']`. It is kept under its own key so it +can't change the values COSMOS adds to `packet.extra`, such as `cmd_string`, `username` and +`approver`, or the `HTTP_*` values an accessor sets from the command definition. + +`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. @@ -854,6 +862,7 @@ cmd("", "", "Param #1 Name" => , "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" }) ``` @@ -866,6 +875,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" }) ``` diff --git a/openc3-cosmos-cmd-tlm-api/app/controllers/queues_controller.rb b/openc3-cosmos-cmd-tlm-api/app/controllers/queues_controller.rb index 557283e824..d43a959638 100644 --- a/openc3-cosmos-cmd-tlm-api/app/controllers/queues_controller.rb +++ b/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 diff --git a/openc3-cosmos-cmd-tlm-api/spec/controllers/queues_controller_spec.rb b/openc3-cosmos-cmd-tlm-api/spec/controllers/queues_controller_spec.rb index 4a1b7bd1d5..10dacc5dc8 100644 --- a/openc3-cosmos-cmd-tlm-api/spec/controllers/queues_controller_spec.rb +++ b/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 \ No newline at end of file +end diff --git a/openc3/lib/openc3/api/cmd_api.rb b/openc3/lib/openc3/api/cmd_api.rb index 9517c03c01..c95a52d5dc 100644 --- a/openc3/lib/openc3/api/cmd_api.rb +++ b/openc3/lib/openc3/api/cmd_api.rb @@ -447,7 +447,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." @@ -459,6 +459,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 @@ -532,6 +535,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". @@ -547,6 +551,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, diff --git a/openc3/lib/openc3/interfaces/protocols/preidentified_protocol.rb b/openc3/lib/openc3/interfaces/protocols/preidentified_protocol.rb index 2ee0b779a4..9d03bdc238 100644 --- a/openc3/lib/openc3/interfaces/protocols/preidentified_protocol.rb +++ b/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 diff --git a/openc3/lib/openc3/logs/packet_log_writer.rb b/openc3/lib/openc3/logs/packet_log_writer.rb index fc1e6a0b56..d32d9f84fa 100644 --- a/openc3/lib/openc3/logs/packet_log_writer.rb +++ b/openc3/lib/openc3/logs/packet_log_writer.rb @@ -318,15 +318,20 @@ def write_entry(entry_type, cmd_or_tlm, target_name, packet_name, time_nsec_sinc flags |= OPENC3_CBOR_FLAG_MASK 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 diff --git a/openc3/lib/openc3/microservices/interface_microservice.rb b/openc3/lib/openc3/microservices/interface_microservice.rb index 99e9f082c3..b161e99862 100644 --- a/openc3/lib/openc3/microservices/interface_microservice.rb +++ b/openc3/lib/openc3/microservices/interface_microservice.rb @@ -305,6 +305,14 @@ def run end command.extra ||= {} + if msg_hash['extra'] + # Caller metadata lives under its own key so it can't collide with keys + # COSMOS sets (cmd_string, username, approver, ...) or that the packet's + # accessor sets during build_cmd (e.g. HttpAccessor's HTTP_PATH, + # HTTP_METHOD, HTTP_HEADERS, HTTP_QUERIES which HttpClientInterface + # uses to build the outgoing request) + command.extra['metadata'] = JSON.parse(msg_hash['extra'], allow_nan: true, create_additions: true) + end command.extra['cmd_string'] = msg_hash['cmd_string'] command.extra['username'] = msg_hash['username'] command.extra['interface_name'] = @interface.name diff --git a/openc3/lib/openc3/microservices/queue_microservice.rb b/openc3/lib/openc3/microservices/queue_microservice.rb index a20066ec1d..c4b19594a7 100644 --- a/openc3/lib/openc3/microservices/queue_microservice.rb +++ b/openc3/lib/openc3/microservices/queue_microservice.rb @@ -61,6 +61,10 @@ def process_queued_commands _queue_name, command_data, _timestamp = Store.bzpopmin("#{@scope}:#{@name}", timeout: 0.2) if command_data command = JSON.parse(command_data) + extra = command['extra'] + if extra.is_a?(String) + extra = JSON.parse(extra, allow_nan: true, create_additions: true) + end # It's important to set queue: false here to avoid infinite recursion when # OPENC3_DEFAULT_QUEUE is set because commands would be re-queued to the default queue # NOTE: cmd() via script rescues hazardous errors and calls prompt_for_hazardous() @@ -78,12 +82,12 @@ def process_queued_commands timeout = command['timeout'] # Pass queue_username so Command History is attributed to the original # author rather than the queue microservice name - cmd(command['target_name'], command['cmd_name'], cmd_params, queue: false, validate: validate, timeout: timeout, queue_username: command['username'], scope: @scope) + cmd(command['target_name'], command['cmd_name'], cmd_params, queue: false, validate: validate, timeout: timeout, queue_username: command['username'], extra: extra, scope: @scope) elsif command['value'] # Legacy format: use single string parameter for backwards compatibility validate = command.key?('validate') ? command['validate'] : true timeout = command['timeout'] - cmd(command['value'], queue: false, validate: validate, timeout: timeout, queue_username: command['username'], scope: @scope) + cmd(command['value'], queue: false, validate: validate, timeout: timeout, queue_username: command['username'], extra: extra, scope: @scope) else @logger.error "QueueProcessor: Invalid command format, missing required fields" end diff --git a/openc3/lib/openc3/models/queue_model.rb b/openc3/lib/openc3/models/queue_model.rb index aa71747ebf..cdd1128dc0 100644 --- a/openc3/lib/openc3/models/queue_model.rb +++ b/openc3/lib/openc3/models/queue_model.rb @@ -40,7 +40,7 @@ def self.all(scope:) end # END NOTE - def self.queue_command(name, command: nil, target_name: nil, cmd_name: nil, cmd_params: nil, validate: true, timeout: nil, username:, scope:) + def self.queue_command(name, command: nil, target_name: nil, cmd_name: nil, cmd_params: nil, extra: nil, validate: true, timeout: nil, username:, scope:) model = get_model(name: name, scope: scope) raise QueueError, "Queue '#{name}' not found in scope '#{scope}'" unless model @@ -53,7 +53,7 @@ def self.queue_command(name, command: nil, target_name: nil, cmd_name: nil, cmd_ id = result.empty? ? 1.0 : result[0][1].to_f + 1 command_data = build_command_data(username: username, command: command, target_name: target_name, - cmd_name: cmd_name, cmd_params: cmd_params, validate: validate, timeout: timeout) + cmd_name: cmd_name, cmd_params: cmd_params, extra: extra, validate: validate, timeout: timeout) Store.zadd("#{scope}:#{name}", id, command_data.to_json) model.notify(kind: 'command') end @@ -61,7 +61,7 @@ def self.queue_command(name, command: nil, target_name: nil, cmd_name: nil, cmd_ # Build the hash that gets serialized into Redis. Always uses symbol keys so # downstream code in this class can access values consistently. cmd_params is # JSON-encoded as a string so binary data survives the round-trip via as_json. - def self.build_command_data(username:, command: nil, target_name: nil, cmd_name: nil, cmd_params: nil, validate: nil, timeout: nil) + def self.build_command_data(username:, command: nil, target_name: nil, cmd_name: nil, cmd_params: nil, extra: nil, validate: nil, timeout: nil) command_data = { username: username, timestamp: Time.now.to_nsec_from_epoch } if target_name && cmd_name command_data[:target_name] = target_name @@ -74,6 +74,10 @@ def self.build_command_data(username:, command: nil, target_name: nil, cmd_name: end command_data[:validate] = validate unless validate.nil? command_data[:timeout] = timeout unless timeout.nil? + unless extra.nil? + # extra is already a JSON string when carried forward from an existing queue entry + command_data[:extra] = extra.is_a?(String) ? extra : JSON.generate(extra.as_json, allow_nan: true) + end command_data end @@ -118,7 +122,7 @@ def notify(kind:) QueueTopic.write_notification(notification, scope: @scope) end - def insert_command(id: nil, username:, command: nil, target_name: nil, cmd_name: nil, cmd_params: nil, validate: nil, timeout: nil) + def insert_command(id: nil, username:, command: nil, target_name: nil, cmd_name: nil, cmd_params: nil, extra: nil, validate: nil, timeout: nil) if @state == 'DISABLE' command_name = command || "#{target_name} #{cmd_name}" raise QueueError, "Queue '#{@name}' is disabled. Command '#{command_name}' not queued." @@ -130,12 +134,12 @@ def insert_command(id: nil, username:, command: nil, target_name: nil, cmd_name: end command_data = self.class.build_command_data(username: username, command: command, target_name: target_name, - cmd_name: cmd_name, cmd_params: cmd_params, validate: validate, timeout: timeout) + cmd_name: cmd_name, cmd_params: cmd_params, extra: extra, validate: validate, timeout: timeout) Store.zadd("#{@scope}:#{@name}", id, command_data.to_json) notify(kind: 'command') end - def update_command(id:, username:, command: nil, target_name: nil, cmd_name: nil, cmd_params: nil, validate: nil, timeout: nil) + def update_command(id:, username:, command: nil, target_name: nil, cmd_name: nil, cmd_params: nil, extra: nil, validate: nil, timeout: nil) if @state == 'DISABLE' raise QueueError, "Queue '#{@name}' is disabled. Command at id #{id} not updated." end @@ -145,9 +149,13 @@ def update_command(id:, username:, command: nil, target_name: nil, cmd_name: nil raise QueueError, "No command found at id #{id} in queue '#{@name}'" end + # Carry forward the caller metadata attached by cmd(extra: ...) when the update + # doesn't supply its own. Editing a queued command shouldn't silently drop it. + extra = JSON.parse(existing[0])['extra'] if extra.nil? + Store.zremrangebyscore("#{@scope}:#{@name}", id, id) command_data = self.class.build_command_data(username: username, command: command, target_name: target_name, - cmd_name: cmd_name, cmd_params: cmd_params, validate: validate, timeout: timeout) + cmd_name: cmd_name, cmd_params: cmd_params, extra: extra, validate: validate, timeout: timeout) Store.zadd("#{@scope}:#{@name}", id, command_data.to_json) notify(kind: 'command') end @@ -243,4 +251,4 @@ def destroy super() end end -end \ No newline at end of file +end diff --git a/openc3/lib/openc3/script/commands.rb b/openc3/lib/openc3/script/commands.rb index 258a2c3ee1..2e24b906a9 100644 --- a/openc3/lib/openc3/script/commands.rb +++ b/openc3/lib/openc3/script/commands.rb @@ -117,17 +117,19 @@ def _cmd_disconnect(cmd, raw, no_range, no_hazardous, *args, scope: $openc3_scop # except for range_check, hazardous_check, and raw as they are part of the cmd name # manual is always false since this is called from script and that is the default # NOTE: This is a helper method and should not be called directly - def _cmd(cmd, cmd_no_hazardous, *args, timeout: nil, log_message: nil, validate: true, queue: nil, scope: $openc3_scope, token: $openc3_token, **kwargs) + def _cmd(cmd, cmd_no_hazardous, *args, timeout: nil, log_message: nil, validate: true, queue: nil, extra: nil, scope: $openc3_scope, token: $openc3_token, **kwargs) extract_string_kwargs_to_args(args, kwargs) raw = cmd.include?('raw') no_range = cmd.include?('no_range') || cmd.include?('no_checks') no_hazardous = cmd.include?('no_hazardous') || cmd.include?('no_checks') + # Only send extra when given so commands still work against servers that predate it + extra_kwargs = extra.nil? ? {} : { extra: extra } if $disconnect _cmd_disconnect(cmd, raw, no_range, no_hazardous, *args, scope: scope) else begin begin - command = $api_server.method_missing(cmd, *args, timeout: timeout, log_message: log_message, validate: validate, queue: queue, scope: scope, token: token) + command = $api_server.method_missing(cmd, *args, timeout: timeout, log_message: log_message, validate: validate, queue: queue, **extra_kwargs, scope: scope, token: token) if log_message.nil? or log_message _log_cmd(command, raw, no_range, no_hazardous) end @@ -135,7 +137,7 @@ def _cmd(cmd, cmd_no_hazardous, *args, timeout: nil, log_message: nil, validate: # This opens a prompt at which point they can cancel and stop the script # or say Yes and send the command. Thus we don't care about the return value. prompt_for_hazardous(e.target_name, e.cmd_name, e.hazardous_description) - command = $api_server.method_missing(cmd_no_hazardous, *args, timeout: timeout, log_message: log_message, validate: validate, queue: queue, scope: scope, token: token) + command = $api_server.method_missing(cmd_no_hazardous, *args, timeout: timeout, log_message: log_message, validate: validate, queue: queue, **extra_kwargs, scope: scope, token: token) if log_message.nil? or log_message _log_cmd(command, raw, no_range, no_hazardous) end diff --git a/openc3/lib/openc3/topics/command_topic.rb b/openc3/lib/openc3/topics/command_topic.rb index 7270458f1c..a5f2bf53c0 100644 --- a/openc3/lib/openc3/topics/command_topic.rb +++ b/openc3/lib/openc3/topics/command_topic.rb @@ -46,6 +46,8 @@ def self.send_command(command, timeout: COMMAND_ACK_TIMEOUT_S, scope:, obfuscate # Save the existing cmd_params Hash and JSON generate before writing to the topic cmd_params = command['cmd_params'] command['cmd_params'] = JSON.generate(command['cmd_params'].as_json, allow_nan: true) + extra = command['extra'] + command['extra'] = JSON.generate(extra.as_json, allow_nan: true) if extra OpenC3.inject_context(command) db_shard = Store.db_shard_for_target(command['target_name'], scope: scope) @@ -54,6 +56,7 @@ def self.send_command(command, timeout: COMMAND_ACK_TIMEOUT_S, scope:, obfuscate if timeout <= 0 Topic.write_topic("{#{scope}__CMD}TARGET__#{command['target_name']}", command, '*', 100, db_shard: db_shard) command["cmd_params"] = cmd_params # Restore the original cmd_params Hash + command['extra'] = extra if extra return command end @@ -61,6 +64,7 @@ def self.send_command(command, timeout: COMMAND_ACK_TIMEOUT_S, scope:, obfuscate Topic.update_topic_offsets([ack_topic], db_shard: db_shard) cmd_id = Topic.write_topic("{#{scope}__CMD}TARGET__#{command['target_name']}", command, '*', 100, db_shard: db_shard) command["cmd_params"] = cmd_params # Restore the original cmd_params Hash + command['extra'] = extra if extra time = Time.now while (Time.now - time) < timeout Topic.read_topics([ack_topic], db_shard: db_shard) do |_topic, _msg_id, msg_hash, _redis| diff --git a/openc3/python/openc3/api/cmd_api.py b/openc3/python/openc3/api/cmd_api.py index 835a954590..d91d923216 100644 --- a/openc3/python/openc3/api/cmd_api.py +++ b/openc3/python/openc3/api/cmd_api.py @@ -611,6 +611,9 @@ def _cmd_implementation( manual=manual, ) queue_username = kwargs.get("queue_username") + extra = kwargs.get("extra") + if extra is not None and not isinstance(extra, dict): + raise RuntimeError(f"Invalid extra parameter: {extra}. Must be a dict.") if not user: user = {} user["username"] = os.environ.get("OPENC3_MICROSERVICE_NAME") @@ -690,6 +693,8 @@ def _cmd_implementation( "log_message": str(log_message), "obfuscated_items": json.dumps(packet.get("obfuscated_items", [])), } + if extra is not None: + command["extra"] = extra # 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". @@ -705,7 +710,7 @@ def _cmd_implementation( # Pull the command out of the script string, e.g. cmd("INST ABORT") queued = cmd_string.split('("')[1].split('")')[0] QueueModel.queue_command( - queue, command=queued, username=username, scope=scope, validate=validate, timeout=timeout + queue, command=queued, username=username, scope=scope, validate=validate, timeout=timeout, extra=extra ) else: CommandTopic.send_command(command, timeout=timeout, scope=scope) diff --git a/openc3/python/openc3/interfaces/protocols/preidentified_protocol.py b/openc3/python/openc3/interfaces/protocols/preidentified_protocol.py index b8e89fc894..9530bf49eb 100644 --- a/openc3/python/openc3/interfaces/protocols/preidentified_protocol.py +++ b/openc3/python/openc3/interfaces/protocols/preidentified_protocol.py @@ -17,6 +17,7 @@ from openc3.config.config_parser import ConfigParser from openc3.interfaces.protocols.burst_protocol import BurstProtocol from openc3.utilities.extract import convert_to_value +from openc3.utilities.json import JsonEncoder # Delineates packets using the OpenC3 preidentification system @@ -68,7 +69,9 @@ def write_packet(self, packet): self.write_extra = None if packet.extra: self.write_flags |= PreidentifiedProtocol.COSMOS4_EXTRA_FLAG_MASK - self.write_extra = json.dumps(packet.extra) + # JsonEncoder so binary and other COSMOS types survive, matching the Ruby + # side which encodes via as_json + self.write_extra = json.dumps(packet.extra, cls=JsonEncoder) return packet def write_data(self, data, extra=None): @@ -78,8 +81,11 @@ def write_data(self, data, extra=None): data_to_send += self.sync_pattern data_to_send += self.write_flags.to_bytes(1, byteorder="big") if self.write_extra: - data_to_send += struct.pack(">I", len(self.write_extra)) - data_to_send += bytes(self.write_extra, "ascii") + # Length field must count bytes, not characters, or a non-ASCII value in + # extra desyncs the receiver + write_extra = self.write_extra.encode("utf-8") + data_to_send += struct.pack(">I", len(write_extra)) + data_to_send += write_extra data_to_send += self.write_time_seconds data_to_send += self.write_time_microseconds data_to_send += struct.pack(">B", len(self.write_target_name)) diff --git a/openc3/python/openc3/microservices/interface_microservice.py b/openc3/python/openc3/microservices/interface_microservice.py index 26fb60a10f..45b6b9989e 100644 --- a/openc3/python/openc3/microservices/interface_microservice.py +++ b/openc3/python/openc3/microservices/interface_microservice.py @@ -365,6 +365,13 @@ def process_cmd(self, topic, msg_id, msg_hash, _redis): return str(e) command.extra = command.extra or {} + if msg_hash.get(b"extra"): + # Caller metadata lives under its own key so it can't collide with keys + # COSMOS sets (cmd_string, username, approver, ...) or that the packet's + # accessor sets during build_cmd (e.g. HttpAccessor's HTTP_PATH, + # HTTP_METHOD, HTTP_HEADERS, HTTP_QUERIES which HttpClientInterface + # uses to build the outgoing request) + command.extra["metadata"] = json.loads(msg_hash[b"extra"], cls=JsonDecoder) command.extra["cmd_string"] = msg_hash.get(b"cmd_string", b"").decode() command.extra["username"] = msg_hash.get(b"username", b"").decode() command.extra["interface_name"] = self.interface.name diff --git a/openc3/python/openc3/models/queue_model.py b/openc3/python/openc3/models/queue_model.py index 9d75134eca..36d69c9bca 100644 --- a/openc3/python/openc3/models/queue_model.py +++ b/openc3/python/openc3/models/queue_model.py @@ -15,6 +15,7 @@ from openc3.models.model import Model from openc3.topics.queue_topic import QueueTopic +from openc3.utilities.json import JsonEncoder from openc3.utilities.store import Store @@ -48,7 +49,14 @@ def all(cls, scope: str): # However we need a lot of methods to enable cls.get_model and model.notify @classmethod def queue_command( - cls, name: str, command: str, username: str, scope: str, validate: bool = True, timeout: float = None + cls, + name: str, + command: str, + username: str, + scope: str, + validate: bool = True, + timeout: float = None, + extra: dict | None = None, ): model = cls.get_model(name=name, scope=scope) if not model: @@ -68,6 +76,8 @@ def queue_command( "timeout": timeout, "timestamp": time.time_ns(), } + if extra is not None: + command_data["extra"] = json.dumps(extra, cls=JsonEncoder) Store.zadd(f"{scope}:{name}", {json.dumps(command_data): index}) model.notify(kind="command") else: diff --git a/openc3/python/openc3/script/commands.py b/openc3/python/openc3/script/commands.py index 59579688c9..bb90736feb 100644 --- a/openc3/python/openc3/script/commands.py +++ b/openc3/python/openc3/script/commands.py @@ -228,6 +228,7 @@ def _cmd( log_message=None, validate=True, queue=None, + extra=None, scope=OPENC3_SCOPE, ): """Send the command and log the results @@ -237,6 +238,8 @@ def _cmd( raw = "raw" in cmd no_range = "no_range" in cmd or "no_checks" in cmd no_hazardous = "no_hazardous" in cmd or "no_checks" in cmd + # Only send extra when given so commands still work against servers that predate it + extra_kwargs = {} if extra is None else {"extra": extra} if openc3.script.DISCONNECT: _cmd_disconnect(cmd, raw, no_range, no_hazardous, *args, scope=scope) @@ -249,6 +252,7 @@ def _cmd( log_message=log_message, validate=validate, queue=queue, + **extra_kwargs, scope=scope, ) if log_message is None or log_message: @@ -270,6 +274,7 @@ def _cmd( log_message=log_message, validate=validate, queue=queue, + **extra_kwargs, scope=scope, ) if log_message is None or log_message: diff --git a/openc3/python/openc3/topics/command_topic.py b/openc3/python/openc3/topics/command_topic.py index a8ba64dec3..ae06fa175b 100644 --- a/openc3/python/openc3/topics/command_topic.py +++ b/openc3/python/openc3/topics/command_topic.py @@ -36,7 +36,7 @@ def write_packet(cls, packet, scope): "buffer": bytes(packet.buffer_no_copy()), } if packet.extra: - msg_hash["extra"] = json.dumps(packet.extra) + msg_hash["extra"] = json.dumps(packet.extra, cls=JsonEncoder) db_shard = Store.db_shard_for_target(packet.target_name, scope=scope) minid = Topic.log_stream_safety_minid(packet.target_name, "CMD", scope) EphemeralStoreQueued.instance(db_shard=db_shard).write_topic(topic, msg_hash, minid=minid) @@ -58,6 +58,9 @@ def send_command(cls, command, timeout, scope, obfuscated_items=None): # Save the existing cmd_params Hash and JSON generate before writing to the topic cmd_params = command["cmd_params"] command["cmd_params"] = json.dumps(command["cmd_params"], cls=JsonEncoder) + extra = command.get("extra") + if extra is not None: + command["extra"] = json.dumps(extra, cls=JsonEncoder) db_shard = Store.db_shard_for_target(command["target_name"], scope=scope) @@ -71,6 +74,8 @@ def send_command(cls, command, timeout, scope, obfuscated_items=None): db_shard=db_shard, ) command["cmd_params"] = cmd_params # Restore the original cmd_params dict + if extra is not None: + command["extra"] = extra return command ack_topic = f"{{{scope}__ACKCMD}}TARGET__{command['target_name']}" @@ -83,6 +88,8 @@ def send_command(cls, command, timeout, scope, obfuscated_items=None): db_shard=db_shard, ) command["cmd_params"] = cmd_params # Restore the original cmd_params dict + if extra is not None: + command["extra"] = extra start_time = time.time() while (time.time() - start_time) < timeout: for _, _, msg_hash, _ in Topic.read_topics([ack_topic], db_shard=db_shard): diff --git a/openc3/python/test/api/test_cmd_api.py b/openc3/python/test/api/test_cmd_api.py index 67ea1a815d..e38f77a193 100644 --- a/openc3/python/test/api/test_cmd_api.py +++ b/openc3/python/test/api/test_cmd_api.py @@ -247,6 +247,25 @@ def test_cmd_records_the_original_queuing_user(self): # username is the executing user/process, not the author self.assertNotEqual(command["username"], "original_author") + def test_cmd_carries_extra_metadata(self): + extra = {"flow_uuid": "1234-5678", "hv_id": 42} + for name in [ + "cmd", + "cmd_no_range_check", + "cmd_no_hazardous_check", + "cmd_no_checks", + "cmd_raw", + "cmd_raw_no_range_check", + "cmd_raw_no_hazardous_check", + "cmd_raw_no_checks", + ]: + command = globals()[name]("INST", "ABORT", extra=extra) + self.assertEqual(command["extra"], extra) + + def test_cmd_rejects_invalid_extra_metadata(self): + with self.assertRaisesRegex(RuntimeError, "Invalid extra parameter: invalid. Must be a dict"): + cmd("INST", "ABORT", extra="invalid") + def test_cmd_warns_about_required_parameters(self): for name in [ "cmd", diff --git a/openc3/python/test/interfaces/protocols/test_preidentified_protocol.py b/openc3/python/test/interfaces/protocols/test_preidentified_protocol.py index 24b2fd83cc..0e9d5ce753 100644 --- a/openc3/python/test/interfaces/protocols/test_preidentified_protocol.py +++ b/openc3/python/test/interfaces/protocols/test_preidentified_protocol.py @@ -19,6 +19,7 @@ from openc3.interfaces.protocols.preidentified_protocol import PreidentifiedProtocol from openc3.interfaces.stream_interface import StreamInterface from openc3.streams.stream import Stream +from openc3.utilities.json import JsonDecoder, JsonEncoder from test.test_helper import * @@ -152,6 +153,25 @@ def test_write_creates_a_packet_header_with_extra(self): offset += len(json_extra) self.verify_time_tgt_pkt_buffer(offset, time, pkt) + def test_write_encodes_binary_extra(self): + # Binary values reach a command's extra via cmd(extra=...), so the protocol has + # to use JsonEncoder like the rest of COSMOS. Plain json.dumps raises + # "Object of type bytes is not JSON serializable" and kills the write. + _time, pkt = self.setup_stream_pkt() + pkt.stored = False + pkt.extra = {"data": b"\xff"} + self.interface.write(pkt) + + json_extra = json.dumps({"data": b"\xff"}, cls=JsonEncoder).encode("utf-8") + offset = 1 + self.assertEqual( + struct.unpack(">I", TestPreidentifiedProtocol.buffer[offset : (offset + 4)])[0], + len(json_extra), + ) + offset += 4 + written = TestPreidentifiedProtocol.buffer[offset : (offset + len(json_extra))] + self.assertEqual(json.loads(written, cls=JsonDecoder), {"data": b"\xff"}) + def test_write_creates_a_packet_header_with_stored_and_extra(self): time, pkt = self.setup_stream_pkt() pkt.stored = True diff --git a/openc3/python/test/microservices/test_interface_microservice.py b/openc3/python/test/microservices/test_interface_microservice.py index f6c842bbf0..d5f8004db2 100644 --- a/openc3/python/test/microservices/test_interface_microservice.py +++ b/openc3/python/test/microservices/test_interface_microservice.py @@ -32,6 +32,7 @@ from openc3.topics.interface_topic import InterfaceTopic from openc3.topics.telemetry_decom_topic import TelemetryDecomTopic from openc3.topics.topic import Topic +from openc3.utilities.json import JsonEncoder from openc3.utilities.store_queued import EphemeralStoreQueued, StoreQueued from openc3.utilities.time import from_nsec_from_epoch from test.test_helper import * @@ -755,6 +756,18 @@ def test_process_cmd_with_all_fields_and_missing_optional_fields(self): b"hazardous_check": b"TRUE", b"cmd_string": b"cmd('INST ABORT')", b"username": b"test_user", + b"extra": json.dumps( + { + "flow_uuid": "1234-5678", + "username": "untrusted", + "queue_username": "untrusted", + "approver": "untrusted", + "cmd_success": False, + "cmd_reason": "untrusted", + "data": b"\xff", + }, + cls=JsonEncoder, + ).encode(), b"queue_username": b"DEFAULT__MULTI__INST", b"validate": b"TRUE", b"manual": b"FALSE", @@ -768,6 +781,17 @@ def test_process_cmd_with_all_fields_and_missing_optional_fields(self): command = mock_write.call_args[0][0] self.assertEqual(command.extra["username"], "test_user") self.assertEqual(command.extra.get("queue_username"), "DEFAULT__MULTI__INST") + # Caller metadata is kept under its own key, so caller values can't + # replace the COSMOS audit fields + self.assertNotIn("flow_uuid", command.extra) + self.assertNotIn("approver", command.extra) + self.assertNotIn("cmd_success", command.extra) + self.assertNotIn("cmd_reason", command.extra) + metadata = command.extra["metadata"] + self.assertEqual(metadata["flow_uuid"], "1234-5678") + self.assertEqual(metadata["username"], "untrusted") + self.assertEqual(metadata["approver"], "untrusted") + self.assertEqual(metadata["data"], b"\xff") # Minimal msg_hash — only required fields; optional fields use .get() defaults minimal_msg_hash = { @@ -792,6 +816,92 @@ def test_process_cmd_with_all_fields_and_missing_optional_fields(self): result = handler.process_cmd(topic, msg_id, full_msg_hash, None) self.assertIsNone(result) + def test_process_cmd_decodes_ruby_encoded_special_floats_in_extra(self): + """A command queued from Ruby (or released by the Ruby queue microservice) has + its extra encoded by Ruby's Float#as_json, which writes non-finite floats as + {"json_class": "Float", "raw": "NaN"|"Infinity"|"-Infinity"}. Those must arrive + at a Python interface as floats, not as dicts.""" + im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") + thread = threading.Thread(target=im.run) + thread.start() + self.addCleanup(thread.join, 5) + self.addCleanup(im.shutdown) + time.sleep(0.1) + + handler = im.handler_thread + # Byte for byte what Ruby's JSON.generate(extra.as_json, allow_nan: true) emits + ruby_extra = ( + b'{"nan":{"json_class":"Float","raw":"NaN"},' + b'"inf":{"json_class":"Float","raw":"Infinity"},' + b'"ninf":{"json_class":"Float","raw":"-Infinity"},' + b'"normal":1.5}' + ) + msg_hash = { + b"target_name": b"INST", + b"cmd_name": b"ABORT", + b"cmd_params": json.dumps({}).encode(), + b"cmd_string": b"cmd('INST ABORT')", + b"username": b"test_user", + b"extra": ruby_extra, + } + with patch("openc3.microservices.interface_microservice.CommandDecomTopic.write_packet") as mock_write: + result = handler.process_cmd("{DEFAULT__CMD}TARGET__INST", "1-0", msg_hash, None) + self.assertEqual(result, "SUCCESS") + + command = mock_write.call_args[0][0] + self.assertTrue(math.isnan(command.extra["metadata"]["nan"])) + self.assertEqual(command.extra["metadata"]["inf"], float("inf")) + self.assertEqual(command.extra["metadata"]["ninf"], float("-inf")) + self.assertEqual(command.extra["metadata"]["normal"], 1.5) + + def test_process_cmd_keeps_caller_metadata_separate_from_accessor_extra(self): + """Accessors write into packet.extra while build_cmd sets the command + parameters. HttpAccessor puts HTTP_PATH / HTTP_METHOD / HTTP_HEADERS / + HTTP_QUERIES there and HttpClientInterface builds the outgoing request from + them, so caller metadata must not be able to set or change these.""" + im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") + thread = threading.Thread(target=im.run) + thread.start() + self.addCleanup(thread.join, 5) + self.addCleanup(im.shutdown) + time.sleep(0.1) + + handler = im.handler_thread + original_build_cmd = System.commands.build_cmd + caller_extra = { + "HTTP_PATH": "/supplied", + "HTTP_METHOD": "delete", + "HTTP_HEADERS": {"authorization": "supplied"}, + "flow_uuid": "1234-5678", + } + + def build_cmd_with_accessor_extra(*args, **kwargs): + command = original_build_cmd(*args, **kwargs) + command.extra = {"HTTP_PATH": "/defined", "HTTP_METHOD": "get"} + return command + + msg_hash = { + b"target_name": b"INST", + b"cmd_name": b"ABORT", + b"cmd_params": json.dumps({}).encode(), + b"cmd_string": b"cmd('INST ABORT')", + b"username": b"test_user", + b"extra": json.dumps(caller_extra).encode(), + } + with ( + patch.object(System.commands, "build_cmd", side_effect=build_cmd_with_accessor_extra), + patch("openc3.microservices.interface_microservice.CommandDecomTopic.write_packet") as mock_write, + ): + result = handler.process_cmd("{DEFAULT__CMD}TARGET__INST", "1-0", msg_hash, None) + self.assertEqual(result, "SUCCESS") + + command = mock_write.call_args[0][0] + self.assertEqual(command.extra["HTTP_PATH"], "/defined") + self.assertEqual(command.extra["HTTP_METHOD"], "get") + self.assertNotIn("HTTP_HEADERS", command.extra) + self.assertNotIn("flow_uuid", command.extra) + self.assertEqual(command.extra["metadata"], caller_extra) + def test_process_cmd_rejects_malformed_connect_params(self): im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") self.addCleanup(im.shutdown) diff --git a/openc3/python/test/models/test_queue_model.py b/openc3/python/test/models/test_queue_model.py index c5cda6125b..097e2714af 100644 --- a/openc3/python/test/models/test_queue_model.py +++ b/openc3/python/test/models/test_queue_model.py @@ -15,6 +15,7 @@ from openc3.models.queue_model import QueueError, QueueModel from openc3.topics.queue_topic import QueueTopic +from openc3.utilities.json import JsonDecoder from test.test_helper import mock_redis @@ -95,6 +96,29 @@ def test_queue_command_validate_false_timeout_zero(self, mock_get_model, mock_st ) mock_store.zadd.assert_called_once_with("DEFAULT:TEST", {expected_data: 1.0}) + @patch("openc3.models.queue_model.Store") + @patch("openc3.models.queue_model.QueueModel.get_model") + def test_queue_command_with_extra_metadata(self, mock_get_model, mock_store): + mock_model = Mock() + mock_model.state = "RUNNING" + mock_get_model.return_value = mock_model + mock_store.zrevrange.return_value = [] + + with patch("time.time_ns", return_value=1234567890): + QueueModel.queue_command( + "TEST", + command="CMD", + username="user", + scope="DEFAULT", + extra={"flow_uuid": "1234-5678", "data": b"\xff"}, + ) + + queued = json.loads(next(iter(mock_store.zadd.call_args.args[1]))) + self.assertEqual( + json.loads(queued["extra"], cls=JsonDecoder), + {"flow_uuid": "1234-5678", "data": b"\xff"}, + ) + @patch("openc3.models.queue_model.Store") @patch("openc3.models.queue_model.QueueModel.get_model") def test_queue_command_with_existing_items(self, mock_get_model, mock_store): diff --git a/openc3/python/test/script/test_commands.py b/openc3/python/test/script/test_commands.py index 13fa6b45eb..beb6204941 100644 --- a/openc3/python/test/script/test_commands.py +++ b/openc3/python/test/script/test_commands.py @@ -149,6 +149,18 @@ def test_sends_a_cmd(self): ) self.assertEqual(gArgs, ("INST ABORT",)) + def test_sends_extra_metadata(self): + extra = {"flow_uuid": "1234-5678", "hv_id": 42} + for _stdout in capture_io(): + cmd("INST ABORT", extra=extra) + self.assertEqual(gKwargs["extra"], extra) + + def test_does_not_send_extra_when_not_given(self): + # Older servers reject an unknown extra keyword, even when None + for _stdout in capture_io(): + cmd("INST ABORT") + self.assertNotIn("extra", gKwargs) + def test_sends_a_cmd_raw(self): global gArgs global gKwargs diff --git a/openc3/python/test/topics/test_command_topic.py b/openc3/python/test/topics/test_command_topic.py index 0c9637439e..96ee14bb1a 100644 --- a/openc3/python/test/topics/test_command_topic.py +++ b/openc3/python/test/topics/test_command_topic.py @@ -17,6 +17,7 @@ from openc3.system.system import System from openc3.topics.command_topic import CommandTopic +from openc3.utilities.json import JsonDecoder from test.test_helper import ack_next_message, mock_redis, setup_system, shard_redis @@ -83,6 +84,11 @@ def test_includes_extra_when_set(self): self.assertIn("extra", self.captured["msg_hash"]) self.assertEqual(json.loads(self.captured["msg_hash"]["extra"]), extra) + def test_encodes_binary_extra(self): + extra = {"data": b"\xff"} + CommandTopic.write_packet(self._make_packet(extra=extra), scope="DEFAULT") + self.assertEqual(json.loads(self.captured["msg_hash"]["extra"], cls=JsonDecoder), extra) + def test_omits_extra_when_none(self): CommandTopic.write_packet(self._make_packet(extra=None), scope="DEFAULT") self.assertNotIn("extra", self.captured["msg_hash"]) diff --git a/openc3/python/test/utilities/test_json.py b/openc3/python/test/utilities/test_json.py index 84e7265d54..1b131fe9b5 100644 --- a/openc3/python/test/utilities/test_json.py +++ b/openc3/python/test/utilities/test_json.py @@ -48,3 +48,16 @@ def test_decodes_ruby_float_additions(self): def test_leaves_unknown_float_raw_as_dict(self): string = '{"json_class": "Float", "raw": "bogus"}' self.assertEqual(json.loads(string, cls=JsonDecoder), {"json_class": "Float", "raw": "bogus"}) + + def test_decodes_python_special_floats(self): + # Python's json module writes bare NaN/Infinity literals and reads them back + # natively, so both forms have to decode to the same values + string = json.dumps({"nan": float("nan"), "inf": float("inf"), "ninf": float("-inf")}, cls=JsonEncoder) + decoded = json.loads(string, cls=JsonDecoder) + self.assertTrue(math.isnan(decoded["nan"])) + self.assertEqual(decoded["inf"], float("inf")) + self.assertEqual(decoded["ninf"], float("-inf")) + + def test_leaves_unknown_json_class_as_dict(self): + string = '{"json_class": "Object", "raw": [1, 2]}' + self.assertEqual(json.loads(string, cls=JsonDecoder), {"json_class": "Object", "raw": [1, 2]}) diff --git a/openc3/spec/api/cmd_api_spec.rb b/openc3/spec/api/cmd_api_spec.rb index c83d608e6a..f97ab63e1b 100644 --- a/openc3/spec/api/cmd_api_spec.rb +++ b/openc3/spec/api/cmd_api_spec.rb @@ -147,6 +147,18 @@ def test_cmd_unknown(method) expect(command['username']).to_not eql 'original_author' end + it "carries extra metadata" do + extra = { 'flow_uuid' => '1234-5678', 'hv_id' => 42 } + command = @api.send(method, "INST", "ABORT", extra: extra) + expect(command['extra']).to eql extra + end + + it "rejects invalid extra metadata" do + expect { @api.send(method, "INST", "ABORT", extra: 'invalid') }.to raise_error( + "Invalid extra parameter: invalid. Must be a Hash." + ) + end + it "warns about required parameters" do expect { @api.send(method, "INST COLLECT with DURATION 5") }.to raise_error(/Required/) end diff --git a/openc3/spec/interfaces/protocols/preidentified_protocol_spec.rb b/openc3/spec/interfaces/protocols/preidentified_protocol_spec.rb index cb8e0b6e0f..b427f3d466 100644 --- a/openc3/spec/interfaces/protocols/preidentified_protocol_spec.rb +++ b/openc3/spec/interfaces/protocols/preidentified_protocol_spec.rb @@ -172,6 +172,26 @@ def write(data); $buffer = data; end expect($buffer[offset..-1]).to eql pkt.buffer end + it "writes a byte accurate length for non-ASCII extra" do + @interface.instance_variable_set(:@stream, PreStream.new) + @interface.add_protocol(PreidentifiedProtocol, [nil, 5], :READ_WRITE) + pkt = System.telemetry.packet("SYSTEM", "META").clone + pkt.received_time = Time.new(2020, 1, 31, 12, 15, 30.5) + pkt.stored = false + # Multi-byte UTF-8 makes String#length differ from String#bytesize. The + # length field counts bytes, so a character count would desync the receiver. + extra_data = { "note" => "café" } + pkt.extra = extra_data + @interface.write(pkt) + + json_extra = extra_data.as_json().to_json(allow_nan: true) + expect(json_extra.bytesize).to_not eql json_extra.length + offset = 1 # flags + expect($buffer[offset..(offset + 3)].unpack('N')[0]).to eql json_extra.bytesize + offset += 4 + expect($buffer[offset...(offset + json_extra.bytesize)]).to eql json_extra.b + end + it "creates a packet header with stored and extra" do @interface.instance_variable_set(:@stream, PreStream.new) @interface.add_protocol(PreidentifiedProtocol, [nil, 5], :READ_WRITE) diff --git a/openc3/spec/logs/packet_log_writer_spec.rb b/openc3/spec/logs/packet_log_writer_spec.rb index a3ac92d927..497147cc9d 100644 --- a/openc3/spec/logs/packet_log_writer_spec.rb +++ b/openc3/spec/logs/packet_log_writer_spec.rb @@ -220,6 +220,41 @@ module OpenC3 FileUtils.rm_f 'test_log.bin' end + it "round trips non-ASCII extra" do + # JSON.generate returns UTF-8. Appending that to the binary log entry raises + # Encoding::CompatibilityError once the entry holds a byte >= 0x80 (the packet + # time nearly always does), which silently dropped the packet from the log. + # The extra length field also has to count bytes rather than characters. + now = Time.now.to_nsec_from_epoch + extra = { 'note' => 'café', 'flow_uuid' => '1234-5678' } + encoded = JSON.generate(extra.as_json, allow_nan: true) + expect(encoded.bytesize).to_not eq encoded.length + + plw = PacketLogWriter.new(@log_dir, 'test') + # JSON format so extra goes through JSON.generate. to_cbor already returns a + # binary string where length == bytesize. + plw.data_format = :JSON + plw.write(:RAW_PACKET, :CMD, 'TGT1', 'PKT1', now, false, "\x01\x02", nil, '0-0', extra: extra) + plw.write(:RAW_PACKET, :CMD, 'TGT2', 'PKT2', now, false, "\x03\x04", nil, '0-0') + threads = plw.shutdown + threads.each { |t| t.join } + + expect(@files.keys.length).to eq 1 + bin = Zlib::GzipReader.new(StringIO.new(@files.values.first)).read + File.open('test_log.bin', 'wb') { |file| file.write bin } + reader = PacketLogReader.new + reader.open('test_log.bin') + got = [] + while (pkt = reader.read) + got << [pkt.target_name, pkt.buffer] + end + # The packet carrying extra must be present, and the one after it must still + # be readable, which only happens if the entry length counted bytes + expect(got).to eql [['TGT1', "\x01\x02"], ['TGT2', "\x03\x04"]] + reader.close() + FileUtils.rm_f 'test_log.bin' + end + %i[CBOR JSON].each do |data_format| it "round trips RAW_PACKET extra using #{data_format}" do time = Time.now.to_nsec_from_epoch diff --git a/openc3/spec/microservices/interface_microservice_spec.rb b/openc3/spec/microservices/interface_microservice_spec.rb index 025cd43fa0..4578078c33 100644 --- a/openc3/spec/microservices/interface_microservice_spec.rb +++ b/openc3/spec/microservices/interface_microservice_spec.rb @@ -629,14 +629,67 @@ def raised(error) # queue_username is the original author (shown as "Queued By"), passed # by the queue microservice when a command is released from a queue - @api.cmd("INST", "ABORT", queue_username: "DEFAULT__MULTI__INST") + @api.cmd("INST", "ABORT", queue_username: "DEFAULT__MULTI__INST", + extra: { + 'flow_uuid' => '1234-5678', 'username' => 'untrusted', + 'queue_username' => 'untrusted', 'approver' => 'untrusted', + 'cmd_success' => false, 'cmd_reason' => 'untrusted' + }) sleep 0.01 im.shutdown expect(captured).to_not be_nil expect(captured.target_name).to eql("INST") expect(captured.packet_name).to eql("ABORT") + # Caller metadata is kept under its own key, so caller values can't + # replace the COSMOS audit fields + expect(captured.extra['metadata']).to eql({ + 'flow_uuid' => '1234-5678', 'username' => 'untrusted', + 'queue_username' => 'untrusted', 'approver' => 'untrusted', + 'cmd_success' => false, 'cmd_reason' => 'untrusted' + }) + expect(captured.extra['username']).to_not eql('untrusted') expect(captured.extra['queue_username']).to eql("DEFAULT__MULTI__INST") + expect(captured.extra).not_to have_key('flow_uuid') + expect(captured.extra).not_to have_key('approver') + expect(captured.extra).not_to have_key('cmd_success') + expect(captured.extra).not_to have_key('cmd_reason') + end + + it "keeps caller metadata separate from accessor populated extra" do + im = InterfaceMicroservice.new("DEFAULT__INTERFACE__INST_INT") + + # Accessors write into packet.extra while build_cmd sets the command + # parameters. HttpAccessor puts HTTP_PATH / HTTP_METHOD / HTTP_HEADERS / + # HTTP_QUERIES there and HttpClientInterface builds the outgoing request + # from them, so caller metadata must not be able to set or change these. + allow(System.commands).to receive(:build_cmd).and_wrap_original do |original, *args| + command = original.call(*args) + command.extra = { 'HTTP_PATH' => '/defined', 'HTTP_METHOD' => 'get' } + command + end + + captured = nil + allow(CommandDecomTopic).to receive(:write_packet) do |command, _scope| + captured = command + end + Thread.new { im.run } + sleep 0.01 + + caller_extra = { + 'HTTP_PATH' => '/supplied', 'HTTP_METHOD' => 'delete', + 'HTTP_HEADERS' => { 'authorization' => 'supplied' }, 'flow_uuid' => '1234-5678' + } + @api.cmd("INST", "ABORT", extra: caller_extra) + sleep 0.01 + im.shutdown + + expect(captured).to_not be_nil + expect(captured.extra['HTTP_PATH']).to eql('/defined') + expect(captured.extra['HTTP_METHOD']).to eql('get') + expect(captured.extra).not_to have_key('HTTP_HEADERS') + expect(captured.extra).not_to have_key('flow_uuid') + expect(captured.extra['metadata']).to eql(caller_extra) end it "acks SUCCESS for a write protocol STOP but rejects a DISCONNECT" do diff --git a/openc3/spec/microservices/queue_microservice_spec.rb b/openc3/spec/microservices/queue_microservice_spec.rb index 75fb777558..401cf821b2 100644 --- a/openc3/spec/microservices/queue_microservice_spec.rb +++ b/openc3/spec/microservices/queue_microservice_spec.rb @@ -86,7 +86,10 @@ module OpenC3 describe '#process_queued_commands' do let(:command1) { { 'username' => 'test_user', 'value' => 'cmd("TARGET", "COMMAND", {"PARAM": 1})' } } let(:command2) { { 'username' => 'test_user', 'value' => 'cmd("TARGET", "COMMAND2", {"PARAM": 2})' } } - let(:command3_new_format) { { 'username' => 'test_user', 'target_name' => 'TARGET', 'cmd_name' => 'COMMAND3', 'cmd_params' => JSON.generate({ 'PARAM' => 3 }) } } + let(:command3_new_format) do + { 'username' => 'test_user', 'target_name' => 'TARGET', 'cmd_name' => 'COMMAND3', + 'cmd_params' => JSON.generate({ 'PARAM' => 3 }), 'extra' => JSON.generate({ 'flow_uuid' => '1234-5678' }) } + end let(:command4_new_format) { { 'username' => 'test_user', 'target_name' => 'TARGET', 'cmd_name' => 'COMMAND4' } } before do @@ -114,9 +117,9 @@ module OpenC3 expect(Store).to have_received(:bzpopmin).exactly(3).times expect(processor).to have_received(:cmd) - .with(command1['value'], queue: false, scope: scope, timeout: nil, validate: true, queue_username: 'test_user') + .with(command1['value'], queue: false, scope: scope, timeout: nil, validate: true, queue_username: 'test_user', extra: nil) expect(processor).to have_received(:cmd) - .with(command2['value'], queue: false, scope: scope, timeout: nil, validate: true, queue_username: 'test_user') + .with(command2['value'], queue: false, scope: scope, timeout: nil, validate: true, queue_username: 'test_user', extra: nil) end it 'processes commands with new format (target_name, cmd_name, cmd_params)' do @@ -136,7 +139,8 @@ module OpenC3 expect(Store).to have_received(:bzpopmin).exactly(2).times expect(processor).to have_received(:cmd) - .with('TARGET', 'COMMAND3', { 'PARAM' => 3 }, queue: false, scope: scope, timeout: nil, validate: true, queue_username: 'test_user') + .with('TARGET', 'COMMAND3', { 'PARAM' => 3 }, queue: false, scope: scope, timeout: nil, validate: true, + queue_username: 'test_user', extra: { 'flow_uuid' => '1234-5678' }) end it 'processes commands with new format without cmd_params' do @@ -156,7 +160,7 @@ module OpenC3 expect(Store).to have_received(:bzpopmin).exactly(2).times expect(processor).to have_received(:cmd) - .with('TARGET', 'COMMAND4', {}, queue: false, scope: scope, timeout: nil, validate: true, queue_username: 'test_user') + .with('TARGET', 'COMMAND4', {}, queue: false, scope: scope, timeout: nil, validate: true, queue_username: 'test_user', extra: nil) end it 'processes mixed legacy and new format commands' do @@ -178,9 +182,10 @@ module OpenC3 expect(Store).to have_received(:bzpopmin).exactly(3).times expect(processor).to have_received(:cmd) - .with(command1['value'], queue: false, scope: scope, timeout: nil, validate: true, queue_username: 'test_user') + .with(command1['value'], queue: false, scope: scope, timeout: nil, validate: true, queue_username: 'test_user', extra: nil) expect(processor).to have_received(:cmd) - .with('TARGET', 'COMMAND3', { 'PARAM' => 3 }, queue: false, scope: scope, timeout: nil, validate: true, queue_username: 'test_user') + .with('TARGET', 'COMMAND3', { 'PARAM' => 3 }, queue: false, scope: scope, timeout: nil, validate: true, + queue_username: 'test_user', extra: { 'flow_uuid' => '1234-5678' }) end it 'processes legacy command with validate false and timeout 0' do @@ -200,7 +205,7 @@ module OpenC3 processor.process_queued_commands expect(processor).to have_received(:cmd) - .with(command_no_validate['value'], queue: false, scope: scope, timeout: 0, validate: false, queue_username: 'test_user') + .with(command_no_validate['value'], queue: false, scope: scope, timeout: 0, validate: false, queue_username: 'test_user', extra: nil) end it 'processes new format command with validate false and timeout 0' do @@ -220,7 +225,7 @@ module OpenC3 processor.process_queued_commands expect(processor).to have_received(:cmd) - .with('TARGET', 'COMMAND3', { 'PARAM' => 3 }, queue: false, scope: scope, timeout: 0, validate: false, queue_username: 'test_user') + .with('TARGET', 'COMMAND3', { 'PARAM' => 3 }, queue: false, scope: scope, timeout: 0, validate: false, queue_username: 'test_user', extra: nil) end it 'logs error for invalid command format (missing required fields)' do @@ -441,4 +446,4 @@ module OpenC3 end end end -end \ No newline at end of file +end diff --git a/openc3/spec/models/queue_model_spec.rb b/openc3/spec/models/queue_model_spec.rb index bfc1058ada..1be794c1e8 100644 --- a/openc3/spec/models/queue_model_spec.rb +++ b/openc3/spec/models/queue_model_spec.rb @@ -152,7 +152,9 @@ module OpenC3 model.create allow(QueueTopic).to receive(:write_notification) - QueueModel.queue_command("TEST", target_name: "INST", cmd_name: "COLLECT", cmd_params: { "TYPE" => "NORMAL" }, username: 'test_user', scope: "DEFAULT") + extra = { "flow_uuid" => "1234-5678", "data" => "\xFF".b } + QueueModel.queue_command("TEST", target_name: "INST", cmd_name: "COLLECT", cmd_params: { "TYPE" => "NORMAL" }, + extra: extra, username: 'test_user', scope: "DEFAULT") commands = Store.zrange("DEFAULT:TEST", 0, -1).map { |cmd| JSON.parse(cmd) } expect(commands).to contain_exactly({ @@ -160,6 +162,7 @@ module OpenC3 "target_name" => "INST", "cmd_name" => "COLLECT", "cmd_params" => "{\"TYPE\":\"NORMAL\"}", + "extra" => JSON.generate(extra.as_json, allow_nan: true), "validate" => true, "timestamp" => anything }) @@ -418,6 +421,17 @@ module OpenC3 decoded = result["DATA"]["raw"].pack('C*') expect(decoded).to eq(binary_data) end + + it "stores extra when inserting" do + allow(QueueTopic).to receive(:write_notification) + model = QueueModel.new(name: "TEST", scope: "DEFAULT") + + extra = { "flow_uuid" => "1234-5678", "data" => "\xFF".b } + model.insert_command(id: 1, username: "test_user", command: "TGT CMD", extra: extra) + + commands = Store.zrange("DEFAULT:TEST", 0, -1).map { |cmd| JSON.parse(cmd) } + expect(JSON.parse(commands[0]["extra"], allow_nan: true, create_additions: true)).to eql(extra) + end end describe "update_command" do @@ -486,6 +500,32 @@ module OpenC3 model.update_command(id: 1.0, command: "TGT CMD2", username: "user2") }.to raise_error(QueueError, "Queue 'TEST' is disabled. Command at id 1.0 not updated.") end + + it "carries forward existing extra when the update does not supply it" do + allow(QueueTopic).to receive(:write_notification) + model = QueueModel.new(name: "TEST", scope: "DEFAULT") + + extra = { "flow_uuid" => "1234-5678", "data" => "\xFF".b } + model.insert_command(id: 1.0, username: "user1", command: "TGT CMD1", extra: extra) + # The queue edit APIs have no way to express extra, so an update that omits + # it must not silently drop the metadata attached by cmd(extra: ...) + model.update_command(id: 1.0, command: "TGT CMD2", username: "user2") + + commands = Store.zrange("DEFAULT:TEST", 0, -1).map { |cmd| JSON.parse(cmd) } + expect(commands[0]["value"]).to eq("TGT CMD2") + expect(JSON.parse(commands[0]["extra"], allow_nan: true, create_additions: true)).to eql(extra) + end + + it "replaces existing extra when the update supplies it" do + allow(QueueTopic).to receive(:write_notification) + model = QueueModel.new(name: "TEST", scope: "DEFAULT") + + model.insert_command(id: 1.0, username: "user1", command: "TGT CMD1", extra: { "flow_uuid" => "old" }) + model.update_command(id: 1.0, command: "TGT CMD1", username: "user2", extra: { "flow_uuid" => "new" }) + + commands = Store.zrange("DEFAULT:TEST", 0, -1).map { |cmd| JSON.parse(cmd) } + expect(JSON.parse(commands[0]["extra"])).to eql({ "flow_uuid" => "new" }) + end end describe "list" do @@ -888,4 +928,4 @@ module OpenC3 end end end -end \ No newline at end of file +end diff --git a/openc3/spec/script/commands_spec.rb b/openc3/spec/script/commands_spec.rb index ec64c46bf6..c2a9ffd1ee 100644 --- a/openc3/spec/script/commands_spec.rb +++ b/openc3/spec/script/commands_spec.rb @@ -32,6 +32,7 @@ class CommandsSpecApi include Extract include Api include Authorization + attr_reader :last_kw_params def shutdown end @@ -43,6 +44,7 @@ def generate_url end def method_missing(name, *params, **kw_params) + @last_kw_params = kw_params self.send(name, *params, **kw_params) end end @@ -139,6 +141,20 @@ def setup_cmd_handler_thread end end + if connect == 'connected' + it "sends extra metadata" do + extra = { 'flow_uuid' => '1234-5678', 'hv_id' => 42 } + capture_io { cmd("INST ABORT", extra: extra) } + expect(@api.last_kw_params[:extra]).to eql(extra) + end + + it "does not send extra when not given" do + # Older servers reject an unknown extra keyword, even when nil + capture_io { cmd("INST ABORT") } + expect(@api.last_kw_params).not_to have_key(:extra) + end + end + it "raises without any parameters" do expect { cmd() }.to raise_error(/Invalid number of arguments/) end