Real-time collaboration for Jupyter Notebooks, Linux Terminals, LaTeX, VS Code, R IDE, and more,
all in one place.

1
#
2
# -*- coding: binary -*-
3
require 'rex/post/hwbridge/extensions/automotive/uds_errors'
4
require 'rex/post/hwbridge/client'
5

6
module Rex
7
module Post
8
module HWBridge
9
module Extensions
10
module Automotive
11

12
###
13
# Automotive extension - set of commands to be executed on automotive hw bridges
14
###
15

16
class Automotive < Extension
17

18
  include Rex::Post::HWBridge::Extensions::Automotive::UDSErrors
19

20
  def initialize(client)
21
    super(client, 'automotive')
22

23
    # Alias the following things on the client object so that they
24
    # can be directly referenced
25
    client.register_extension_aliases(
26
      [
27
        {
28
          'name' => 'automotive',
29
          'ext'  => self
30
        }
31
      ])
32
  end
33

34
  #
35
  # Checks to see if the specified bus is valid
36
  #
37
  # @param bus [String] bus name
38
  #
39
  # @return [Boolean] return true if bus is valid
40
  def is_valid_bus?(bus)
41
    valid = false
42
    get_supported_buses if buses.nil?
43
    unless bus.blank?
44
      self.buses.each do |b|
45
        valid = true if b["bus_name"] == bus
46
      end
47
    end
48
    valid
49
  end
50

51
  # Checks for Errors in ISO-TP responses.  If an error is present
52
  # Document the error with an additional "error" => { "ACRONYMN" => "Definition" }
53
  #
54
  # @param data [Hash] client.send_request response
55
  #
56
  # @return [Hash] client.send_request response with "Error" if any exist
57
  def check_for_errors(data)
58
    if data && (data.key? "Packets")
59
      if data["Packets"].size == 1
60
        if data["Packets"][0]["DATA"].size > 3 && data["Packets"][0]["DATA"][1].hex == 0x7F
61
          if ERR_MNEMONIC.key? data["Packets"][0]["DATA"][3].hex
62
            err = data["Packets"][0]["DATA"][3].hex
63
            data["error"] = { ERR_MNEMONIC[err] => ERR_DESC[ERR_MNEMONIC[err]] }
64
          else
65
            data["error"] = { "UNK" => "An Unknown error detected" }
66
          end
67
        end
68
      end
69
    end
70
    data
71
  end
72

73
  #
74
  # Pass an array of bytes and return an array of ASCII byte representation
75
  #
76
  # @param arr [Array] Array of integers (bytes)
77
  #
78
  # @return [Array] Array of Hex string equivalents
79
  def array2hex(arr)
80
    # We give the flexibility of sending Integers or string hexes in the array
81
    arr.map { |b| "%02x" % (b.respond_to?("hex") ? b.hex : b ) }
82
  end
83

84
  #
85
  # Pad the end of a packet with a set byte until it is 8 bytes long
86
  #
87
  # @param data [Array] Packet to padd
88
  # @param padding [Integer] Expected single byte 0x00 style argument
89
  # @return [Array] Packet as data
90
  def padd_packet(data, padding)
91
    return data if padding.nil?
92
    return data if data.size > 7
93
    data + [ padding ] * (8 - data.size)
94
  end
95

96
  def set_active_bus(bus)
97
    self.active_bus = bus
98
  end
99

100
  def get_supported_buses
101
    self.buses = client.send_request("/automotive/supported_buses")
102
    self.buses
103
  end
104

105
  def get_bus_config(bus)
106
    status = client.send_request("/automotive/#{bus}/config")
107
    status
108
  end
109

110
  def get_supported_methods(bus)
111
    client.send_request("/automotive/#{bus}/supported_methods")
112
  end
113

114
  def cansend(bus, id, data)
115
    client.send_request("/automotive/#{bus}/cansend?id=#{id}&data=#{data}")
116
  end
117

118
  def send_isotp_and_wait_for_response(bus, src_id, dst_id, data, opt = {})
119
    # TODO: Implement sending ISO-TP > 8 bytes
120
    data = [ data ] if data.is_a? Integer
121
    if data.size < 8
122
      # Padding is handled differently after 0.0.3
123
      if Rex::Version.new(client.api_version) < Rex::Version.new('0.0.4')
124
        data = padd_packet(data, opt['PADDING']) if opt.key? 'PADDING'
125
      end
126
      data = array2hex(data).join
127
      request_str = "/automotive/#{bus}/isotpsend_and_wait?srcid=#{src_id}&dstid=#{dst_id}&data=#{data}"
128
      request_str += "&timeout=#{opt['TIMEOUT']}" if opt.key? "TIMEOUT"
129
      request_str += "&maxpkts=#{opt['MAXPKTS']}" if opt.key? "MAXPKTS"
130
      request_str += "&padding=#{opt['PADDING']}" if opt.key? "PADDING" # Won't hurt to use in older versions
131
      request_str += "&fc=#{opt['FC']}" if opt.key? "FC" # Force flow control
132
      return check_for_errors(client.send_request(request_str))
133
    end
134
    nil
135
  end
136

137
  attr_reader :buses, :active_bus
138

139
  private
140

141
  attr_writer :buses, :active_bus
142

143
end
144

145
end
146
end
147
end
148
end
149
end
150

151