Merge branch 'merge_branch' (develop v0.3.0)

README.md

@@ -12,25 +12,17 @@ Test automation, traffic generation, ECU development support or just **_for fun_
 In order to configure VLAN (IEEE 802.1q) tagging in your linux machine, Ubuntu's wiki is a good reference : https://wiki.ubuntu.com/vlan.
 
 ### 2.1 Interface configuration (Linux)
-During our testing, we simply used to USB-Ethernet adaptors with the following _/etc/network/interfaces_ configuration, although multi-NIC is not strictly required to fiddle with SOME/IP-SD.
-```
-# VLAN eth1
-auto eth1.10
-iface eth1.1 inet static
-  address 192.168.10.1
-  netmask 255.255.255.0
-
-# VLAN eth2
-auto eth2.10
-iface eth2.1 inet static
-  address 192.168.10.2
-  netmask 255.255.255.0
-```
-In order to bring them to life:
-```
-$sudo ifup eth1.10
-$sudo ifup eth2.10
-```
+Feel free to choose your preferred network topology in order to start fiddling with SOME/IP-SD. In our case, we opted for a couple of USB-Ethernet adaptors but that's not strictly necessary.
+
+Just keep in mind that these conventions are used from the _example collection_:
+- ETH_IFACE_A (normally acting as _sender_)
+  - iface name : eth1.10
+  - iface addr : 192.168.10.2
+  - iface port : 30490
+- ETH_IFACE_B (normally acting as _receiver_)
+  - iface name : eth2.10
+  - iface addr : 192.168.10.3
+  - iface port : 30490
 
 ## 3. Examples
 This folder contains a (hopefully growing) examples collection, build upon unittest package just for convenience. Just fire Wireshark up and enjoy analyzing generated traffic.

eth_scapy_someip/__init__.py

@@ -1,7 +1,7 @@
-import eth_scapy_someip
-import eth_scapy_sd
+from .eth_scapy_someip import SOMEIP
+from .eth_scapy_sd import SD
 
 from scapy.packet import bind_layers
 
 #  Layer binding
-bind_layers(eth_scapy_someip.SOMEIP,eth_scapy_sd.SD)
+bind_layers(SOMEIP,SD)

eth_scapy_someip/eth_scapy_sd.py

@@ -4,7 +4,7 @@
 from scapy.layers.inet6 import IP6Field
 import ctypes
 import collections
-from eth_scapy_someip import SOMEIP
+from .eth_scapy_someip import SOMEIP
 
 
 class _SDPacketBase(Packet):
@@ -66,7 +66,7 @@ class _SDEntry(_SDPacketBase):
 
     def guess_payload_class(self, payload):
         """ decode SDEntry depending on its type."""
-        pl_type = struct.unpack(_SDEntry.TYPE_FMT, payload[_SDEntry.TYPE_PAYLOAD_I])[0]
+        pl_type = struct.unpack(_SDEntry.TYPE_FMT, payload[_SDEntry.TYPE_PAYLOAD_I:_SDEntry.TYPE_PAYLOAD_I+1])[0]
         if pl_type in _SDEntry.TYPE_SRV:
             return SDEntry_Service
         elif pl_type in _SDEntry.TYPE_EVTGRP:
@@ -106,6 +106,9 @@ class SDEntry_EventGroup(_SDEntry):
 #  - IPv6 EndPoint
 class _SDOption(_SDPacketBase):
     """ Base class for SDOption_* packages."""
+    TYPE_FMT = ">B"
+    TYPE_PAYLOAD_I = 2
+
     CFG_TYPE = 0x01
     CFG_OVERALL_LEN = 4       # overall length of CFG SDOption,empty 'cfg_str' (to be used from UT)
     LOADBALANCE_TYPE = 0x02
@@ -128,7 +131,7 @@ class _SDOption(_SDPacketBase):
 
     def guess_payload_class(self, payload):
         """ decode SDOption depending on its type."""
-        pl_type = struct.unpack(">B", payload[2])[0]
+        pl_type = struct.unpack(_SDOption.TYPE_FMT, payload[_SDOption.TYPE_PAYLOAD_I:_SDOption.TYPE_PAYLOAD_I+1])[0]
 
         if pl_type == _SDOption.CFG_TYPE:
             return SDOption_Config

eth_scapy_someip/eth_scapy_someip.py

@@ -136,6 +136,6 @@ def post_build(self, p, pay):
         return p + pay
 
 
