[codex] secure extension hot reload control

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## [0.3.8] - 2026-04-25
+
+### Added
+- Configurable cluster-control payload signing via `extensions.node.cluster_control.auth.mode` (`auto`, `required`, `disabled`), with timestamp freshness and nonce format validation when signing is active.
+
+### Changed
+- `ClusterControl` queues now read durability and retention settings from `extensions.node.cluster_control.queue`; defaults remain durable and non-auto-delete so per-node commands survive node restarts.
+- Beat no longer runs immediately at actor construction, avoiding startup reconciliation before extension boot completes.
+- `update_gem` now installs through `Legion::Extensions::GemSource` and calls extension-scoped reload instead of full daemon reload.
+
 ## [0.3.7] - 2026-03-31
 
 ### Fixed

README.md

@@ -68,6 +68,37 @@ Each node periodically calls `beat` to broadcast its presence. On startup, nodes
 
 Dynamic config changes (`update_settings`) and gem upgrades (`update_gem`) can be pushed to individual nodes or broadcast to all nodes at runtime. Both operations publish an `UpdateResult` message to `node.<name>.update_result` so the outcome can be observed cluster-wide.
 
+### Cluster Control Settings
+
+Cluster-control broadcasts are configurable under `extensions.node.cluster_control`. By default, auth mode is `auto`: messages are signed and verified when a shared secret is configured, and unsigned messages are allowed when no secret exists for simpler local or homelab deployments.
+
+```json
+{
+  "extensions": {
+    "node": {
+      "cluster_control": {
+        "auth": {
+          "mode": "auto",
+          "timestamp_skew_seconds": 300,
+          "nonce_bytes": 16
+        },
+        "queue": {
+          "durable": true,
+          "exclusive": false,
+          "auto_delete": false,
+          "queue_type": "classic",
+          "expires_ms": 604800000,
+          "message_ttl_ms": 86400000,
+          "max_length": 1000
+        }
+      }
+    }
+  }
+}
+```
+
+Set `auth.mode` to `required` to force HMAC signatures, or `disabled` to run without cluster-control signatures even when a shared secret is present.
+
 ## Transport
 
 - **Exchange**: `node` (topic exchange)

lib/legion/extensions/node.rb

@@ -1,12 +1,17 @@
 # frozen_string_literal: true
 
 require 'legion/extensions/node/version'
+require 'legion/extensions/node/config'
 require 'legion/extensions/node/helpers/rabbitmq'
 
 module Legion
   module Extensions
     module Node
       extend Legion::Extensions::Core if Legion::Extensions.const_defined? :Core, false
+
+      def self.default_settings
+        Config.default_settings
+      end
     end
   end
 end

lib/legion/extensions/node/actors/beat.rb

@@ -22,7 +22,7 @@ def generate_task?
           end
 
           def run_now?
-            true
+            false
           end
 
           def time

lib/legion/extensions/node/actors/cluster_control.rb

@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require 'legion/extensions/node/control_auth'
+
 module Legion
   module Extensions
     module Node
@@ -28,6 +30,11 @@ def check_subtask?
           def generate_task?
             false
           end
+
+          def process_message(message, metadata, delivery_info)
+            verified_message = Legion::Extensions::Node::ControlAuth.verify!(message)
+            super(verified_message, metadata, delivery_info)
+          end
         end
       end
     end

lib/legion/extensions/node/config.rb

