Module: PWN::SDR::Decoder::WiFi

Defined in:

lib/pwn/sdr/decoder/wifi.rb

Overview

Protocol frames via .decode; energy observations only via .detect.

Defined Under Namespace

Classes: DSSSIQ, DetectorIQ, MACReassembler

Constant Summary

SUPPORTED_MODES =

Realtime options forwarded to Base: on_frame (Hash callback), output (writable IO), interactive (default true), duration (seconds), stop (callable), queue_size (bounded chunks), log_file (path or false).

%i[dsss_1mbps].freeze

DemodIQ =

DSSSIQ

TSHARK_FIELDS =

%w[
  frame.time_relative wlan.fc.type_subtype wlan.bssid wlan.sa
  wlan.da wlan_radio.channel wlan_radio.signal_dbm wlan.ssid
].freeze

Class Method Summary

• .authors ⇒ Object

  Author(s)

  0day Inc.

• .decode(opts = {}) ⇒ Object

  Never silently downgrade decode to energy detection.

• .detect(opts = {}) ⇒ Object
• .help ⇒ Object
• .parse_frame(opts = {}) ⇒ Object
• .parse_line(opts = {}) ⇒ Object
• .plcp_crc(opts = {}) ⇒ Object

  IEEE 802.11b 18.2.3.6, reflected CCITT, complemented remainder.

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com

# File 'lib/pwn/sdr/decoder/wifi.rb', line 399

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

.decode(opts = {}) ⇒ Object

Never silently downgrade decode to energy detection.

Raises:

• (ArgumentError)

# File 'lib/pwn/sdr/decoder/wifi.rb', line 338

public_class_method def self.decode(opts = {})
  mode = opts.fetch(:mode, :dsss_1mbps).to_s.to_sym
  raise ArgumentError, "unsupported WiFi mode #{mode}; supported: #{SUPPORTED_MODES.join(', ')}" unless SUPPORTED_MODES.include?(mode)

  freq_obj = opts[:freq_obj] || {}
  rate = opts[:sample_rate] || freq_obj[:iq_rate] || 11_000_000
  demod = DSSSIQ.new(rate: rate)
  PWN::SDR::Decoder::Base.run_iq(opts.merge(freq_obj: freq_obj, protocol: 'WiFi',
                                            sample_rate: rate, demod: demod, fallback: :raise))
end

.detect(opts = {}) ⇒ Object

# File 'lib/pwn/sdr/decoder/wifi.rb', line 349

public_class_method def self.detect(opts = {})
  freq_obj = opts[:freq_obj]

  rate  = (opts[:sample_rate] || freq_obj[:iq_rate] || 2_000_000).to_i
  proto = 'WiFi-802.11'
  demod = DetectorIQ.new(
    rate: rate, protocol: proto, modulation: 'OFDM',
    extra: { threshold: 6.0 }
  )
  PWN::SDR::Decoder::Base.run_iq(
    **opts,
    freq_obj: freq_obj,
    protocol: proto,
    sample_rate: rate,
    source: opts[:source],
    file: opts[:file],
    demod: demod,
    threshold: 6.0,
    note: '20+ MHz OFDM — true-air I/Q path reports channel occupancy/duty.',
    describe: proc { |b| { modulation: 'OFDM', airtime_ms: b[:duration_ms] } }
  )
end

.help ⇒ Object

# File 'lib/pwn/sdr/decoder/wifi.rb', line 403

public_class_method def self.help
  puts "USAGE:
    # Supported mode: :dsss_1mbps only (default). Long PLCP, 1 Mbps DBPSK/Barker.
    # sample_rate: 11000000 (default), 22000000 or 44000000; centred IQ.
    # MAC: management/data (+QoS/WDS) headers, legacy control, beacon/probe IEs,
    # LLC/SNAP for unfragmented non-aggregated plaintext. Other bodies remain hex.
    # No OFDM, CCK, 2 Mbps, short preamble, decryption, or HT control decoding.
    # Ordered plaintext data-fragment reassembly: 64 flows, 2304 bytes, 1s capture-time expiry.
    # Final fragment includes msdu_hex/LLC fields; individual MPDU payload_hex stays unchanged.
    # Protected/aggregated frames stay opaque. No keys guessed, no ciphertext parsing.
    # No equalizer, resampler or clock-drift loop. Offline/streaming, NOT realtime guaranteed.
    # decode never falls back to a detector; unsupported mode/rate raises ArgumentError.
    # #{self}.detect uses the same source/lifecycle options for energy-only observations.
    # Check MAC FCS and return parsed frame fields, not PHY CRC status.
    #{self}.parse_frame(bytes: 'required - binary MAC frame String including trailing FCS')
    # Calculate the PLCP CRC16 over header octets.
    #{self}.plcp_crc(bytes: 'required - Array of PLCP header octets before the CRC')
    # Run decode and return its result
    #{self}.decode(
      freq_obj: 'required - freq_obj returned from PWN::SDR::GQRX.init_freq',
      on_frame: 'optional - callback receiving each emitted Hash',
      output: 'optional - writable IO (default stdout)',
      interactive: 'optional - false disables ENTER input',
      duration: 'optional - finite seconds to run',
      stop: 'optional - callable returning true to stop',
      queue_size: 'optional - bounded pending chunks (default 8)',
      log_file: 'optional - JSONL path or false to disable logging',
      sample_rate: 'optional - sample rate value consumed by #decode',
      source: 'optional - source value consumed by #decode',
      file: 'optional - filesystem path'
    )

    # Run parse line and return its result
    #{self}.parse_line(
      line: 'optional - line value consumed by #parse_line'
    )

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

