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 



114
115
116
 
# File 'lib/meshtastic/atak.rb', line 114

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

.build_v2(opts = {}) ⇒ Object 



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

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 



61
62
63
 
# File 'lib/meshtastic/atak.rb', line 61

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

.decode(opts = {}) ⇒ Object 



71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/meshtastic/atak.rb', line 71

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(V1Strings.new(payload.to_s.b).decode)
  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 



56
57
58
59
 
# File 'lib/meshtastic/atak.rb', line 56

public_class_method def self.decode_v2(opts = {})
  body = Meshtastic::PayloadCompression.decode_v2(payload: opts[:wire] || opts[:payload])
  Meshtastic::TAKPacketV2.decode(body)
end

.decompress_cot(opts = {}) ⇒ Object 



65
66
67
68
69
 
# File 'lib/meshtastic/atak.rb', line 65

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 



21
22
23
 
# File 'lib/meshtastic/atak.rb', line 21

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

.encode_v1(opts = {}) ⇒ Object 



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

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 



52
53
54
 
# File 'lib/meshtastic/atak.rb', line 52

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

.help ⇒ Object 



118
119
120
121
122
123
124
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
 
# File 'lib/meshtastic/atak.rb', line 118

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 a bounded V2 frame using official dictionaries or raw flags 0xFF.
    #{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 



86
87
88
 
# File 'lib/meshtastic/atak.rb', line 86

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

.send_chat(opts = {}) ⇒ Object 



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

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

.send_cot(opts = {}) ⇒ Object

Raises:



106
107
108
109
110
111
112
 
# File 'lib/meshtastic/atak.rb', line 106

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 



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

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

.send_v1(opts = {}) ⇒ Object 



90
91
92
 
# File 'lib/meshtastic/atak.rb', line 90

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 



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

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

.wrap_v2(opts = {}) ⇒ Object 



47
48
49
50
 
# File 'lib/meshtastic/atak.rb', line 47

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