Module: Meshtastic::Admin::Backup

Defined in:

lib/meshtastic/admin/backup.rb

Overview

Host-side snapshots, not firmware filesystem backup commands.

Constant Summary

FORMAT =

'meshtastic-admin-backup'

WARNING =

'Contains secrets returned by firmware. Firmware may redact secrets; absent/default values are not proof of completeness. Owner identity and hardware metadata are excluded. ACKs do not prove persistence.'

CONFIG_TYPES =

%i[DEVICE_CONFIG POSITION_CONFIG POWER_CONFIG NETWORK_CONFIG DISPLAY_CONFIG LORA_CONFIG BLUETOOTH_CONFIG SECURITY_CONFIG].freeze

MODULE_TYPES =

%i[MQTT_CONFIG SERIAL_CONFIG EXTNOTIF_CONFIG STOREFORWARD_CONFIG RANGETEST_CONFIG TELEMETRY_CONFIG CANNEDMSG_CONFIG AUDIO_CONFIG REMOTEHARDWARE_CONFIG NEIGHBORINFO_CONFIG AMBIENTLIGHTING_CONFIG DETECTIONSENSOR_CONFIG PAXCOUNTER_CONFIG STATUSMESSAGE_CONFIG TRAFFICMANAGEMENT_CONFIG TAK_CONFIG MESHBEACON_CONFIG].freeze

OWNER_FIELDS =

%w[long_name short_name is_licensed].freeze

Class Method Summary

• .authors ⇒ Object
• .export(opts = {}) ⇒ Object
• .help ⇒ Object
• .import(opts = {}) ⇒ Object

Class Method Details

.authors ⇒ Object

# File 'lib/meshtastic/admin/backup.rb', line 515

public_class_method def self.authors
  "AUTHOR(S):\n        0day Inc. <support@0dayinc.com>\n"
end

.export(opts = {}) ⇒ Object

# File 'lib/meshtastic/admin/backup.rb', line 31

public_class_method def self.export(opts = {})
  validate_options(options: opts, operation: :export)
  connection = connection_options(opts)
  return export_profile(opts.merge(connection: connection)) if opts.fetch(:format, :json) == :device_profile

  records = []
  selection(opts).each do |section, slot|
    value = Admin.request(connection.merge(read_command(section: section, slot: slot)))[:value]
    value = Meshtastic::User.new(long_name: value.long_name, short_name: value.short_name, is_licensed: value.is_licensed) if section == 'owner'
    json = JSON.parse(value.class.encode_json(value, emit_defaults: true, preserve_proto_fieldnames: true))
    json.select! { |key, _| OWNER_FIELDS.include?(key) } if section == 'owner'
    records << { 'section' => section, 'slot' => slot, 'value' => json }
  end
  document = { 'format' => FORMAT, 'version' => 1, 'warning' => WARNING, 'records' => records }
  validate_document(document: document)
  if opts[:path]
    File.open(opts[:path], File::WRONLY | File::CREAT | File::EXCL | File::NOFOLLOW, 0o600) do |file|
      file.chmod(0o600)
      file.write(JSON.pretty_generate(document))
      file.flush
      file.fsync
    end
  end
  { status: :exported, count: records.length, backup: document, warnings: [WARNING] }
end

.help ⇒ Object

# File 'lib/meshtastic/admin/backup.rb', line 519

