Module: Meshtastic::ATAK

Defined in:
lib/meshtastic/atak.rb

Constant Summary collapse

V1_PORT = 
Meshtastic::PortNum::ATAK_PLUGIN
V2_PORT = 
Meshtastic::PortNum::ATAK_PLUGIN_V2
FORWARDER_PORT = 
Meshtastic::PortNum::ATAK_FORWARDER
V2_UNCOMPRESSED = 
0xFF
COORD_SCALE = 
10_000_000
SKIP_KEYS = 
i[
  serial_obj bluetooth_obj tcp_obj mqtt_obj to from channel want_ack hop_limit
  port_num data want_response via psks message extra_skip packet
].freeze

Class Method Summary collapse

Class Method Details

.authors ⇒ Object 



121
122
123
 
# File 'lib/meshtastic/atak.rb', line 121

public_class_method def self.authors
  "AUTHOR(S):\n        0day Inc. <[email protected]>\n      "
end

.build_v2(opts = {}) ⇒ Object 



35
36
37
38
39
40
41
42
43
 
# File 'lib/meshtastic/atak.rb', line 35

public_class_method def self.build_v2(opts = {})
  packet = opts[:packet] || Meshtastic::TAKPacketV2.new
  assign_fields(opts.merge(message: packet, extra_skip: i[packet lat lon altitude message to]))
  packet.latitude_i = scale_coord(value: opts[:lat]) if opts[:lat]
  packet.longitude_i = scale_coord(value: opts[:lon]) if opts[:lon]
  packet.altitude = opts[:altitude].to_i if opts[:altitude]
  packet.chat = chat_from(opts) if chat?(opts) && opts[:chat].nil?
  packet
end

.compress_cot(opts = {}) ⇒ Object 



68
69
70
 
# File 'lib/meshtastic/atak.rb', line 68

public_class_method def self.compress_cot(opts = {})
  Zlib::Deflate.deflate(opts[:cot].to_s)
end

.decode(opts = {}) ⇒ Object 



78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/meshtastic/atak.rb', line 78

public_class_method def self.decode(opts = {})
  payload = opts[:payload]
  portnum = normalize_port(portnum: opts[:portnum])
  case portnum
  when V1_PORT, :ATAK_PLUGIN
    Meshtastic::TAKPacket.decode(payload.to_s.b)
  when V2_PORT, :ATAK_PLUGIN_V2
    decode_v2(payload: payload)
  when FORWARDER_PORT, :ATAK_FORWARDER
    decompress_cot(payload: payload)
  else
    raise ArgumentError, "unsupported TAK portnum: #{opts[:portnum].inspect}"
  end
end

.decode_v2(opts = {}) ⇒ Object

Raises:



54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/meshtastic/atak.rb', line 54

public_class_method def self.decode_v2(opts = {})
  bytes = (opts[:wire] || opts[:payload]).to_s.b
  raise ArgumentError, 'empty ATAK V2 payload' if bytes.empty?

  flags = bytes.getbyte(0)
  body = bytes.byteslice(1..)
  if flags == V2_UNCOMPRESSED
    Meshtastic::TAKPacketV2.decode(body)
  else
    raise ArgumentError,
          "compressed ATAK V2 dictionary id #{flags & 0x3F} is not unpacked (flags=0x#{flags.to_s(16)})"
  end
end

.decompress_cot(opts = {}) ⇒ Object 



72
73
74
75
76
 
# File 'lib/meshtastic/atak.rb', line 72

public_class_method def self.decompress_cot(opts = {})
  bytes = opts[:payload].to_s.b
  bytes = bytes.byteslice(1..) if bytes.getbyte(0).zero? && bytes.bytesize > 1
  Zlib::Inflate.inflate(bytes)
end

.encode(opts = {}) ⇒ Object 



19
20
21
 
# File 'lib/meshtastic/atak.rb', line 19

public_class_method def self.encode(opts = {})
  encode_v1(opts.merge({}))
end

.encode_v1(opts = {}) ⇒ Object 



23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/meshtastic/atak.rb', line 23

public_class_method def self.encode_v1(opts = {})
  packet = opts[:packet] || Meshtastic::TAKPacket.new
  packet.is_compressed = opts.fetch(:is_compressed, packet.is_compressed)
  packet.contact = contact_from(opts) if contact?(opts)
  packet.group = group_from(opts) if group?(opts)
  packet.status = status_from(opts) if status?(opts)
  packet.pli = pli_from(opts) if pli?(opts)
  packet.chat = chat_from(opts) if chat?(opts)
  packet.detail = opts[:detail] if opts[:detail]
  packet
end

.encode_v2(opts = {}) ⇒ Object 



50
51
52
 
# File 'lib/meshtastic/atak.rb', line 50

public_class_method def self.encode_v2(opts = {})
  wrap_v2(packet: build_v2(opts))
end

.help ⇒ Object 



125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
 
# File 'lib/meshtastic/atak.rb', line 125

