| % Regression tests for the DoIP layer |
| |
| # More information at http://www.secdev.org/projects/UTscapy/ |
| |
| |
| ############ |
| ############ |
| |
| + Doip contrib tests |
| |
| = Load Contrib Layer |
| |
| load_contrib("automotive.doip", globals_dict=globals()) |
| load_contrib("automotive.uds", globals_dict=globals()) |
| |
| = Defaults test |
| |
| p = DoIP(payload_type=1) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == None |
| assert p.payload_type == 1 |
| |
| = Build test 0 |
| |
| p = DoIP(bytes(DoIP(payload_type=0))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 1 |
| assert p.payload_type == 0 |
| assert p.nack == 0 |
| |
| = Build test 1 |
| |
| p = DoIP(bytes(DoIP(payload_type=1))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 0 |
| assert p.payload_type == 1 |
| |
| = Build test 2 |
| |
| p = DoIP(bytes(DoIP(payload_type=2))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 6 |
| assert p.payload_type == 2 |
| assert bytes(p.eid) == b"\x00" * 6 |
| |
| = Build test 3 |
| |
| p = DoIP(bytes(DoIP(payload_type=3))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 17 |
| assert p.payload_type == 3 |
| assert bytes(p.vin) == b"\x00" * 17 |
| |
| = Build test 4 |
| |
| p = DoIP(bytes(DoIP(payload_type=4))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 33 |
| assert p.payload_type == 4 |
| assert bytes(p.vin) == b"\x00" * 17 |
| assert p.logical_address == 0 |
| assert bytes(p.eid) == b"\x00" * 6 |
| assert bytes(p.gid) == b"\x00" * 6 |
| assert p.further_action == 0 |
| assert p.vin_gid_status == 0 |
| |
| = Build test 5 |
| |
| p = DoIP(bytes(DoIP(payload_type=5))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 7 |
| assert p.payload_type == 5 |
| assert p.source_address == 0 |
| assert p.activation_type == 0 |
| assert p.reserved_iso == 0 |
| assert p.reserved_oem == b"" |
| |
| = Build test 5.1 |
| |
| p = DoIP(bytes(DoIP(payload_type=5, reserved_oem=b"1234"))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 11 |
| assert p.payload_type == 5 |
| assert p.source_address == 0 |
| assert p.activation_type == 0 |
| assert p.reserved_iso == 0 |
| p.show() |
| print(p.reserved_oem) |
| assert p.reserved_oem == b"1234" |
| |
| = Build test 5.2 |
| |
| p = DoIP(bytes(DoIP(payload_type=5, reserved_oem=b"12"))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 9 |
| assert p.payload_type == 5 |
| assert p.source_address == 0 |
| assert p.activation_type == 0 |
| assert p.reserved_iso == 0 |
| assert p.reserved_oem == b"12" |
| |
| = Build test 6 |
| |
| p = DoIP(bytes(DoIP(payload_type=6))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 9 |
| assert p.payload_type == 6 |
| assert p.logical_address_tester == 0 |
| assert p.logical_address_doip_entity == 0 |
| assert p.reserved_iso == 0 |
| assert p.reserved_oem == b"" |
| |
| = Build test 7 |
| |
| p = DoIP(bytes(DoIP(payload_type=7))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 0 |
| assert p.payload_type == 7 |
| |
| = Build test 8 |
| |
| p = DoIP(bytes(DoIP(payload_type=8))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 2 |
| assert p.payload_type == 8 |
| assert p.source_address == 0 |
| |
| = Build test 4001 |
| |
| p = DoIP(bytes(DoIP(payload_type=0x4001))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 0 |
| assert p.payload_type == 0x4001 |
| |
| |
| = Build test 4002 |
| |
| p = DoIP(bytes(DoIP(payload_type=0x4002))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 7 |
| assert p.payload_type == 0x4002 |
| assert p.node_type == 0 |
| assert p.max_open_sockets == 1 |
| assert p.cur_open_sockets == 0 |
| assert p.max_data_size == 0 |
| |
| |
| = Build test 4003 |
| |
| p = DoIP(bytes(DoIP(payload_type=0x4003))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 0 |
| assert p.payload_type == 0x4003 |
| |
| |
| = Build test 4004 |
| |
| p = DoIP(bytes(DoIP(payload_type=0x4004))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 1 |
| assert p.payload_type == 0x4004 |
| assert p.diagnostic_power_mode == 0 |
| |
| = Build test 8001 |
| |
| p = DoIP(bytes(DoIP(payload_type=0x8001))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 4 |
| assert p.payload_type == 0x8001 |
| assert p.source_address == 0 |
| assert p.target_address == 0 |
| |
| = Build test 8002 |
| |
| p = DoIP(bytes(DoIP(payload_type=0x8002))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 5 |
| assert p.payload_type == 0x8002 |
| assert p.source_address == 0 |
| assert p.target_address == 0 |
| assert p.ack_code == 0 |
| assert p.previous_msg == b'' |
| |
| p = DoIP(bytes(DoIP(payload_type=0x8002, previous_msg=b'\x22\xfd\x32'))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 8 |
| assert p.payload_type == 0x8002 |
| assert p.source_address == 0 |
| assert p.target_address == 0 |
| assert p.ack_code == 0 |
| assert p.previous_msg == b'\x22\xfd\x32' |
| |
| p = DoIP(bytes(DoIP(payload_type=0x8002, previous_msg=b'\x19\x02\x09\x9C\x00'))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 10 |
| assert p.payload_type == 0x8002 |
| assert p.source_address == 0 |
| assert p.target_address == 0 |
| assert p.ack_code == 0 |
| assert p.previous_msg == b'\x19\x02\t\x9c\x00' |
| |
| p = DoIP(b'\x02\xfd\x80\x02\x00\x00\x00\x07\x00\x08\x00\x0e\x00\x10\x01') |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xFD |
| assert p.payload_length == 7 |
| assert p.payload_type == 0x8002 |
| assert p.source_address == 0x8 |
| assert p.target_address == 0xE |
| assert p.ack_code == 0 |
| assert p.previous_msg == b'\x10\x01' |
| |
| = Build test 8003 |
| |
| p = DoIP(bytes(DoIP(payload_type=0x8003))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 5 |
| assert p.payload_type == 0x8003 |
| assert p.source_address == 0 |
| assert p.target_address == 0 |
| assert p.nack_code == 0 |
| |
| |
| p = DoIP(bytes(DoIP(payload_type=0x8003, previous_msg=b'\x2E\xfd\x32\x01\x02'))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 10 |
| assert p.payload_type == 0x8003 |
| assert p.source_address == 0 |
| assert p.target_address == 0 |
| assert p.nack_code == 0 |
| assert p.previous_msg == b'.\xfd2\x01\x02' |
| |
| p = DoIP(bytes(DoIP(payload_type=0x8003, previous_msg=b'\x19\x02\x09\x9A\x00'))) |
| |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 10 |
| assert p.payload_type == 0x8003 |
| assert p.source_address == 0 |
| assert p.target_address == 0 |
| assert p.nack_code == 0 |
| assert p.previous_msg == b'\x19\x02\t\x9a\x00' |
| |
| p = DoIP(b'\x02\xfd\x80\x03\x00\x00\x00\x07\x00\x0A\x00\x0C\x00\x10\x03') |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xFD |
| assert p.payload_length == 7 |
| assert p.payload_type == 0x8003 |
| assert p.source_address == 0xA |
| assert p.target_address == 0xC |
| assert p.nack_code == 0 |
| assert p.previous_msg == b'\x10\x03' |
| |
| + pcap based tests |
| |
| = read diag_ack pcap file |
| pkt = rdpcap(scapy_path("test/pcaps/doip_ack.pcap")).res[0] |
| |
| assert len(pkt) == 70 |
| |
| = dissect test of diag ACK with previous_msg field filled |
| assert pkt.protocol_version == 0x02 |
| assert pkt.inverse_version == 0xFD |
| assert pkt.payload_length == 8 |
| assert pkt.source_address == 0x4B |
| assert pkt.target_address == 0xE00 |
| assert pkt.ack_code == 0 |
| assert pkt.previous_msg == b'\x22\xFD\x31' |
| |
| |
| = read main pcap file |
| |
| pkts = rdpcap(scapy_path("test/pcaps/doip.pcap.gz")) |
| ips = [p for p in pkts if p.proto == 6] |
| |
| assert len(ips) > 1 |
| |
| = dissect test of routing activation pkts req |
| |
| req = ips[0] |
| p = req |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 11 |
| assert p.payload_type == 0x5 |
| assert p.source_address == 0xe80 |
| assert p.activation_type == 0 |
| assert p.reserved_iso == 0 |
| assert p.reserved_oem == b"\x00\x00\x00\x00" |
| |
| = dissect test of routing activation pkts resp |
| |
| resp = ips[1] |
| p = resp |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 9 |
| assert p.payload_type == 0x6 |
| assert p.logical_address_tester == 0xe80 |
| assert p.logical_address_doip_entity == 0x4010 |
| assert p.routing_activation_response == 16 |
| assert p.reserved_iso == 0 |
| |
| = answers test of routing activation pkts |
| |
| assert resp.answers(req) |
| assert resp.hashret() == req.hashret() |
| |
| = dissect diagnostic message |
| |
| req = ips[-4] |
| resp = ips[-1] |
| |
| p = req |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 6 |
| assert p.payload_type == 0x8001 |
| assert p.source_address == 0xe80 |
| assert p.target_address == 0x4010 |
| assert bytes(p)[-2:] == bytes(UDS()/UDS_DSC(b"\x02")) |
| assert p.service == 0x10 |
| assert p.diagnosticSessionType == 2 |
| |
| p = resp |
| assert p.protocol_version == 0x02 |
| assert p.inverse_version == 0xfd |
| assert p.payload_length == 10 |
| assert p.payload_type == 0x8001 |
| assert p.target_address == 0xe80 |
| assert p.source_address == 0x4010 |
| assert bytes(p)[-6:] == bytes(UDS()/UDS_DSCPR(b"\x02\x002\x01\xf4")) |
| assert p.service == 0x50 |
| assert p.diagnosticSessionType == 2 |
| |
| assert req.hashret() == resp.hashret() |
| # exclude TCP layer from answers check |
| assert resp[3].answers(req[3]) |
| assert not req[3].answers(resp[3]) |
| |
| = TCPSession Test |
| |
| tmp_file = get_temp_file() |
| |
| wrpcap(tmp_file, [ |
| IP(src="10.10.10.10", dst="10.10.10.11") / TCP(sport=61000, seq=1) / DoIP(payload_type=0x8001, payload_length=6) / b"\x3E", |
| IP(src="10.10.10.10", dst="10.10.10.11") / TCP(sport=61000, dport=13400, seq=14) / Raw(load=b"\xff") |
| ]) |
| |
| pkts = sniff(offline=tmp_file, session=TCPSession) |
| assert pkts[0].haslayer(UDS_TP) |
| assert pkts[0].service == 0x3E |
| |
| = TCPSession Test multiple DoIP messages |
| |
| filename = scapy_path("/test/pcaps/multiple_doip_layers.pcap.gz") |
| |
| pkts = sniff(offline=filename, session=TCPSession) |
| print(repr(pkts[0])) |
| print(repr(pkts[1])) |
| assert len(pkts) == 2 |
| assert pkts[0][DoIP].payload_length == 2 |
| assert pkts[0][DoIP:2].payload_length == 7 |
| assert pkts[1][DoIP].payload_length == 103 |
| |
| + DoIP Communication tests |
| |
| = Load libraries |
| import base64 |
| import ssl |
| import tempfile |
| |
| = Test DoIPSocket |
| |
| server_up = threading.Event() |
| sniff_up = threading.Event() |
| def server(): |
| buffer = b'\x02\xfd\x80\x02\x00\x00\x00\x05\x00\x00\x00\x00\x00\x02\xfd\x80\x01\x00\x00\x00\n\x10\x10\x0e\x80P\x03\x002\x01\xf4' |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| try: |
| sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| sock.bind(('127.0.0.1', 13400)) |
| sock.listen(1) |
| server_up.set() |
| connection, address = sock.accept() |
| sniff_up.wait(timeout=1) |
| connection.send(buffer) |
| connection.close() |
| finally: |
| sock.close() |
| |
| |
| server_thread = threading.Thread(target=server) |
| server_thread.start() |
| server_up.wait(timeout=1) |
| sock = DoIPSocket(activate_routing=False) |
| |
| pkts = sock.sniff(timeout=1, count=2, started_callback=sniff_up.set) |
| server_thread.join(timeout=1) |
| assert len(pkts) == 2 |
| |
| |
| = Test DoIPSocket 2 |
| ~ linux |
| |
| server_up = threading.Event() |
| sniff_up = threading.Event() |
| def server(): |
| buffer = b'\x02\xfd\x80\x02\x00\x00\x00\x05\x00\x00\x00\x00\x00\x02\xfd\x80\x01\x00\x00\x00\n\x10\x10\x0e\x80P\x03\x002\x01\xf4' |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| try: |
| sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| sock.bind(('127.0.0.1', 13400)) |
| sock.listen(1) |
| server_up.set() |
| connection, address = sock.accept() |
| sniff_up.wait(timeout=1) |
| for i in range(len(buffer)): |
| connection.send(buffer[i:i+1]) |
| time.sleep(0.01) |
| connection.close() |
| finally: |
| sock.close() |
| |
| |
| server_thread = threading.Thread(target=server) |
| server_thread.start() |
| server_up.wait(timeout=1) |
| sock = DoIPSocket(activate_routing=False) |
| |
| pkts = sock.sniff(timeout=1, count=2, started_callback=sniff_up.set) |
| server_thread.join(timeout=1) |
| assert len(pkts) == 2 |
| |
| = Test DoIPSocket 3 |
| |
| server_up = threading.Event() |
| sniff_up = threading.Event() |
| def server(): |
| buffer = b'\x02\xfd\x80\x02\x00\x00\x00\x05\x00\x00\x00\x00\x00\x02\xfd\x80\x01\x00\x00\x00\n\x10\x10\x0e\x80P\x03\x002\x01\xf4' |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| try: |
| sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| sock.bind(('127.0.0.1', 13400)) |
| sock.listen(1) |
| server_up.set() |
| connection, address = sock.accept() |
| sniff_up.wait(timeout=1) |
| while buffer: |
| randlen = random.randint(0, len(buffer)) |
| connection.send(buffer[:randlen]) |
| buffer = buffer[randlen:] |
| time.sleep(0.01) |
| connection.close() |
| finally: |
| sock.close() |
| |
| |
| server_thread = threading.Thread(target=server) |
| server_thread.start() |
| server_up.wait(timeout=1) |
| sock = DoIPSocket(activate_routing=False) |
| |
| pkts = sock.sniff(timeout=1, count=2, started_callback=sniff_up.set) |
| server_thread.join(timeout=1) |
| assert len(pkts) == 2 |
| |
| |
| = Test DoIPSocket6 |
| |
| server_up = threading.Event() |
| sniff_up = threading.Event() |
| def server(): |
| buffer = b'\x02\xfd\x80\x02\x00\x00\x00\x05\x00\x00\x00\x00\x00\x02\xfd\x80\x01\x00\x00\x00\n\x10\x10\x0e\x80P\x03\x002\x01\xf4' |
| sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) |
| try: |
| sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| sock.bind(('::1', 13400)) |
| sock.listen(1) |
| server_up.set() |
| connection, address = sock.accept() |
| sniff_up.wait(timeout=1) |
| connection.send(buffer) |
| connection.close() |
| finally: |
| sock.close() |
| |
| |
| server_thread = threading.Thread(target=server) |
| server_thread.start() |
| server_up.wait(timeout=1) |
| sock = DoIPSocket(ip="::1", activate_routing=False) |
| |
| pkts = sock.sniff(timeout=1, count=2, started_callback=sniff_up.set) |
| server_thread.join(timeout=1) |
| assert len(pkts) == 2 |
| |
| = Test DoIPSslSocket |
| ~ broken_windows |
| |
| certstring = """ |
| LS0tLS1CRUdJTiBQUklWQVRFIEtFWS0tLS0tCk1JSUV2QUlCQURBTkJna3Foa2lHOXcwQkFRRUZB |
| QVNDQktZd2dnU2lBZ0VBQW9JQkFRRFUvK0hRbVpzSDl2QVcKQ3ZMQjRxalpnZFJSSXE1b2JBanB4 |
| YUhoUGxCVEMvUlBzMHIxRVF0V0FtbXNEZFE3UGlLaCtYa1hES3pNY3lJSQp1a0ZpNThUQW1idGFj |
| N0U5VmJHSnNlTWp2RkJKSkFqQXVtbFdRZk5XcSs2TkZhdmRkTDQrSTNBTVJ5TldJTkJYCjhHMzRo |
| dldIbDdTOGhhSFFZN0FXcUZWVTNVL2xKR2pubnF3MEJraEIvVGRCTWIwM0habzkrVjIrWU9RZmk5 |
| QWsKTVRSRXpSeWVObWJqT0sxbHpXdFJXWkZZU0RnMEtqUVh4SkdFNVc5MzFPWitHL1NkbytTM1ZW |
| SVRPdWxQbHRmVwpXMEdjeCsvZERSNFIxNG5mcUl5L1daMElHUVNXMlRsQytmeGJ0dURDUkFqelRz |
| b0J3YjJ0cnpoR0VtYVFveUtNCnpBKzVSUHNyQWdNQkFBRUNnZ0VBRUJHaEoyWm5OVHh5YVY5TnZY |
| QjI1NDNZQnRUMGVSUHBhanJLMXg0bk1OU3oKNE9LNFVzWlo1MnBnTHRHT1EzZm1aS0l0cEo1WlY1 |
| cVBUejdwN3VjUzhnQWNZUnNJUnpCMHA5d3FpWExMK3h0RApxUjB4dnR4VDJpUGlFblVNNndudHpr |
| SHpKK0g0QkZLT2FvdjNaK3Fha2E1UmFCcmhheGRuaDBDNklLQmZtM3cyCm5zUWI2N0lCYWwrSnBs |
| L1g5TENWRkdRT2owb0lmVWI5ZFp3OWQ3MCthSGVVb2xvMGdYZmxxcXFFcnl3ZDlPN2QKNnp4dGlx |
| cnRyZUJhK1IraWs3NE1SK0xvaFNVR3o2VTRQaXhWQ3l1SnQ2U0hvRHR2L3dtSnltWDd2a0FRS2w1 |
| RQplK1JqUGVyakpUWTNzNXNXbEd2V21UTEtEbnVyS2pBYzZUOHhKb0pXWlFLQmdRRHdsd2RRdmww |
| S28wNHhDUmtiCklYRGVJZE1jZkp2ejRGZEtka1BmVnZVT2xHVEpNZkRzbWNoUzZhcEJCQUdQMUU2 |
| VkN2VzJmUFdjaXhScHE3MW8KR2xtbWZ5RnlJRW0rL08yamMvSFRXWHp6Qjdoc0JISEltQklHczFU |
| TC9iWFU3amhVQW5kWDdMK3RSRDBKNWRGVwpiN1VOOXNxaWdtRG42REJWZkxaUHgxRnlWUUtCZ1FE |
| aXBIT1BhNmVMSlk5R1FZdkw3OTIyTHNoU3ZYSUFVMERGCjBabTlqbjM2b3ZIY0kvWEZDdHVXank2 |
| WG9wbk9pbjlycmtUY2FDUnBvSEFNb00ycHdiR0tFY0dVVEY2RHQ3akYKRHVnd2srR21sbDkrbjM2 |
| M3Iwb09YNktSbWFhRStiZHoyNjNQVEhMaktYUnFyc3h5WEtMT3ZyTXhVNWNzMXJCeQpTMWI2ZGhr |
| M2Z3S0JnRjlONUliMnNkS3ArQ3B5aVRCM0ljZk1yRjBuZTN1ekRjRWdjaWlCd05lQ3J4NElHNEVP |
| Ck5nMnFKRmhXNXV0NzFaa3kyenpyNlR1VzJJSTNsdk1ySlFKUWNBWk9oZ2dURjJ2ZFhSazA1TXM4 |
| N3JCVFhtTncKNGdzbmROck42UDZ0VTBEc0xTeDJTME91dVdNM1Y2S2U0NkRoZDBuQ3pmSnZ4dDNH |
| WmszYURnaDFBb0dBWFhIcQpoNDZlZEx1V3VDUGNUTWhvUkc1RGdBSEdHQ1k3UlpTbTY4WHRZVUov |
| c0FGUG10OWdMRko2cG1DUFE5NU1yUXdjCkxqZnVFM0xuMy8wSTd0NENvbWV4eGNBN0U5blRIOFNH |
| clVpN3QrQzJITklNQUJZUTFaNU91L042K2Nhd0FkL28KYU5rZllWTzlRU015L2svOWZIcWFEVk5t |
| dUVFSVhRZDlKQ1UvUG1jQ2dZQWI0RTBRWTdDZmlrV293OFIzSlhoZgo0MHFVVkdud09QKzJNbXE5 |
| d2ZmWkpTRHNFSTQvb2g0VGRnN0sybHNNazVsWnRaMyszTjljSDVUc1pMYlJtd2FMCm9sRVl6K1BB |
| WU91MlMrY1l2bFlNL0V2WmlpRHJybjZuTStNbTNnaXJPYkNwMzcxd1ZxRFVsUnB4OUlwWVdYcnAK |
| T3YxUXFHdXkwODdyQkk1cStWL3hqQT09Ci0tLS0tRU5EIFBSSVZBVEUgS0VZLS0tLS0KLS0tLS1C |
| RUdJTiBDRVJUSUZJQ0FURS0tLS0tCk1JSUQ3VENDQXRXZ0F3SUJBZ0lVVTNsendsTVNSa294Tkdk |
| SFJzZllIcUtxcDAwd0RRWUpLb1pJaHZjTkFRRUwKQlFBd2dZVXhDekFKQmdOVkJBWVRBa1JGTVJN |
| d0VRWURWUVFJREFwVGIyMWxMVk4wWVhSbE1Rd3dDZ1lEVlFRSApEQU5TUlVjeEVUQVBCZ05WQkFv |
| TUNHUnBjM05sWTNSdk1Rd3dDZ1lEVlFRTERBTkVSVll4RFRBTEJnTlZCQU1NCkJGUkZVMVF4SXpB |
| aEJna3Foa2lHOXcwQkNRRVdGR052Ym5SaFkzUXRkWE5BWkdsemMyVmpMblJ2TUI0WERUSTAKTURN |
| eE9ERTVNek13TlZvWERUSTBNRFF4TnpFNU16TXdOVm93Z1lVeEN6QUpCZ05WQkFZVEFrUkZNUk13 |
| RVFZRApWUVFJREFwVGIyMWxMVk4wWVhSbE1Rd3dDZ1lEVlFRSERBTlNSVWN4RVRBUEJnTlZCQW9N |
| Q0dScGMzTmxZM1J2Ck1Rd3dDZ1lEVlFRTERBTkVSVll4RFRBTEJnTlZCQU1NQkZSRlUxUXhJekFo |
| QmdrcWhraUc5dzBCQ1FFV0ZHTnYKYm5SaFkzUXRkWE5BWkdsemMyVmpMblJ2TUlJQklqQU5CZ2tx |
| aGtpRzl3MEJBUUVGQUFPQ0FROEFNSUlCQ2dLQwpBUUVBMVAvaDBKbWJCL2J3RmdyeXdlS28yWUhV |
| VVNLdWFHd0k2Y1doNFQ1UVV3djBUN05LOVJFTFZnSnByQTNVCk96NGlvZmw1Rnd5c3pITWlDTHBC |
| WXVmRXdKbTdXbk94UFZXeGliSGpJN3hRU1NRSXdMcHBWa0h6VnF2dWpSV3IKM1hTK1BpTndERWNq |
| VmlEUVYvQnQrSWIxaDVlMHZJV2gwR093RnFoVlZOMVA1U1JvNTU2c05BWklRZjAzUVRHOQpOeDJh |
| UGZsZHZtRGtINHZRSkRFMFJNMGNualptNHppdFpjMXJVVm1SV0VnNE5DbzBGOFNSaE9WdmQ5VG1m |
| aHYwCm5hUGt0MVZTRXpycFQ1YlgxbHRCbk1mdjNRMGVFZGVKMzZpTXYxbWRDQmtFbHRrNVF2bjhX |
| N2Jnd2tRSTgwN0sKQWNHOXJhODRSaEpta0tNaWpNd1B1VVQ3S3dJREFRQUJvMU13VVRBZEJnTlZI |
| UTRFRmdRVVZhbUFkUjR1ZW8zQgpmV0RjUlMyUkQ3OEtlZXd3SHdZRFZSMGpCQmd3Rm9BVVZhbUFk |
| UjR1ZW8zQmZXRGNSUzJSRDc4S2Vld3dEd1lEClZSMFRBUUgvQkFVd0F3RUIvekFOQmdrcWhraUc5 |
| dzBCQVFzRkFBT0NBUUVBRjE1TTNvL3RyUVdYeHdHamlxZjgKNXBUTEM0bHJwQkZaTFZDbStQdHd4 |
| aENlN1ZSd2dLMElBb01EMW0vSjNEYnVJSjVURXlTVElnR2N0WHVNbG5pWgpsY3IwekZOZVVhQ08w |
| YkdhaExYUXpCWTRxSkhTTUNWNnhiNXNqUDlEdk9HYnFxbHVTbk51ZFJ5UWNIbkd4SE0rCk1adXpO |
| WUNseklOMEtYbFJuSTZqRXUrcG9XZ0pEMGN1NFM2b1lwT2R3bElRYmtaNnIrUE1jQ3hpRmhRd3E2 |
| em4KcE1nQzB0WlpSM3pCOEpVcTJwRHlGVy9jVlFjWkp5YUhnQkkwWlJWWG5wbDFqYng2YlNIOCts |
| cnMxVk1xZDlkcQozd1BMcjBheWI2VkpNa29WMjNWSXAzLzlYQVpTR3Z6Y0dadnM2VThSUTdFbUtx |
| akJibWxudm1CTkpUMk9xbFFRCllRPT0KLS0tLS1FTkQgQ0VSVElGSUNBVEUtLS0tLQo=""" |
| |
| certstring = certstring.replace('\n', '') |
| |
| def _load_certificate_chain(context) -> None: |
| with tempfile.NamedTemporaryFile(delete=False) as fp: |
| fp.write(base64.b64decode(certstring)) |
| fp.close() |
| context.load_cert_chain(fp.name) |
| |
| |
| server_up = threading.Event() |
| sniff_up = threading.Event() |
| def server(): |
| context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) |
| _load_certificate_chain(context) |
| context.check_hostname = False |
| context.verify_mode = ssl.CERT_NONE |
| buffer = b'\x02\xfd\x80\x02\x00\x00\x00\x05\x00\x00\x00\x00\x00\x02\xfd\x80\x01\x00\x00\x00\n\x10\x10\x0e\x80P\x03\x002\x01\xf4' |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| ssock = context.wrap_socket(sock) |
| try: |
| ssock.bind(('127.0.0.1', 3496)) |
| ssock.listen(1) |
| server_up.set() |
| connection, address = ssock.accept() |
| sniff_up.wait(timeout=1) |
| connection.send(buffer) |
| connection.close() |
| finally: |
| ssock.close() |
| |
| |
| server_thread = threading.Thread(target=server) |
| server_thread.start() |
| server_up.wait(timeout=1) |
| context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) |
| context.check_hostname = False |
| context.verify_mode = ssl.CERT_NONE |
| sock = DoIPSocket(activate_routing=False, force_tls=True, context=context) |
| |
| pkts = sock.sniff(timeout=1, count=2, started_callback=sniff_up.set) |
| server_thread.join(timeout=1) |
| assert len(pkts) == 2 |
| |
| = Test DoIPSslSocket6 |
| ~ broken_windows |
| |
| server_up = threading.Event() |
| sniff_up = threading.Event() |
| def server(): |
| context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) |
| _load_certificate_chain(context) |
| context.check_hostname = False |
| context.verify_mode = ssl.CERT_NONE |
| buffer = b'\x02\xfd\x80\x02\x00\x00\x00\x05\x00\x00\x00\x00\x00\x02\xfd\x80\x01\x00\x00\x00\n\x10\x10\x0e\x80P\x03\x002\x01\xf4' |
| sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) |
| sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| ssock = context.wrap_socket(sock) |
| try: |
| ssock.bind(('::1', 3496)) |
| ssock.listen(1) |
| server_up.set() |
| connection, address = ssock.accept() |
| sniff_up.wait(timeout=1) |
| connection.send(buffer) |
| connection.close() |
| finally: |
| ssock.close() |
| |
| |
| server_thread = threading.Thread(target=server) |
| server_thread.start() |
| server_up.wait(timeout=1) |
| context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) |
| context.check_hostname = False |
| context.verify_mode = ssl.CERT_NONE |
| sock = DoIPSocket(ip="::1", activate_routing=False, force_tls=True, context=context) |
| |
| pkts = sock.sniff(timeout=1, count=2, started_callback=sniff_up.set) |
| server_thread.join(timeout=1) |
| assert len(pkts) == 2 |
| |
| = Test UDS_DoIPSslSocket6 |
| ~ broken_windows |
| |
| server_up = threading.Event() |
| sniff_up = threading.Event() |
| def server(): |
| context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) |
| _load_certificate_chain(context) |
| context.check_hostname = False |
| context.verify_mode = ssl.CERT_NONE |
| buffer = b'\x02\xfd\x80\x02\x00\x00\x00\x05\x00\x00\x00\x00\x00\x02\xfd\x80\x01\x00\x00\x00\n\x10\x10\x0e\x80P\x03\x002\x01\xf4' |
| sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) |
| sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| ssock = context.wrap_socket(sock) |
| try: |
| ssock.bind(('::1', 3496)) |
| ssock.listen(1) |
| server_up.set() |
| connection, address = ssock.accept() |
| sniff_up.wait(timeout=1) |
| connection.send(buffer) |
| connection.close() |
| finally: |
| ssock.close() |
| |
| |
| server_thread = threading.Thread(target=server) |
| server_thread.start() |
| server_up.wait(timeout=1) |
| context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) |
| context.check_hostname = False |
| context.verify_mode = ssl.CERT_NONE |
| sock = UDS_DoIPSocket(ip="::1", activate_routing=False, force_tls=True, context=context) |
| |
| pkts = sock.sniff(timeout=1, count=2, started_callback=sniff_up.set) |
| server_thread.join(timeout=1) |
| assert len(pkts) == 2 |
| |
| = Test UDS_DualDoIPSslSocket6 |
| ~ broken_windows not_pypy |
| |
| server_tcp_up = threading.Event() |
| server_tls_up = threading.Event() |
| sniff_up = threading.Event() |
| def server_tls(): |
| context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) |
| _load_certificate_chain(context) |
| context.check_hostname = False |
| context.verify_mode = ssl.CERT_NONE |
| buffer = bytes.fromhex("02fd0006000000090e8011061000000000") |
| buffer += b'\x02\xfd\x80\x02\x00\x00\x00\x05\x00\x00\x00\x00\x00\x02\xfd\x80\x01\x00\x00\x00\n\x10\x10\x0e\x80P\x03\x002\x01\xf4' |
| sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) |
| sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| ssock = context.wrap_socket(sock) |
| try: |
| ssock.bind(('::1', 3496)) |
| ssock.listen(1) |
| server_tls_up.set() |
| connection, address = ssock.accept() |
| sniff_up.wait(timeout=1) |
| connection.send(buffer) |
| connection.close() |
| finally: |
| ssock.close() |
| |
| def server_tcp(): |
| buffer = bytes.fromhex("02fd0006000000090e8011060700000000") |
| sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) |
| try: |
| sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| sock.bind(('::1', 13400)) |
| sock.listen(1) |
| server_tcp_up.set() |
| connection, address = sock.accept() |
| connection.send(buffer) |
| connection.shutdown(socket.SHUT_RDWR) |
| connection.close() |
| finally: |
| sock.close() |
| |
| |
| server_tcp_thread = threading.Thread(target=server_tcp) |
| server_tcp_thread.start() |
| server_tcp_up.wait(timeout=1) |
| server_tls_thread = threading.Thread(target=server_tls) |
| server_tls_thread.start() |
| server_tls_up.wait(timeout=1) |
| context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) |
| context.check_hostname = False |
| context.verify_mode = ssl.CERT_NONE |
| |
| |
| sock = UDS_DoIPSocket(ip="::1", context=context) |
| |
| pkts = sock.sniff(timeout=1, count=2, started_callback=sniff_up.set) |
| server_tcp_thread.join(timeout=1) |
| server_tls_thread.join(timeout=1) |
| assert len(pkts) == 2 |
