134 lines
7.7 KiB
Python
134 lines
7.7 KiB
Python
|
"""TEST-V2X2MAP-VAM-001: host regression evidence; see docs/vam-support.md."""
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
import struct
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
from its_g5_bridge import decode_itsg5_full
|
||
|
|
from vam import codec, decode_vam
|
||
|
|
|
||
|
|
VECTORS = json.loads((Path(__file__).parent / "tests/vam-vectors.json").read_text())
|
||
|
|
|
||
|
|
|
||
|
|
def packet(data, port=2018, hdr=24, length=None, transport=2):
|
||
|
|
basic = b"\x11\x00\x00\x01"
|
||
|
|
common = bytes([transport << 4, 0x50, 0, 0]) + struct.pack("!H", len(data) + 4 if length is None else length) + b"\x01\x00"
|
||
|
|
# Deliberately differs from the VAM: map values must come from ASN.1.
|
||
|
|
lpv = bytes.fromhex("0800112233445566") + struct.pack("!IiiHH", 0, 520000000, 130000000, 700, 1800)
|
||
|
|
return bytes(hdr) + bytes.fromhex("aaaa030000008947") + basic + common + lpv + bytes(4) + struct.pack("!HH", port, 0) + data
|
||
|
|
|
||
|
|
|
||
|
|
class VamTests(unittest.TestCase):
|
||
|
|
def test_independent_asn1c_vectors(self):
|
||
|
|
for vector in VECTORS:
|
||
|
|
sid, lat, lon, time, speed, heading = vector["fields"]
|
||
|
|
with self.subTest(station=sid):
|
||
|
|
vam = decode_vam(bytes.fromhex(vector["uper"]))
|
||
|
|
self.assertEqual((vam["station_id"], vam["lat"], vam["lon"], vam["generation_delta_time"]),
|
||
|
|
(sid, lat / 1e7, lon / 1e7, time))
|
||
|
|
self.assertEqual(vam.get("speed_mps"), None if speed == 16383 else speed / 100)
|
||
|
|
self.assertEqual(vam.get("heading_deg"), None if heading == 3601 else heading / 10)
|
||
|
|
self.assertNotIn("altitude_m", vam)
|
||
|
|
self.assertNotIn("longitudinal_acceleration_mps2", vam)
|
||
|
|
|
||
|
|
def test_all_truncations_and_trailing_data(self):
|
||
|
|
for vector in VECTORS:
|
||
|
|
data = bytes.fromhex(vector["uper"])
|
||
|
|
for cut in range(len(data)):
|
||
|
|
with self.subTest(cut=cut), self.assertRaises(ValueError):
|
||
|
|
decode_vam(data[:cut])
|
||
|
|
for bad in (data + b"\x00", data[:-1] + bytes([data[-1] | 1]), b"\x01" + data[1:], data[:1] + b"\x0e" + data[2:]):
|
||
|
|
with self.assertRaises(ValueError):
|
||
|
|
decode_vam(bad)
|
||
|
|
|
||
|
|
def test_optional_containers_and_json(self):
|
||
|
|
message = codec().decode("VAM", bytes.fromhex(VECTORS[1]["uper"]))
|
||
|
|
containers = message["vam"]["vamParameters"]
|
||
|
|
hf = containers["vruHighFrequencyContainer"]
|
||
|
|
hf.update(environment=3, movementControl=1, deviceUsage=4,
|
||
|
|
orientation={"value": 1800, "confidence": 10})
|
||
|
|
containers["vruLowFrequencyContainer"] = {
|
||
|
|
"profileAndSubprofile": ("bicyclistAndLightVruVehicle", 1),
|
||
|
|
"exteriorLights": {"vehicular": (b"\x80", 8), "vruSpecific": (b"\x40", 8)}}
|
||
|
|
containers["vruClusterOperationContainer"] = {"clusterIdChangeTimeInfo": 5}
|
||
|
|
containers["vruMotionPredictionContainer"] = {"pathHistory": []}
|
||
|
|
vam = decode_vam(codec().encode("VAM", message, check_constraints=True))
|
||
|
|
self.assertEqual(vam["containers"]["vruHighFrequencyContainer"]["environment"], 3)
|
||
|
|
self.assertEqual(vam["containers"]["vruLowFrequencyContainer"]["exteriorLights"]["vehicular"], ["80", 8])
|
||
|
|
json.dumps(vam) # SSE must accept CHOICE and BIT STRING data.
|
||
|
|
|
||
|
|
def test_unavailable_coordinates_and_real_zero(self):
|
||
|
|
data = bytes.fromhex(VECTORS[0]["uper"])
|
||
|
|
self.assertEqual(decode_itsg5_full(packet(data))["lat"], 0)
|
||
|
|
message = codec().decode("VAM", data)
|
||
|
|
position = message["vam"]["vamParameters"]["basicContainer"]["referencePosition"]
|
||
|
|
position["latitude"] = 900000001
|
||
|
|
result = decode_itsg5_full(packet(codec().encode("VAM", message)))
|
||
|
|
self.assertNotIn("lat", result)
|
||
|
|
self.assertNotIn("lon", result)
|
||
|
|
|
||
|
|
def test_transport_and_map_fields(self):
|
||
|
|
data = bytes.fromhex(VECTORS[1]["uper"])
|
||
|
|
for hdr in (24, 26, 30, 32):
|
||
|
|
for transport in (1, 2):
|
||
|
|
result = decode_itsg5_full(packet(data, hdr=hdr, transport=transport) + b"FCS!")
|
||
|
|
self.assertEqual(result["msg_type"], "VAM")
|
||
|
|
self.assertEqual(result["btp_dst_port"], 2018)
|
||
|
|
self.assertEqual(result["lat"], 53.9)
|
||
|
|
self.assertEqual(result["speed_mps"], 5)
|
||
|
|
self.assertEqual(result["station_type_label"], "cyclist")
|
||
|
|
for bad in (packet(data[:-1]), packet(data, length=999), packet(data, length=3), packet(data, transport=0)):
|
||
|
|
result = decode_itsg5_full(bad)
|
||
|
|
self.assertEqual(result["msg_type"], "VAM")
|
||
|
|
self.assertNotEqual(result["vam_status"], "decoded")
|
||
|
|
self.assertNotIn("lat", result)
|
||
|
|
self.assertEqual(decode_itsg5_full(packet(data, port=2001))["msg_type"], "CAM")
|
||
|
|
self.assertEqual(decode_itsg5_full(packet(data, port=2001))["lat"], 52)
|
||
|
|
|
||
|
|
def test_secured_header_is_not_guessed(self):
|
||
|
|
data = bytearray(packet(bytes.fromhex(VECTORS[0]["uper"])))
|
||
|
|
data[32] = 0x12
|
||
|
|
result = decode_itsg5_full(data)
|
||
|
|
self.assertTrue(result["secured"])
|
||
|
|
self.assertEqual(result["msg_type"], "UNKNOWN")
|
||
|
|
self.assertNotIn("lat", result)
|
||
|
|
|
||
|
|
# A signed VAM the micrOBU (ESP32-C5, vanetza-idf) transmitted on 2026-09-14, recorded by this
|
||
|
|
# bridge (recordings/its5_20260914_202525.pcap, frame 1): TS 103 097 signedData with the
|
||
|
|
# certificate inline (PSID 638), the GN common header, BTP-B 2018 and the VAM as unsecuredData.
|
||
|
|
SIGNED_VAM = bytes.fromhex("88000000ffffffffffff62d0e89fe341ffffffffffff80000300aaaa030000008947120005010381004003804c2050028000280100080062d0e89fe341d2700e411feb4f4505f4036181900b230000000007e200000310e95759360efd400aac81c2b389eea7904b0320000c5a8500058ab106418a833222204002027e00028ba605ba90dd8101018003008087e5d18581e8dbd7108300000000002ab4995b840018010480012481040301fffc80012581050401ffffff00018d8002027e81020101808082f5fc0a86649cfbebb4eb74b606659d3efea74852c742f53fa32b136f29d7ee9a8080bb6fea25df074c2db2d36bcbbeba665236668b645efbb1bdf425569f7bf3a14715edaad8e4b38c1d82368e435c861a48f146e3871cd804a56b0ff4e226d213fc8080efa94132b1e45eca98bbf0af2f391266a86920230f15faab5f15d8d37e01ef5e45c5d373470e530cc42834a7212eb94ae035829960d05db8bdd87ab87c3cd9910000007401000088")
|
||
|
|
|
||
|
|
def test_signed_vam_is_unwrapped_but_not_verified(self):
|
||
|
|
result = decode_itsg5_full(self.SIGNED_VAM)
|
||
|
|
self.assertEqual(result["msg_type"], "VAM")
|
||
|
|
self.assertTrue(result["secured"])
|
||
|
|
self.assertEqual(result["vam_status"], "decoded")
|
||
|
|
self.assertEqual(result["security"]["psid"], 638)
|
||
|
|
self.assertEqual(result["security"]["signer"], "certificate")
|
||
|
|
self.assertEqual(result["security"]["signer_id"], "ba1862d0e89fe341") # HashedId8 of the ticket
|
||
|
|
self.assertFalse(result["security"]["verified"])
|
||
|
|
self.assertEqual(result["vam"]["station_type"], 2)
|
||
|
|
self.assertEqual(result["vam"]["speed_mps"], 4.0)
|
||
|
|
self.assertEqual(result["gn_addr"], "62:d0:e8:9f:e3:41")
|
||
|
|
# the bridge never guesses: every truncation of the envelope (up to its signature; the
|
||
|
|
# receiver appends 8 octets after it) leaves the packet undecoded
|
||
|
|
from secured import unwrap_signed
|
||
|
|
envelope_end = 38 + unwrap_signed(self.SIGNED_VAM[38:])["consumed"]
|
||
|
|
for cut in range(38, envelope_end, 7):
|
||
|
|
cut_result = decode_itsg5_full(self.SIGNED_VAM[:cut])
|
||
|
|
self.assertNotEqual(cut_result.get("vam_status"), "decoded", cut)
|
||
|
|
self.assertNotIn("lat", cut_result, cut)
|
||
|
|
|
||
|
|
def test_malformed_never_falls_back_to_gn_position(self):
|
||
|
|
data = bytes.fromhex(VECTORS[0]["uper"])
|
||
|
|
for bit in range(len(data) * 8):
|
||
|
|
bad = bytearray(data)
|
||
|
|
bad[bit // 8] ^= 1 << (7 - bit % 8)
|
||
|
|
result = decode_itsg5_full(packet(bad))
|
||
|
|
if result["vam_status"] != "decoded":
|
||
|
|
self.assertNotIn("lat", result)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|