public_class_method def self.help
  puts "USAGE:
    # Encode a V1 TAKPacket (alias of encode_v1).
    #{self}.encode(
      message: 'optional - GeoChat text body to place on the packet',
      packet: 'optional - existing Meshtastic::TAKPacket to fill instead of a new one'
    )

    # Build a V1 TAKPacket with GeoChat, PLI, contact, group, or status.
    #{self}.encode_v1(
      packet: 'optional - existing Meshtastic::TAKPacket to fill instead of a new one',
      detail: 'optional - raw CoT detail bytes for the V1 detail field'
    )

    # Build a V2 TAKPacketV2 with scaled lat/lon and typed payloads.
    #{self}.build_v2(
      packet: 'optional - existing Meshtastic::TAKPacketV2 to fill instead of a new one',
      chat: 'optional - Meshtastic::GeoChat object if not using message:',
      lat: 'optional - latitude in decimal degrees (stored times 1e7)',
      lon: 'optional - longitude in decimal degrees (stored times 1e7)',
      altitude: 'optional - altitude in meters as an integer'
    )

    # Prefix a TAKPacketV2 protobuf with the uncompressed V2 flags byte.
    #{self}.wrap_v2(
      packet: 'required - Meshtastic::TAKPacketV2 to serialize onto the wire'
    )

    # Build and wrap an uncompressed V2 wire frame.
    #{self}.encode_v2(
      message: 'optional - GeoChat text for the V2 chat variant'
    )

    # Decode an uncompressed V2 wire frame (flags 0xFF plus protobuf).
    #{self}.decode_v2(
      wire: 'optional - full V2 wire bytes including the flags byte',
      payload: 'optional - same as wire when wire is omitted'
    )

    # zlib-compress Cursor-on-Target XML for ATAK_FORWARDER.
    #{self}.compress_cot(
      cot: 'required - CoT XML string to deflate for the forwarder port'
    )

    # zlib-decompress ATAK_FORWARDER payload bytes back to CoT XML.
    #{self}.decompress_cot(
      payload: 'required - zlib bytes from an ATAK_FORWARDER Data payload'
    )

    # Dispatch decode by Meshtastic TAK portnum.
    #{self}.decode(
      payload: 'required - raw port payload bytes from a decoded mesh Data',
      portnum: 'required - :ATAK_PLUGIN, :ATAK_PLUGIN_V2, or :ATAK_FORWARDER'
    )

    # Send a V1 TAKPacket (alias of send_v1).
    #{self}.send(
      serial_obj: 'optional - serial handle from Meshtastic::Serial.connect',
      message: 'optional - GeoChat text to send on ATAK_PLUGIN'
    )

    # Send a V1 TAKPacket on ATAK_PLUGIN (port 72).
    #{self}.send_v1(
      serial_obj: 'optional - serial handle from Meshtastic::Serial.connect',
      bluetooth_obj: 'optional - BLE handle from Meshtastic::Bluetooth.connect',
      tcp_obj: 'optional - TCP handle from Meshtastic::TCP.connect'
    )

    # Send V1 GeoChat on ATAK_PLUGIN.
    #{self}.send_chat(
      serial_obj: 'optional - serial handle from Meshtastic::Serial.connect',
      message: 'optional - GeoChat text body to transmit'
    )

    # Send V1 PLI on ATAK_PLUGIN.
    #{self}.send_pli(
      serial_obj: 'optional - serial handle from Meshtastic::Serial.connect',
      lat: 'optional - latitude in decimal degrees (stored times 1e7)',
      lon: 'optional - longitude in decimal degrees (stored times 1e7)'
    )

    # Send an uncompressed V2 frame on ATAK_PLUGIN_V2 (port 78).
    #{self}.send_v2(
      serial_obj: 'optional - serial handle from Meshtastic::Serial.connect',
      message: 'optional - GeoChat text for the V2 chat variant'
    )

    # Send zlib-compressed CoT XML on ATAK_FORWARDER (port 257).
    #{self}.send_cot(
      serial_obj: 'optional - serial handle from Meshtastic::Serial.connect',
      cot: 'required - CoT XML string small enough to fit after zlib'
    )

    # Print the AUTHOR(S) string for this module.
    #{self}.authors
  "
end

.send(opts = {}) ⇒ Object 



93
94
95
 
# File 'lib/meshtastic/atak.rb', line 93

public_class_method def self.send(opts = {})
  send_v1(opts.merge({}))
end

.send_chat(opts = {}) ⇒ Object 



101
102
103
 
# File 'lib/meshtastic/atak.rb', line 101

public_class_method def self.send_chat(opts = {})
  send_v1(opts.merge({}))
end

.send_cot(opts = {}) ⇒ Object

Raises:



113
114
115
116
117
118
119
 
# File 'lib/meshtastic/atak.rb', line 113

public_class_method def self.send_cot(opts = {})
  payload = compress_cot(cot: opts[:cot])
  max_len = Meshtastic::Constants::DATA_PAYLOAD_LEN
  raise ArgumentError, "ERROR: CoT Length > #{max_len} Bytes after zlib" if payload.bytesize > max_len

  deliver(opts.merge(port_num: FORWARDER_PORT, payload: payload))
end

.send_pli(opts = {}) ⇒ Object 



105
106
107
 
# File 'lib/meshtastic/atak.rb', line 105

public_class_method def self.send_pli(opts = {})
  send_v1(opts.merge({}))
end

.send_v1(opts = {}) ⇒ Object 



97
98
99
 
# File 'lib/meshtastic/atak.rb', line 97

public_class_method def self.send_v1(opts = {})
  deliver(opts.merge(port_num: V1_PORT, payload: encode_v1(opts).to_proto))
end

.send_v2(opts = {}) ⇒ Object 



109
110
111
 
# File 'lib/meshtastic/atak.rb', line 109

public_class_method def self.send_v2(opts = {})
  deliver(opts.merge(port_num: V2_PORT, payload: encode_v2(opts)))
end

.wrap_v2(opts = {}) ⇒ Object 



45
46
47
48
 
# File 'lib/meshtastic/atak.rb', line 45

public_class_method def self.wrap_v2(opts = {})
  packet = opts[:packet]
  [V2_UNCOMPRESSED].pack('C') + packet.to_proto
end