-for i in xrange(15):
+for i in range(15):
     bind_layers(UDP, SOMEIP, sport=30490 + i)
     bind_layers(TCP, SOMEIP, sport=30490 + i)

examples/ex_01_someip.py

@@ -6,22 +6,22 @@
 from eth_scapy_someip import eth_scapy_sd as sd
 
 iface = namedtuple('iface','name ip port')
-ETH_IFACE_A = iface(name='eth1.10', ip='192.168.10.1', port=30490)
-ETH_IFACE_B = iface(name='eth2.10', ip='192.168.10.2', port=30490)
+ETH_IFACE_A = iface(name='eth1.10', ip='192.168.10.2', port=30490)
+ETH_IFACE_B = iface(name='eth2.10', ip='192.168.10.3', port=30490)
 
 class ex_01_someip(unittest.TestCase):
   def setUp(self):
     pass
   def tearDown(self):
     pass
 
-  def _test_00(self):
+  def test_00(self):
     """ SOME/IP magic cookie (client >> server). TR_SOMEIP_00159."""
     # build SOME/IP packet
     sip = someip.SOMEIP()
     sip.msg_id.srv_id = 0xffff
     sip.msg_id.sub_id = 0x0
-    sip.msg_id.mtd_id = 0x0000
+    sip.msg_id.method_id = 0x0000
 
     sip.req_id.client_id = 0xdead
     sip.req_id.session_id = 0xbeef

examples/ex_02_sd.py

@@ -7,8 +7,8 @@
 from eth_scapy_someip import eth_scapy_sd as sd
 
 iface = namedtuple('iface','name ip port')
-ETH_IFACE_A = iface(name='eth1.10', ip='192.168.10.1', port=30490)
-ETH_IFACE_B = iface(name='eth2.10', ip='192.168.10.2', port=30490)
+ETH_IFACE_A = iface(name='eth1.10', ip='192.168.10.2', port=30490)
+ETH_IFACE_B = iface(name='eth2.10', ip='192.168.10.3', port=30490)
 
 class ex_02_sd(unittest.TestCase):
   def setUp(self):

requirements.txt

@@ -1,4 +1,6 @@
-scapy==2.4.0
+pytest
+pytest-cov
+scapy>=2.4.1
 wheel==0.24.0
 junit-xml==1.8
 coverage

setup.py