public_class_method def self.help
  puts "USAGE:
    # Export fresh selected configuration sections.
    #{self}.export(
      transport_obj: 'required - connected Serial, Bluetooth or TCP handle',
      format: 'optional - :json (default version-1 Hash) or :device_profile (binary String in result[:backup]); result also has status, count and warnings',
      path: 'optional - new secret JSON or binary .cfg file matching format; never overwritten; mode 0600',
      to: 'optional - explicit unicast target; defaults to local node',
      timeout: 'optional - positive finite per-request seconds; default 10',
      config_types: 'optional - ConfigType symbols; default eight writable core sections',
      module_config_types: 'optional - ModuleConfigType symbols; default empty',
      channel_indexes: 'optional - unique zero-based indexes 0 through 7; default all eight; binary requires contiguous prefix from 0 with PRIMARY then SECONDARY roles, no disabled slots',
      include_owner: 'optional - portable owner names and license flag; binary also preserves present is_unmessagable; default true',
      include_ui: 'optional - dedicated UI configuration; default false; true is rejected for binary before requests',
      fixed_position: 'optional - binary only: explicit Position protobuf or field Hash; no getter exists; omitted by default',
      include_ringtone: 'optional - binary only: fresh ringtone including empty string; boolean default false',
      include_canned_messages: 'optional - binary only: fresh canned messages including empty string; boolean default false',
      channel: 'optional - mesh transport channel index; default Admin behavior',
      hop_limit: 'optional - mesh transport hop limit; default Admin behavior'
    )
    # Restore with bounded ACKs or exact timeout readback; never replay writes.
    #{self}.import(
      transport_obj: 'required - connected Serial, Bluetooth or TCP handle; no MQTT',
      path: 'optional - existing JSON or binary DeviceProfile .cfg file; mutually exclusive with backup',
      backup: 'optional - versioned JSON Hash, JSON String or binary DeviceProfile String; mutually exclusive with path',
      format: 'optional - :auto (default), :json or :device_profile; cfg paths select binary, other content is validated',
      dry_run: 'optional - validate and plan without any radio requests; default false',
      edit_transaction: 'optional - begin/commit on firmware known to support edits; default false',
      verify: 'optional - additional fresh comparison after completed writes/commit; timeout recovery reads occur regardless; not durability proof; default false',
      timeout: 'optional - positive finite seconds per request; default 10',
      to: 'optional - explicit unicast target; defaults to connected local node',
      channel: 'optional - mesh transport channel index; default Admin behavior',
      hop_limit: 'optional - mesh transport hop limit; default Admin behavior'
    )
    # Contains secrets; firmware may redact values. Never logs document contents.
    # Display the module authors.
    #{self}.authors
  "
end

.import(opts = {}) ⇒ Object

# File 'lib/meshtastic/admin/backup.rb', line 115

public_class_method def self.import(opts = {})
  validate_options(options: opts, operation: :import)
  connection = connection_options(opts)
  raise ArgumentError, 'supply exactly one of path or backup' unless opts.key?(:path) ^ opts.key?(:backup)

  document = opts[:backup]
  if opts.key?(:path)
    document = File.open(opts[:path], File::RDONLY | File::NOFOLLOW) do |file|
      raise ArgumentError, 'backup path must be a regular file' unless file.stat.file?

      file.binmode
      file.read
    end
  end
  entries = import_plan(document: document, format: opts.fetch(:format, :auto), path: opts[:path])
  plan = entries.each_with_index.sort_by do |entry, index|
    priority = if entry[:section] == 'config'
                 { 'LORA_CONFIG' => 20, 'BLUETOOTH_CONFIG' => 21, 'NETWORK_CONFIG' => 22, 'SECURITY_CONFIG' => 23 }.fetch(entry[:slot], 0)
               else
                 entry[:section] == 'channel' ? 10 : 0
               end
    [priority, index]
  end.map(&:first)
  report = { status: :dry_run, planned: plan.length, attempted: 0, acknowledged: 0, readback_confirmed: 0, persistence_verified: false,
             transaction: :not_requested, records: [], plan: plan.map { |entry| entry.slice(:section, :slot) }, warnings: [WARNING] }
  return report if opts.fetch(:dry_run, false)

  failure = nil
  if opts.fetch(:edit_transaction, false) && !plan.empty?
    failure = { operation: :begin_edit_settings }
    report[:transaction] = :begin_uncertain
    Admin.request(connection.merge(begin_edit_settings: true))
    report[:transaction] = :open
  end
  plan.each do |entry|
    field = { 'owner' => :set_owner, 'config' => :set_config, 'module_config' => :set_module_config,
              'channel' => :set_channel, 'ui' => :store_ui_config, 'fixed_position' => :set_fixed_position,
              'ringtone' => :set_ringtone_message, 'canned_messages' => :set_canned_message_module_messages }.fetch(entry[:section])
    failure = entry.slice(:section, :slot).merge(operation: field)
    report[:attempted] += 1
    status = apply_entry(connection: connection, field: field, entry: entry, failure: failure)
    report[status] += 1
    report[:records] << entry.slice(:section, :slot).merge(status: status)
  end
  if report[:transaction] == :open
    failure = { operation: :commit_edit_settings }
    report[:transaction] = :commit_uncertain
    Admin.request(connection.merge(commit_edit_settings: true))
    report[:transaction] = :commit_acknowledged
  end
  report[:status] = report[:readback_confirmed].positive? ? :applied : :acknowledged
  verify_readback(connection: connection, plan: plan, report: report) if opts.fetch(:verify, false)
  report
rescue StandardError => e
  raise unless failure

  report[:status] = :partial_failure
  report[:failure] = failure.merge(error: e.class.name)
  report[:failure][:reason] = e.reason if e.is_a?(Admin::RoutingError)
  report
end