.parse_frame(opts = {}) ⇒ Object

# File 'lib/pwn/sdr/decoder/wifi.rb', line 243

public_class_method def self.parse_frame(opts = {})
  bytes = opts[:bytes]
  return nil unless bytes.is_a?(String) && bytes.bytesize >= 14
  return nil unless Zlib.crc32(bytes[0...-4]) == bytes[-4, 4].unpack1('V')

  fc = bytes.unpack1('v')
  return nil unless fc.nobits?(3)

  type = (fc >> 2) & 3
  subtype = (fc >> 4) & 15
  return nil if type == 3

  control_lengths = { 10 => 16, 11 => 16, 12 => 10, 13 => 10, 14 => 16, 15 => 16 }
  length = type == 1 ? control_lengths[subtype] : 24
  return nil unless length && bytes.bytesize >= length + 4
  return nil if type == 1 && bytes.bytesize != length + 4

  address = ->(offset) { bytes[offset, 6].unpack('C*').map { |byte| format('%02x', byte) }.join(':') }
  frame = { protocol: 'WiFi', mode: :dsss_1mbps, source: 'mac', decoded: true,
            integrity: 'mac-fcs32', type: type, subtype: subtype, duration: bytes[2, 2].unpack1('v'),
            receiver: address.call(4), protected: fc.anybits?(0x4000), more_fragments: fc.anybits?(0x0400),
            retry: fc.anybits?(0x0800), frame_hex: bytes.unpack1('H*'),
            summary: "WiFi type=#{type} subtype=#{subtype}" }
  frame[:transmitter] = address.call(10) if length >= 16
  if type != 1
    ds = (fc >> 8) & 3
    return nil if type.zero? && ds != 0

    sequence = bytes[22, 2].unpack1('v')
    frame.merge!(sequence: sequence >> 4, fragment: sequence & 15, to_ds: ds.anybits?(1), from_ds: ds.anybits?(2))
    case ds
    when 0 then frame.merge!(destination: address.call(4), source_address: address.call(10), bssid: address.call(16))
    when 1 then frame.merge!(destination: address.call(16), source_address: address.call(10), bssid: address.call(4))
    when 2 then frame.merge!(destination: address.call(4), source_address: address.call(16), bssid: address.call(10))
    when 3
      length += 6
      return nil if bytes.bytesize < length + 4

      frame.merge!(destination: address.call(16), source_address: address.call(24))
    end
    if type == 2 && subtype >= 8
      return nil if bytes.bytesize < length + 6

      frame[:qos_control] = bytes[length, 2].unpack1('v')
      length += 2
      length += 4 if fc.anybits?(0x8000)
    end
  end
  return nil if bytes.bytesize < length + 4

  payload = bytes[length...-4]
  frame[:payload_hex] = payload.unpack1('H*')
  unless frame[:protected]
    if type.zero? && [4, 5, 8].include?(subtype)
      fixed = subtype == 4 ? 0 : 12
      return nil if payload.bytesize < fixed

      frame[:timestamp], frame[:beacon_interval], frame[:capabilities] = payload.unpack('Q<vv') if fixed.positive?
      elements = []
      while fixed < payload.bytesize
        return nil if fixed + 2 > payload.bytesize

        id, size = payload[fixed, 2].unpack('CC')
        return nil if fixed + 2 + size > payload.bytesize

        value = payload[fixed + 2, size]
        return nil if (id.zero? && size > 32) || (id == 3 && size != 1)

        elements << { id: id, hex: value.unpack1('H*') }
        frame[:ssid] = value.dup.force_encoding('UTF-8').scrub if id.zero?
        frame[:channel] = value.getbyte(0) if id == 3
        fixed += 2 + size
      end
      frame[:information_elements] = elements
    elsif type == 2 && payload.bytesize >= 8 && payload.start_with?([0xaa, 0xaa, 3, 0, 0, 0].pack('C*')) &&
          frame[:fragment].zero? && fc.nobits?(0x0400) && frame.fetch(:qos_control, 0).nobits?(0x80)
      frame[:ethertype] = payload[6, 2].unpack1('n')
      frame[:network_payload_hex] = payload[8..].unpack1('H*')
    end
  end
  frame
end

.parse_line(opts = {}) ⇒ Object

# File 'lib/pwn/sdr/decoder/wifi.rb', line 387

public_class_method def self.parse_line(opts = {})
  f = opts[:line].to_s.split('|', -1)
  out = {
    protocol: 'WiFi', subtype: f[1], bssid: f[2], sa: f[3], da: f[4],
    channel: f[5], rssi: f[6], ssid: f[7]
  }.reject { |_, v| v.to_s.empty? }
  out[:summary] = "WiFi ch=#{out[:channel]} BSSID=#{out[:bssid]} SSID=#{out[:ssid]} RSSI=#{out[:rssi]}"
  out
end

.plcp_crc(opts = {}) ⇒ Object

IEEE 802.11b 18.2.3.6, reflected CCITT, complemented remainder.

# File 'lib/pwn/sdr/decoder/wifi.rb', line 373

public_class_method def self.plcp_crc(opts = {})
  crc = 0xffff
  opts[:bytes].each do |byte|
    crc ^= byte
    8.times { crc = (crc >> 1) ^ (crc.odd? ? 0x8408 : 0) }
  end
  crc ^ 0xffff
end