@@ -0,0 +1,90 @@
+# frozen_string_literal: true
+
+module Legion
+  module Extensions
+    module Node
+      module Config
+        DEFAULT_SETTINGS = {
+          cluster_control: {
+            auth:  {
+              mode:                   'auto',
+              timestamp_skew_seconds: 300,
+              nonce_bytes:            16
+            },
+            queue: {
+              durable:        true,
+              exclusive:      false,
+              auto_delete:    false,
+              queue_type:     'classic',
+              expires_ms:     604_800_000,
+              message_ttl_ms: 86_400_000,
+              max_length:     1000
+            }
+          }
+        }.freeze
+
+        module_function
+
+        def default_settings
+          deep_dup(DEFAULT_SETTINGS)
+        end
+
+        def cluster_control
+          deep_merge(default_settings[:cluster_control], extension_settings[:cluster_control] || {})
+        end
+
+        def control_auth
+          cluster_control[:auth] || {}
+        end
+
+        def control_queue
+          cluster_control[:queue] || {}
+        end
+
+        def extension_settings
+          return {} unless defined?(Legion::Settings) && Legion::Settings.respond_to?(:dig)
+
+          settings = Legion::Settings.dig(:extensions, :node)
+          settings.is_a?(Hash) ? symbolize_keys(settings) : {}
+        rescue StandardError => e
+          log.debug("node settings lookup failed: #{e.message}")
+          {}
+        end
+
+        def deep_merge(base, override)
+          base_hash = symbolize_keys(base)
+          override_hash = symbolize_keys(override)
+          base_hash.merge(override_hash) do |_key, old_value, new_value|
+            old_value.is_a?(Hash) && new_value.is_a?(Hash) ? deep_merge(old_value, new_value) : new_value
+          end
+        end
+
+        def deep_dup(value)
+          case value
+          when Hash
+            value.each_with_object({}) { |(key, val), result| result[key] = deep_dup(val) }
+          when Array
+            value.map { |item| deep_dup(item) }
+          else
+            value
+          end
+        end
+
+        def symbolize_keys(value)
+          case value
+          when Hash
+            value.each_with_object({}) { |(key, val), result| result[key.to_sym] = symbolize_keys(val) }
+          when Array
+            value.map { |item| symbolize_keys(item) }
+          else
+            value
+          end
+        end
+
+        def log
+          Legion::Logging.respond_to?(:logger) ? Legion::Logging.logger : Legion::Logging
+        end
+      end
+    end
+  end
+end

lib/legion/extensions/node/control_auth.rb