@@ -2,7 +2,7 @@
 
 setup(
   name="eth_scapy_someip",
-  version="0.2.0",
+  version="0.3.0",
   author="Jose Amores",
   description="Automotive Ethernet SOME/IP-SD Scapy protocol",
   license="MIT",

tests/test_sd.py

@@ -0,0 +1,246 @@
+import binascii
+import struct
+import sys
+import os
+import pytest
+
+from eth_scapy_someip import eth_scapy_someip as someip
+from eth_scapy_someip import eth_scapy_sd as sd
+
+HERE = os.path.dirname(os.path.realpath(__file__))
+
+def test_00_SDEntry_Service():
+    p = sd.SDEntry_Service()
+
+    # packet length
+    assert(len(binascii.hexlify(bytes(p)))/2 == sd._SDEntry.OVERALL_LEN)
+
+    # fields' setting
+    p.type = sd._SDEntry.TYPE_SRV_OFFERSERVICE
+    p.index_1 = 0x11
+    p.index_2 = 0x22
+    p.srv_id = 0x3333
+    p.inst_id = 0x4444
+    p.major_ver = 0x55
+    p.ttl = 0x666666
+    p.minor_ver = 0xdeadbeef
+
+    p_str = binascii.hexlify(bytes(p))
+    bin_str = b'011122003333444455666666deadbeef'
+    assert(p_str == bin_str)
+
+    # fields' setting : N_OPT
+    # value above 4 bits, serialized packet should feature  0x1 and 0x2
+    del(p)
+    p = sd.SDEntry_Service()
+    p.n_opt_1 = 0xf1 
+    p.n_opt_2 = 0xf2
+    p_str = binascii.hexlify(bytes(p))
+    bin_str = b'00'*3+b'12'+b'00'*12
+    assert(p_str == bin_str)
+    assert(len(p_str)/2 == sd._SDEntry.OVERALL_LEN)
+
+    # Payload guess
+    p_entry = sd._SDEntry()
+    p_entry_srv = sd.SDEntry_Service()
+
+    assert(p_entry.guess_payload_class(bytes(p_entry_srv)) == sd.SDEntry_Service)
+
+
+def test_01_SDEntry_EventGroup():
+    p = sd.SDEntry_EventGroup()
+
+    # packet length
+    assert(len(binascii.hexlify(bytes(p)))/2 == sd._SDEntry.OVERALL_LEN)
+
+    # fields' setting
+    p.index_1 = 0x11
+    p.index_2 = 0x22
+    p.srv_id = 0x3333
+    p.inst_id = 0x4444
+    p.major_ver = 0x55
+    p.ttl = 0x666666
+    p.cnt = 0x7
+    p.eventgroup_id = 0x8888
+
+    p_str = binascii.hexlify(bytes(p))
+    bin_str = b'06112200333344445566666600078888'
+    assert(p_str == bin_str)
+
+    # Payload guess
+    p_entry = sd._SDEntry()
+    p_entry_evtgrp = sd.SDEntry_EventGroup()
+
+    assert(p_entry.guess_payload_class(bytes(p_entry_evtgrp)) == sd.SDEntry_EventGroup)
+
+def test_02_SDOption_Config():
+    p = sd.SDOption_Config()
+
+    # pkg type
+    assert(p.type == sd._SDOption.CFG_TYPE)
+    # length without payload
+    assert(len(binascii.hexlify(bytes(p)))/2 == sd._SDOption.CFG_OVERALL_LEN)
+
+    # add payload and check length
+    p.cfg_str = "5abc=x7def=1230"
+    assert(binascii.hexlify(bytes(p)) == b'00100100'+binascii.hexlify(b'5abc=x7def=1230'))
+
+    # Payload guess
+    p_option = sd._SDOption()
+    assert(p_option.guess_payload_class(bytes(p)) == sd.SDOption_Config)
+
+def test_03_SDOption_LoadBalance():
+    p = sd.SDOption_LoadBalance()
+
+    # pkg type & lengths (static and overall)
+    assert(p.type == sd._SDOption.LOADBALANCE_TYPE)
+    assert(p.len == sd._SDOption.LOADBALANCE_LEN)
+    assert(len(binascii.hexlify(bytes(p)))/2 == sd._SDOption.LOADBALANCE_OVERALL_LEN)
+
+    # Payload guess
+    p_option = sd._SDOption()
+    assert(p_option.guess_payload_class(bytes(p)) == sd.SDOption_LoadBalance)
+
+def test_04_SDOption_IP4_EndPoint():
+    p = sd.SDOption_IP4_EndPoint()
+
+    # pkg type & length
+    assert(p.type == sd._SDOption.IP4_ENDPOINT_TYPE)
+    assert(p.len == sd._SDOption.IP4_ENDPOINT_LEN)
+
+    # Payload guess
+    p_option = sd._SDOption()
+    assert(p_option.guess_payload_class(bytes(p)) == sd.SDOption_IP4_EndPoint)
+
+def test_05_SDOption_IP4_Multicast():
+    p = sd.SDOption_IP4_Multicast()
+
+    # pkg type & length
+    assert(p.type == sd._SDOption.IP4_MCAST_TYPE)
+    assert(p.len == sd._SDOption.IP4_MCAST_LEN)
+
+    # Payload guess
+    p_option = sd._SDOption()
+    assert(p_option.guess_payload_class(bytes(p)) == sd.SDOption_IP4_Multicast)
+
+def test_06_SDOption_IP4_SD_EndPoint():
+    p = sd.SDOption_IP4_SD_EndPoint()
+
+    # pkg type & length
+    assert(p.type == sd._SDOption.IP4_SDENDPOINT_TYPE)
+    assert(p.len == sd._SDOption.IP4_SDENDPOINT_LEN)
+
+    # Payload guess
+    p_option = sd._SDOption()
+    assert(p_option.guess_payload_class(bytes(p)) == sd.SDOption_IP4_SD_EndPoint)
+
+def test_07_SDOption_IP6_EndPoint():
+    p = sd.SDOption_IP6_EndPoint()
+
+    # pkg type & length
+    assert(p.type == sd._SDOption.IP6_ENDPOINT_TYPE)
+    assert(p.len == sd._SDOption.IP6_ENDPOINT_LEN)
+
+    # Payload guess
+    p_option = sd._SDOption()
+    assert(p_option.guess_payload_class(bytes(p)) == sd.SDOption_IP6_EndPoint)
+
+def test_08_SDOption_IP6_Multicast():
+    p = sd.SDOption_IP6_Multicast()
+
+    # pkg type & length
+    assert(p.type == sd._SDOption.IP6_MCAST_TYPE)
+    assert(p.len == sd._SDOption.IP6_MCAST_LEN)
+
+    # Payload guess
+    p_option = sd._SDOption()
+    assert(p_option.guess_payload_class(bytes(p)) == sd.SDOption_IP6_Multicast)
+
+def test_09_SDOption_IP6_SD_EndPoint():
+    p = sd.SDOption_IP6_SD_EndPoint()
+
+    # pkg type & length
+    assert(p.type == sd._SDOption.IP6_SDENDPOINT_TYPE)
+    assert(p.len == sd._SDOption.IP6_SDENDPOINT_LEN)
+
+    # Payload guess
+    p_option = sd._SDOption()
+    assert(p_option.guess_payload_class(bytes(p)) == sd.SDOption_IP6_SD_EndPoint)
+
+def test_0a_SD_Flags():
+    p = sd.SD()
+
+    p.setFlag("REBOOT",1)
+    assert(p.flags == 0x80)
+    assert(p.getFlag("REBOOT") == 1)
+    p.setFlag("REBOOT",0)
+    assert(p.flags == 0x00)
+    assert(p.getFlag("REBOOT") == 0)
+    p.setFlag("UNICAST",1)
+    assert(p.flags == 0x40)
+    assert(p.getFlag("UNICAST") == 1)
+    p.setFlag("UNICAST",0)
+    assert(p.flags == 0x00)
+    assert(p.getFlag("UNICAST") == 0)
+
+    p.setFlag("REBOOT",1)
+    p.setFlag("UNICAST",1)
+    assert(p.flags == 0xc0)
+    assert(p.getFlag("REBOOT") == 1)
+    assert(p.getFlag("UNICAST") == 1)
+
+    # non-existing Flag
+    assert(p.getFlag('NON_EXISTING_FLAG') == None)
+
+def test_0b_SD_GetSomeipPacket():
+    p_sd = sd.SD()
+
+    sd_len = binascii.hexlify(bytes(p_sd))
+
+    p_someip = p_sd.getSomeip()
+    assert(len(binascii.hexlify(bytes(p_someip)))/2, someip.SOMEIP._OVERALL_LEN_NOPAYLOAD)
+
+    p = p_sd.getSomeip(stacked=True)
+    assert(len(binascii.hexlify(bytes(p)))/2, someip.SOMEIP._OVERALL_LEN_NOPAYLOAD + 12)
+
+
+def test_0c_SD():
+    p = sd.SD()
+
+    # length of package without entries nor options
+    assert(len(binascii.hexlify(bytes(p)))/2 == 12)
+
+    # some Entries to array and size check
+    p.setEntryArray([sd.SDEntry_Service(),sd.SDEntry_EventGroup()])
+    assert(struct.unpack("!L",bytes(p)[4:8])[0] == 32)
+    # make sure individual entry added as list
+    p.setEntryArray(sd.SDEntry_Service())
+    assert(isinstance(p.entry_array,list))
+    assert(len(p.entry_array) == 1)
+    # empty entry array
+    p.setEntryArray([])
+    assert(struct.unpack("!L",bytes(p)[4:8])[0] == 0)
+
+
+    # some Options to array and size check
+    p.setOptionArray([sd.SDOption_IP4_EndPoint(),sd.SDOption_IP4_EndPoint()])
+    assert(struct.unpack("!L",bytes(p)[8:12])[0] == 24)
+    # make sure individual option added as list
+    p.setOptionArray(sd.SDOption_IP4_EndPoint())
+    assert(isinstance(p.option_array,list))
+    assert(len(p.option_array) == 1)
+    # empty option array
+    p.setOptionArray([])
+    assert(struct.unpack("!L",bytes(p)[8:12])[0] == 0)
+
+    # some Entries&Options to array and size check
+    p.setEntryArray([sd.SDEntry_Service(),sd.SDEntry_EventGroup()])
+    p.setOptionArray([sd.SDOption_IP4_EndPoint(),sd.SDOption_IP4_EndPoint()])
+    assert(struct.unpack("!L",bytes(p)[4:8])[0] == 32)
+    assert(struct.unpack("!L",bytes(p)[40:44])[0] == 24)
+
+class _SDOption_IP4_EndPoint_defaults(sd._SDOption_IP4):
+    name = "IP4 Endpoint Option (UT)"
+    _defaults = {'non_existing_key':'does_not_matter_value'}
+def test_0d_defaults():
+    p = _SDOption_IP4_EndPoint_defaults()