@@ -0,0 +1,186 @@
+# frozen_string_literal: true
+
+require 'openssl'
+require 'securerandom'
+require 'json'
+require 'legion/extensions/node/config'
+
+module Legion
+  module Extensions
+    module Node
+      module ControlAuth
+        AUTH_MODES = %w[auto required disabled].freeze
+
+        class UnauthorizedControlMessage < StandardError; end
+
+        module_function
+
+        def sign(payload)
+          normalized = deep_symbolize(payload).dup
+          return normalized unless sign_control_messages?
+
+          control = {
+            sender:    node_name,
+            timestamp: Time.now.utc.to_i,
+            nonce:     SecureRandom.hex(nonce_bytes)
+          }
+          normalized[:control] = control.merge(signature: signature_for(normalized.merge(control: control)))
+          normalized
+        end
+
+        def verify!(payload)
+          normalized = deep_symbolize(payload)
+          raise UnauthorizedControlMessage, 'missing control signature' unless normalized.is_a?(Hash)
+          return normalized unless verify_control_messages?
+
+          control = normalized[:control]
+          raise UnauthorizedControlMessage, 'missing control signature' unless control.is_a?(Hash)
+
+          signature = control[:signature].to_s
+          raise UnauthorizedControlMessage, 'missing control signature' if signature.empty?
+          raise UnauthorizedControlMessage, 'stale control message' unless fresh_timestamp?(control[:timestamp])
+          raise UnauthorizedControlMessage, 'invalid control nonce' unless valid_nonce?(control[:nonce])
+
+          unsigned_control = control.dup
+          unsigned_control.delete(:signature)
+          expected = signature_for(normalized.merge(control: unsigned_control))
+          raise UnauthorizedControlMessage, 'invalid control signature' unless secure_compare(signature, expected)
+
+          normalized
+        end
+
+        def sign_control_messages?
+          return false if auth_disabled?
+          return true if auth_required?
+
+          !secret.to_s.empty?
+        end
+
+        def verify_control_messages?
+          return false if auth_disabled?
+          return true if auth_required?
+
+          !secret.to_s.empty?
+        end
+
+        def auth_mode
+          configured = auth_settings[:mode] || auth_settings[:enabled]
+          mode = case configured
+                 when true then 'required'
+                 when false then 'disabled'
+                 else configured.to_s
+                 end
+          AUTH_MODES.include?(mode) ? mode : 'auto'
+        end
+
+        def auth_required?
+          auth_mode == 'required'
+        end
+
+        def auth_disabled?
+          auth_mode == 'disabled'
+        end
+
+        def auth_settings
+          Legion::Extensions::Node::Config.control_auth
+        end
+
+        def secret
+          configured_secret || ENV.fetch('LEGION_NODE_CONTROL_SECRET', nil)
+        end
+
+        def configured_secret
+          if defined?(Legion::Settings) && Legion::Settings.respond_to?(:dig)
+            auth_settings[:secret] ||
+              Legion::Settings.dig(:cluster, :control_secret) ||
+              Legion::Settings.dig(:crypt, :cluster_secret)
+          end
+        rescue StandardError => e
+          log.debug("control secret lookup failed: #{e.message}")
+          nil
+        end
+
+        def signature_for(payload)
+          value = secret
+          raise UnauthorizedControlMessage, 'cluster control secret is not configured' if value.to_s.empty?
+
+          OpenSSL::HMAC.hexdigest('SHA256', value.to_s, canonical_json(payload))
+        end
+
+        def canonical_json(value)
+          ::JSON.generate(canonicalize(value))
+        end
+
+        def canonicalize(value)
+          case value
+          when Hash
+            value.each_with_object({}) do |(key, val), result|
+              result[key.to_s] = canonicalize(val)
+            end.sort.to_h
+          when Array
+            value.map { |item| canonicalize(item) }
+          else
+            value
+          end
+        end
+
+        def deep_symbolize(value)
+          case value
+          when Hash
+            value.each_with_object({}) do |(key, val), result|
+              result[key.to_sym] = deep_symbolize(val)
+            end
+          when Array
+            value.map { |item| deep_symbolize(item) }
+          else
+            value
+          end
+        end
+
+        def fresh_timestamp?(timestamp)
+          ts = Integer(timestamp)
+          (Time.now.utc.to_i - ts).abs <= timestamp_skew_seconds
+        rescue ArgumentError, TypeError => e
+          log.debug("control timestamp validation failed: #{e.message}")
+          false
+        end
+
+        def timestamp_skew_seconds
+          Integer(auth_settings[:timestamp_skew_seconds])
+        rescue ArgumentError, TypeError => e
+          log.debug("control timestamp skew setting invalid: #{e.message}")
+          300
+        end
+
+        def nonce_bytes
+          bytes = Integer(auth_settings[:nonce_bytes])
+          bytes.positive? ? bytes : 16
+        rescue ArgumentError, TypeError => e
+          log.debug("control nonce byte setting invalid: #{e.message}")
+          16
+        end
+
+        def valid_nonce?(nonce)
+          nonce.to_s.match?(/\A[0-9a-f]{#{nonce_bytes * 2}}\z/i)
+        end
+
+        def secure_compare(left, right)
+          return false unless left.bytesize == right.bytesize
+
+          left.bytes.zip(right.bytes).reduce(0) { |acc, (a, b)| acc | (a ^ b) }.zero?
+        end
+
+        def node_name
+          (Legion::Settings[:client]&.fetch(:name, nil) if defined?(Legion::Settings)) || 'unknown'
+        rescue StandardError => e
+          log.debug("node name lookup failed: #{e.message}")
+          'unknown'
+        end
+
+        def log
+          Legion::Logging.respond_to?(:logger) ? Legion::Logging.logger : Legion::Logging
+        end
+      end
+    end
+  end
+end