Keep vanetza-idf in obu-firmware, so a plain clone builds the firmware
obu-firmware builds against the vanetza-idf C-ITS library, which until now came from the colleague's microbu-esp32c5 tree beside the repository and was not tracked here, so a clone of this repository could not build the firmware it ships. The library alone is now part of obu-firmware, as obu-firmware/external/vanetza-idf: their external/vanetza-idf at commit cf4b99f, unchanged (9775 files; see its PROVENANCE.md). CMake takes it from there by default; -DVANETZA_IDF_DIR still points the build elsewhere. The rest of the colleague's tree (their own VAM firmware, PKI tooling, station-link Python tools, the V2X2MAP bridge) stays out of this repository and gitignored; nothing is pushed to their repository. NOTES.md, docs/06, TODO.md and the pcap verifier's usage line point at the new location.
This commit is contained in:
Ashin Walpola
+51
@@ -0,0 +1,51 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Link the BTP SUT adapter against a separately built official ETSI framework.
|
||||
|
||||
Run on the same Linux/TITAN runtime as the supplied build. Never installs TTCN,
|
||||
edits the ETSI source tree, or modifies its testcase objects.
|
||||
"""
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--etsi', type=Path, required=True)
|
||||
parser.add_argument('--titan', type=Path, required=True)
|
||||
parser.add_argument('--out', type=Path, required=True)
|
||||
args = parser.parse_args()
|
||||
root, titan, out = args.etsi.resolve(), args.titan.resolve(), args.out.resolve()
|
||||
out.mkdir(parents=True, exist_ok=True)
|
||||
build = root / 'build/AtsBTP'
|
||||
source = Path(__file__).resolve().parents[1] / 'tests/etsi_btp_adapter.cpp'
|
||||
if not (build / 'ItsBtp_TestCases.o').is_file():
|
||||
raise SystemExit('Build the official AtsBTP suite separately first')
|
||||
includes = {build, titan / 'include', root / 'ccsrc/Ports/LibIts_ports/BTP_ports'}
|
||||
for parent in (root / 'ccsrc', root / 'titan-test-system-framework/ccsrc'):
|
||||
includes.update(p.parent for p in parent.rglob('*.hh'))
|
||||
obj = out / 'etsi_btp_adapter.o'
|
||||
command = ['g++', '-std=c++17', '-g', '-O0', '-DTITAN_RUNTIME_2', '-D_NO_SOFTLINKS_',
|
||||
'-DLINUX', '-DAtsBTP', '-DAS_USE_SSL', '-I/usr/include/libxml2', '-I/usr/include/jsoncpp']
|
||||
command += ['-I' + str(p) for p in sorted(includes)]
|
||||
subprocess.run(command + ['-c', str(source), '-o', str(obj)], check=True)
|
||||
objects = sorted(p for p in build.rglob('*.o') if 'asn1' not in p.relative_to(build).parts
|
||||
and p.name not in ('BtpPort.o', 'UpperTesterPort_BTP.o'))
|
||||
binary = out / 'AtsBTP'
|
||||
link = ['g++', '-o', str(binary), str(obj)] + [str(p) for p in objects]
|
||||
link += [str(build / 'asn1/libItsAsn.a'), str(titan / 'lib/libttcn3-rt2-parallel.a'),
|
||||
'-lstdc++fs', '-lpcap', '-lrt', '-lpthread', '-lssl', '-lcrypto', '-lxml2', '-ljsoncpp', '-lzip', '-lsctp']
|
||||
subprocess.run(link, check=True)
|
||||
sha = lambda path: hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
evidence = {'adapter_sha256': sha(source), 'binary_sha256': sha(binary),
|
||||
'testcase_object_sha256': sha(build / 'ItsBtp_TestCases.o'),
|
||||
'testcase_source_sha256': sha(root / 'ttcn/AtsBTP/ItsBtp_TestCases.ttcn'),
|
||||
'scope': 'Official BTP testcase objects; custom host/device SUT ports; no conformance verdict implied'}
|
||||
(out / 'build.json').write_text(json.dumps(evidence, indent=2) + '\n')
|
||||
print(binary)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
+107
@@ -0,0 +1,107 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Link the GeoNetworking SUT adapter against a separately built official ETSI framework.
|
||||
|
||||
Run on the same Linux/TITAN runtime as the supplied build. Never installs TTCN,
|
||||
edits the ETSI source tree in place, or modifies its testcase objects or bodies.
|
||||
|
||||
The one exception is a disposable build overlay for a framework null-pointer bug
|
||||
(see PAYLOAD_NULL_PARAMS_PATCH below): the pinned source on disk is read but never
|
||||
written, only a patched copy compiled into this tool's own --out directory.
|
||||
"""
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
# geonetworking_codec::decode_ (ccsrc/Protocols/GeoNetworking/geonetworking_codec.cc)
|
||||
# unconditionally dereferences `_params` while decoding GnNonSecuredPacket.payload, in
|
||||
# all three branches of the length-alignment logic. Every other parameter write in
|
||||
# this file (in the enclosing decode(), a few lines above decode_) is correctly
|
||||
# guarded with `if (_params != NULL)`; this one just wasn't. fx__dec__GeoNetworkingPdu
|
||||
# (ccsrc/EncDec/LibItsGeoNetworking_Encdec.cc) always calls decode() with the default
|
||||
# params (nullptr), so decoding *any* GeoNetworkingPdu through the official codec
|
||||
# wrapper segfaults as soon as it reaches this field -- confirmed via a core dump
|
||||
# (gdb bt full) naming this exact line, not a maybe. Never triggered by BTP (whose
|
||||
# adapter never calls this decoder) or by any earlier GeoNetworking run, because no
|
||||
# GN packet had ever actually been transmitted-and-observed through TITAN before
|
||||
# (store-carry-forward buffering meant every prior SHB attempt just sat unsent).
|
||||
PAYLOAD_NULL_PARAMS_PATCH = [
|
||||
(' os = OCTETSTRING(s.lengthof(), p);\n'
|
||||
' (*_params)[params_its::gn_payload] = static_cast<const char *>(oct2str(os));\n',
|
||||
' os = OCTETSTRING(s.lengthof(), p);\n'
|
||||
' if (_params != NULL) (*_params)[params_its::gn_payload] = static_cast<const char *>(oct2str(os));\n'),
|
||||
(' os = OCTETSTRING(_dc.get_length(), p);\n'
|
||||
' (*_params)[params_its::gn_payload] = static_cast<const char *>(oct2str(os));\n',
|
||||
' os = OCTETSTRING(_dc.get_length(), p);\n'
|
||||
' if (_params != NULL) (*_params)[params_its::gn_payload] = static_cast<const char *>(oct2str(os));\n'),
|
||||
(' os = OCTETSTRING(0, nullptr);\n'
|
||||
' (*_params)[params_its::gn_payload] = "";\n',
|
||||
' os = OCTETSTRING(0, nullptr);\n'
|
||||
' if (_params != NULL) (*_params)[params_its::gn_payload] = "";\n'),
|
||||
]
|
||||
|
||||
|
||||
def patch_geonetworking_codec(build, out, includes, compile_flags):
|
||||
source = build.parents[1] / 'ccsrc/Protocols/GeoNetworking/geonetworking_codec.cc'
|
||||
text = source.read_text()
|
||||
before_sha = hashlib.sha256(text.encode()).hexdigest()
|
||||
for old, new in PAYLOAD_NULL_PARAMS_PATCH:
|
||||
if text.count(old) != 1:
|
||||
raise SystemExit(f'geonetworking_codec.cc overlay: expected exactly one match for {old!r}')
|
||||
text = text.replace(old, new)
|
||||
patched = out / 'geonetworking_codec.patched.cc'
|
||||
patched.write_text(text)
|
||||
obj = out / 'geonetworking_codec.patched.o'
|
||||
command = compile_flags + ['-I' + str(p) for p in sorted(includes)] + ['-c', str(patched), '-o', str(obj)]
|
||||
subprocess.run(command, check=True)
|
||||
return obj, before_sha, hashlib.sha256(text.encode()).hexdigest()
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--etsi', type=Path, required=True)
|
||||
parser.add_argument('--titan', type=Path, required=True)
|
||||
parser.add_argument('--out', type=Path, required=True)
|
||||
args = parser.parse_args()
|
||||
root, titan, out = args.etsi.resolve(), args.titan.resolve(), args.out.resolve()
|
||||
out.mkdir(parents=True, exist_ok=True)
|
||||
build = root / 'build/AtsGeoNetworking'
|
||||
source = Path(__file__).resolve().parents[1] / 'tests/etsi_geonetworking_adapter.cpp'
|
||||
if not (build / 'ItsGeoNetworking_TestCases.o').is_file():
|
||||
raise SystemExit('Build the official AtsGeoNetworking suite separately first')
|
||||
includes = {build, titan / 'include', root / 'ccsrc/Ports/LibIts_ports/GN_ports'}
|
||||
for parent in (root / 'ccsrc', root / 'titan-test-system-framework/ccsrc'):
|
||||
includes.update(p.parent for p in parent.rglob('*.hh'))
|
||||
compile_flags = ['g++', '-std=c++17', '-g', '-O0', '-DTITAN_RUNTIME_2', '-D_NO_SOFTLINKS_',
|
||||
'-DLINUX', '-DAtsGeoNetworking', '-DAS_USE_SSL', '-I/usr/include/libxml2', '-I/usr/include/jsoncpp']
|
||||
obj = out / 'etsi_geonetworking_adapter.o'
|
||||
subprocess.run(compile_flags + ['-I' + str(p) for p in sorted(includes)] + ['-c', str(source), '-o', str(obj)], check=True)
|
||||
codec_obj, codec_before_sha, codec_after_sha = patch_geonetworking_codec(build, out, includes, compile_flags)
|
||||
objects = sorted(p for p in build.rglob('*.o') if 'asn1' not in p.relative_to(build).parts
|
||||
and p.name not in ('GeoNetworkingPort.o', 'UpperTesterPort_GN.o', 'AdapterControlPort_GN.o',
|
||||
'geonetworking_codec.o'))
|
||||
binary = out / 'AtsGeoNetworking'
|
||||
link = ['g++', '-o', str(binary), str(obj), str(codec_obj)] + [str(p) for p in objects]
|
||||
link += [str(build / 'asn1/libItsAsn.a'), str(titan / 'lib/libttcn3-rt2-parallel.a'),
|
||||
'-lstdc++fs', '-lpcap', '-lrt', '-lpthread', '-lssl', '-lcrypto', '-lxml2', '-ljsoncpp', '-lzip', '-lsctp']
|
||||
subprocess.run(link, check=True)
|
||||
sha = lambda path: hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
evidence = {'adapter_sha256': sha(source), 'binary_sha256': sha(binary),
|
||||
'testcase_object_sha256': sha(build / 'ItsGeoNetworking_TestCases.o'),
|
||||
'testcase_source_sha256': sha(root / 'ttcn/AtsGeoNetworking/ItsGeoNetworking_TestCases.ttcn'),
|
||||
'scope': 'Official GeoNetworking testcase objects; custom host/device SUT ports, SHB source only; no conformance verdict implied',
|
||||
'suite_overlay': [{
|
||||
'path': 'ccsrc/Protocols/GeoNetworking/geonetworking_codec.cc',
|
||||
'before': codec_before_sha, 'after': codec_after_sha,
|
||||
'reason': 'Null-guard three unconditional (*_params)[...] writes while decoding '
|
||||
'GnNonSecuredPacket.payload; fx__dec__GeoNetworkingPdu always decodes with '
|
||||
'params=nullptr, so every decode of an observed GN packet segfaulted. Pinned '
|
||||
'source on disk is untouched; only this tool\'s own --out build uses the patch.'
|
||||
}]}
|
||||
(out / 'build.json').write_text(json.dumps(evidence, indent=2) + '\n')
|
||||
print(binary)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
+70
@@ -0,0 +1,70 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Link this library's AtsSecurity SUT adapter with the user's compiled ETSI suite.
|
||||
|
||||
Mirrors build_etsi_geonetworking_adapter.py: the official testcase objects of a
|
||||
separately built AtsSecurity are reused unchanged; the framework's own
|
||||
GeoNetworking/CAM/DENM port objects are replaced by ports/esp_idf/tests/
|
||||
etsi_security_adapter.cpp, and the same disposable geonetworking_codec.cc
|
||||
overlay (null params guard) is applied to this build's output only. Retain the
|
||||
printed build.json next to the campaign result.
|
||||
"""
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
from build_etsi_geonetworking_adapter import patch_geonetworking_codec # noqa: E402
|
||||
|
||||
REPLACED_OBJECTS = ('GeoNetworkingPort.o', 'UpperTesterPort_GN.o', 'AdapterControlPort_GN.o',
|
||||
'UpperTesterPort_CAM.o', 'UpperTesterPort_DENM.o', 'geonetworking_codec.o')
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--etsi', type=Path, required=True)
|
||||
parser.add_argument('--titan', type=Path, required=True)
|
||||
parser.add_argument('--out', type=Path, required=True)
|
||||
args = parser.parse_args()
|
||||
root, titan, out = args.etsi.resolve(), args.titan.resolve(), args.out.resolve()
|
||||
out.mkdir(parents=True, exist_ok=True)
|
||||
build = root / 'build/AtsSecurity'
|
||||
source = Path(__file__).resolve().parents[1] / 'tests/etsi_security_adapter.cpp'
|
||||
if not (build / 'ItsSecurity_TestCases.o').is_file():
|
||||
raise SystemExit('Build the official AtsSecurity suite separately first')
|
||||
ports = root / 'ccsrc/Ports/LibIts_ports'
|
||||
includes = {build, titan / 'include', ports / 'GN_ports', ports / 'CAM_ports', ports / 'DENM_ports',
|
||||
Path(__file__).resolve().parents[1] / 'include'} # vanetza_idf/its_time.hpp (header-only)
|
||||
for parent in (root / 'ccsrc', root / 'titan-test-system-framework/ccsrc'):
|
||||
includes.update(p.parent for p in parent.rglob('*.hh'))
|
||||
compile_flags = ['g++', '-std=c++17', '-g', '-O0', '-DTITAN_RUNTIME_2', '-D_NO_SOFTLINKS_',
|
||||
'-DLINUX', '-DAtsSecurity', '-DAS_USE_SSL', '-I/usr/include/libxml2', '-I/usr/include/jsoncpp']
|
||||
obj = out / 'etsi_security_adapter.o'
|
||||
subprocess.run(compile_flags + ['-I' + str(p) for p in sorted(includes)] + ['-c', str(source), '-o', str(obj)], check=True)
|
||||
codec_obj, codec_before_sha, codec_after_sha = patch_geonetworking_codec(build, out, includes, compile_flags)
|
||||
objects = sorted(p for p in build.rglob('*.o') if 'asn1' not in p.relative_to(build).parts
|
||||
and p.name not in REPLACED_OBJECTS)
|
||||
binary = out / 'AtsSecurity'
|
||||
link = ['g++', '-o', str(binary), str(obj), str(codec_obj)] + [str(p) for p in objects]
|
||||
link += [str(build / 'asn1/libItsAsn.a'), str(titan / 'lib/libttcn3-rt2-parallel.a'),
|
||||
'-lstdc++fs', '-lpcap', '-lrt', '-lpthread', '-lssl', '-lcrypto', '-lxml2', '-ljsoncpp', '-lzip', '-lsctp']
|
||||
subprocess.run(link, check=True)
|
||||
sha = lambda path: hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
evidence = {'adapter_sha256': sha(source), 'binary_sha256': sha(binary),
|
||||
'testcase_object_sha256': sha(build / 'ItsSecurity_TestCases.o'),
|
||||
'testcase_source_sha256': sha(root / 'ttcn/AtsSecurity/ItsSecurity_TestCases.ttcn'),
|
||||
'scope': 'Official Security testcase objects; custom host SUT ports for the sending side '
|
||||
'(GN-MGMT beacons, CAM/DENM carriers); receiving cases not wired; no conformance verdict implied',
|
||||
'suite_overlay': [{
|
||||
'path': 'ccsrc/Protocols/GeoNetworking/geonetworking_codec.cc',
|
||||
'before': codec_before_sha, 'after': codec_after_sha,
|
||||
'reason': 'Same null-params guard as the GeoNetworking adapter build; pinned source untouched.'
|
||||
}]}
|
||||
(out / 'build.json').write_text(json.dumps(evidence, indent=2) + '\n')
|
||||
print(binary)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,153 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Record what the SUT sends as an IEEE 802.11 pcap for independent verifiers.
|
||||
|
||||
Drives ``vidf_sut`` (hex-line protocol, see tests/hil_stdio.cpp) with a credential
|
||||
chain from a pool directory, or a device running the test application over its USB
|
||||
serial diagnostic channel (``--port`` with a credential bundle from
|
||||
credential_bundle.py), triggers the CAM and DENM carriers and writes every AL-DATA.request as a
|
||||
data frame (linktype 105, no radiotap header) as sent outside the context of a
|
||||
BSS (EN 302 663 V1.3.1 clause 4.3.4, dot11OCBActivated; Annex C: the BSSID is the
|
||||
wildcard in every frame), LLC/SNAP with the GeoNetworking EtherType 0x8947
|
||||
(clause 4.3.3).
|
||||
A verifier such as c-its-pcap (https://github.com/TheEnbyperor/c-its) can then
|
||||
check the signatures and certificate chains against the root it is given.
|
||||
|
||||
This is a test tool, not part of the library; it makes no verdict of its own.
|
||||
"""
|
||||
import argparse
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
ITS_EPOCH_UNIX = 1072915200 # 2004-01-01T00:00:00Z
|
||||
LEAP_SECONDS = 5 # TAI - UTC since the ITS epoch (TS 102 894-2 TimestampIts)
|
||||
GEONETWORKING_ETHERTYPE = b'\x89\x47'
|
||||
WILDCARD_BSSID = bytes([0xff] * 6)
|
||||
|
||||
|
||||
class HostSut:
|
||||
def __init__(self, executable, pool=None, root=None, authorities=(), ticket=None, bundle=None):
|
||||
if bundle:
|
||||
command = [executable, '--security-bundle', bundle]
|
||||
else:
|
||||
command = [executable, '--security-pool', pool, '--root', root, '--at', ticket]
|
||||
for authority in authorities:
|
||||
command += ['--aa', authority]
|
||||
self.process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, bufsize=1)
|
||||
|
||||
def execute(self, payload):
|
||||
self.process.stdin.write(payload.hex() + '\n')
|
||||
self.process.stdin.flush()
|
||||
reply = bytes.fromhex(self.process.stdout.readline().strip())
|
||||
result, count = reply[0], reply[1]
|
||||
records, at = [], 2
|
||||
for _ in range(count):
|
||||
kind, size = reply[at], struct.unpack('>H', reply[at + 1:at + 3])[0]
|
||||
records.append((kind, reply[at + 3:at + 3 + size]))
|
||||
at += 3 + size
|
||||
return result, records
|
||||
|
||||
def close(self):
|
||||
self.process.stdin.close()
|
||||
self.process.wait()
|
||||
|
||||
|
||||
class DeviceSut:
|
||||
"""the same command protocol over the USB serial diagnostic channel (serial_sut.py)"""
|
||||
def __init__(self, port, bundle):
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
import serial_sut
|
||||
self.serial = serial_sut.SerialSut(port)
|
||||
serial_sut.provision(self.serial, bundle)
|
||||
|
||||
def execute(self, payload):
|
||||
reply = self.serial.execute(payload)
|
||||
result, count = reply[0], reply[1]
|
||||
records, at = [], 2
|
||||
for _ in range(count):
|
||||
kind, size = reply[at], struct.unpack('>H', reply[at + 1:at + 3])[0]
|
||||
records.append((kind, reply[at + 3:at + 3 + size]))
|
||||
at += 3 + size
|
||||
return result, records
|
||||
|
||||
def close(self):
|
||||
self.serial.close()
|
||||
|
||||
|
||||
def its_microseconds(unix_seconds):
|
||||
return int((unix_seconds - ITS_EPOCH_UNIX + LEAP_SECONDS) * 1_000_000)
|
||||
|
||||
|
||||
def write_pcap(path, frames, source):
|
||||
with open(path, 'wb') as out:
|
||||
out.write(struct.pack('<IHHiIII', 0xa1b2c3d4, 2, 4, 0, 0, 65535, 105))
|
||||
for sequence, (timestamp, pdu) in enumerate(frames):
|
||||
header = (struct.pack('<HH', 0x0008, 0) + WILDCARD_BSSID + source + WILDCARD_BSSID +
|
||||
struct.pack('<H', (sequence & 0x0fff) << 4))
|
||||
frame = header + b'\xaa\xaa\x03\x00\x00\x00' + GEONETWORKING_ETHERTYPE + pdu
|
||||
out.write(struct.pack('<IIII', int(timestamp), int((timestamp % 1) * 1_000_000), len(frame), len(frame)))
|
||||
out.write(frame)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
parser.add_argument('--sut', help='vidf_sut executable (host station)')
|
||||
parser.add_argument('--pool', help='directory with <name>.oer / <name>.vkey (with --sut)')
|
||||
parser.add_argument('--root', help='root name in the pool (with --sut)')
|
||||
parser.add_argument('--aa', action='append', default=[], help='trusted authority (repeatable, with --sut)')
|
||||
parser.add_argument('--at', help='authorization ticket of the SUT (with --sut)')
|
||||
parser.add_argument('--port', help='serial port of a device running the test application (device station)')
|
||||
parser.add_argument('--bundle', help='credential bundle (credential_bundle.py build): with --port for the device, with --sut instead of --pool/--root/--at')
|
||||
parser.add_argument('--out', required=True, help='pcap file to write')
|
||||
parser.add_argument('--cams', type=int, default=3, help='CAM carriers to trigger (default 3)')
|
||||
parser.add_argument('--interval', type=float, default=0.6, help='ITS clock advance between carriers in seconds')
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.port:
|
||||
if not args.bundle:
|
||||
raise SystemExit('--port needs --bundle')
|
||||
sut = DeviceSut(args.port, args.bundle)
|
||||
elif args.sut and args.bundle:
|
||||
sut = HostSut(args.sut, bundle=args.bundle)
|
||||
elif args.sut and args.pool and args.root and args.at:
|
||||
sut = HostSut(args.sut, args.pool, args.root, args.aa or ['CERT_IUT_A_AA'], args.at)
|
||||
else:
|
||||
raise SystemExit('either --sut with --bundle or --pool/--root/--at, or --port with --bundle')
|
||||
result, _ = sut.execute(bytes([0]))
|
||||
if result != 0:
|
||||
raise SystemExit('SUT reset failed (result %d): credentials not loaded' % result)
|
||||
frames = []
|
||||
now = time.time()
|
||||
|
||||
def tick(at):
|
||||
result, records = sut.execute(bytes([5]) + its_microseconds(at).to_bytes(8, 'big'))
|
||||
frames.extend((at, data) for kind, data in records if kind == 1)
|
||||
return result
|
||||
|
||||
def carrier(at, kind):
|
||||
# command 8 defers the carrier to the next clock advance (command 5), the path the
|
||||
# ETSI adapter uses; the frame surfaces in that advance's records
|
||||
result, records = sut.execute(bytes([8, kind]))
|
||||
result, records = sut.execute(bytes([5]) + its_microseconds(at + 0.01).to_bytes(8, 'big'))
|
||||
frames.extend((at, data) for k, data in records if k == 1)
|
||||
print('carrier %d at +%.1fs -> result %d, %d frame(s)' % (kind, at - now, result, sum(1 for k, _ in records if k == 1)))
|
||||
|
||||
tick(now)
|
||||
for n in range(args.cams):
|
||||
at = now + n * args.interval
|
||||
tick(at)
|
||||
carrier(at, 0)
|
||||
at = now + args.cams * args.interval
|
||||
tick(at)
|
||||
carrier(at, 1)
|
||||
tick(at + 0.1)
|
||||
tick(at + 1.5)
|
||||
sut.close()
|
||||
write_pcap(args.out, frames, bytes([0x02, 0, 0, 0, 0, 1]))
|
||||
print('%d frame(s) written to %s' % (len(frames), args.out))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,94 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Build or inspect a credential bundle (vanetza_idf/credentials.hpp, format "VCR1").
|
||||
|
||||
A bundle carries what a station keeps: root CA certificates, subordinate CA
|
||||
certificates, and authorization tickets with their private keys, as records
|
||||
[type 1 octet][length 2 octets big-endian][payload]; types 1 root, 2 authority,
|
||||
3 ticket certificate, 4 ticket key ([curve 1 octet: 1 NIST P-256, 2 brainpoolP256r1,
|
||||
3 brainpoolP384r1][scalar]) directly after its certificate.
|
||||
|
||||
credential_bundle.py build --pool DIR --root NAME [--aa NAME]... --at NAME [--at NAME]... --out FILE
|
||||
NAME.oer (and NAME.vkey for tickets) from the pool directory, e.g. what
|
||||
vidf_issue or vidf_test_pool wrote.
|
||||
credential_bundle.py show FILE
|
||||
list the records (lengths and HashedId8 of each certificate).
|
||||
|
||||
Private keys travel in the clear inside the bundle: keep the file like the keys.
|
||||
This is a test tool, not part of the library.
|
||||
"""
|
||||
import argparse
|
||||
import hashlib
|
||||
import struct
|
||||
from pathlib import Path
|
||||
|
||||
MAGIC = b'VCR1'
|
||||
ROOT, AUTHORITY, TICKET, TICKET_KEY = 1, 2, 3, 4
|
||||
CURVE_BY_LENGTH = {32: 1, 48: 3} # a 32-octet scalar is taken as NIST P-256 (the pool's default)
|
||||
|
||||
|
||||
def record(kind, payload):
|
||||
if not 0 < len(payload) < 65536:
|
||||
raise ValueError('record payload must be 1..65535 octets')
|
||||
return struct.pack('>BH', kind, len(payload)) + payload
|
||||
|
||||
|
||||
def build(pool, root, authorities, tickets):
|
||||
out = bytearray(MAGIC)
|
||||
out += record(ROOT, (pool / f'{root}.oer').read_bytes())
|
||||
for name in authorities:
|
||||
out += record(AUTHORITY, (pool / f'{name}.oer').read_bytes())
|
||||
for name in tickets:
|
||||
out += record(TICKET, (pool / f'{name}.oer').read_bytes())
|
||||
scalar = (pool / f'{name}.vkey').read_bytes()
|
||||
if len(scalar) not in CURVE_BY_LENGTH:
|
||||
raise ValueError(f'{name}.vkey: {len(scalar)} octets is no supported scalar length')
|
||||
out += record(TICKET_KEY, bytes([CURVE_BY_LENGTH[len(scalar)]]) + scalar)
|
||||
return bytes(out)
|
||||
|
||||
|
||||
def parse(bundle):
|
||||
if bundle[:4] != MAGIC:
|
||||
raise ValueError('not a VCR1 bundle')
|
||||
at, records = 4, []
|
||||
while at < len(bundle):
|
||||
kind, length = struct.unpack_from('>BH', bundle, at)
|
||||
at += 3
|
||||
payload = bundle[at:at + length]
|
||||
if len(payload) != length or length == 0:
|
||||
raise ValueError('truncated or empty record')
|
||||
at += length
|
||||
records.append((kind, payload))
|
||||
return records
|
||||
|
||||
|
||||
def hashed_id8(coer):
|
||||
return hashlib.sha256(coer).digest()[-8:].hex().upper()
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
sub = parser.add_subparsers(dest='command', required=True)
|
||||
b = sub.add_parser('build')
|
||||
b.add_argument('--pool', type=Path, required=True)
|
||||
b.add_argument('--root', required=True)
|
||||
b.add_argument('--aa', action='append', default=[])
|
||||
b.add_argument('--at', action='append', default=[], required=True)
|
||||
b.add_argument('--out', type=Path, required=True)
|
||||
s = sub.add_parser('show')
|
||||
s.add_argument('file', type=Path)
|
||||
args = parser.parse_args()
|
||||
if args.command == 'build':
|
||||
bundle = build(args.pool, args.root, args.aa, args.at)
|
||||
args.out.write_bytes(bundle)
|
||||
print(f'{len(bundle)} octets: 1 root, {len(args.aa)} authorit{"y" if len(args.aa) == 1 else "ies"}, {len(args.at)} ticket(s) -> {args.out}')
|
||||
else:
|
||||
names = {ROOT: 'root', AUTHORITY: 'authority', TICKET: 'ticket', TICKET_KEY: 'ticket key'}
|
||||
for kind, payload in parse(args.file.read_bytes()):
|
||||
if kind == TICKET_KEY:
|
||||
print(f' {names[kind]:11} curve {payload[0]}, {len(payload) - 1} octets (not shown)')
|
||||
else:
|
||||
print(f' {names.get(kind, kind):11} {len(payload):4} octets HashedId8 {hashed_id8(payload)}')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,59 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Reset a flashed C5 test application and retain its component-test output.
|
||||
|
||||
Requires pyserial and esptool from the user's ESP-IDF environment.
|
||||
This produces component-test evidence, never an ETSI ATS verdict.
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import time
|
||||
from pathlib import Path
|
||||
from esptool.reset import HardReset
|
||||
from serial_sut import SerialSut
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--port', required=True)
|
||||
parser.add_argument('--out', type=Path, required=True)
|
||||
parser.add_argument('--timeout', type=float, default=30.0,
|
||||
help='seconds to wait for VIDF_TEST_RESULT (security builds need about 90)')
|
||||
parser.add_argument('--no-reset', action='store_true',
|
||||
help='only listen: use right after "idf.py flash", whose own reset boots the '
|
||||
'application (a DTR/RTS reset can leave the ESP32-C5 ROM waiting for UART0, '
|
||||
'see docs/idf/validation.md)')
|
||||
args = parser.parse_args()
|
||||
args.out.mkdir(parents=True, exist_ok=False)
|
||||
result = {'kind': 'component-tests', 'port': args.port, 'status': 'error'}
|
||||
output = bytearray()
|
||||
device = None
|
||||
try:
|
||||
device = SerialSut(args.port)
|
||||
device.port.reset_input_buffer()
|
||||
if not args.no_reset:
|
||||
HardReset(device.port, uses_usb=True)()
|
||||
result['reset'] = 'none' if args.no_reset else 'usb-hard-reset'
|
||||
deadline = time.monotonic() + args.timeout
|
||||
while time.monotonic() < deadline:
|
||||
output.extend(device.port.read(4096))
|
||||
if b'VIDF_TEST_RESULT=' in output and re.search(rb'VIDF_TEST_RESULT=\d+\r?\n', output):
|
||||
break
|
||||
text = output.decode(errors='replace')
|
||||
match = re.search(r'PASS: (\d+) checks', text)
|
||||
result['checks'] = int(match.group(1)) if match else None
|
||||
result['status'] = 'pass' if match and 'VIDF_TEST_RESULT=0' in text else 'fail'
|
||||
except Exception as error:
|
||||
result['reason'] = str(error)
|
||||
finally:
|
||||
if device:
|
||||
device.close()
|
||||
(args.out / 'console.txt').write_bytes(output)
|
||||
(args.out / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
|
||||
print(json.dumps(result, indent=2))
|
||||
print(output.decode(errors='replace'))
|
||||
raise SystemExit(0 if result['status'] == 'pass' else 1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,41 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Fetch a root CA's CTL and CRL from a distribution centre (TS 102 941 V2.2.1 Annex D).
|
||||
|
||||
fetch_trust_lists.py --dc http://127.0.0.1:8080/ --root ROOT.oer --out DIR
|
||||
|
||||
GET <dc>/getctl/<HASHEDID8> and <dc>/getcrl/<HASHEDID8> with the root's HashedId8
|
||||
(upper-case hex of the last eight octets of SHA-256 over the certificate), saved as
|
||||
DIR/ctl-<HASHEDID8>.oer and DIR/crl-<HASHEDID8>.oer. Nothing is verified here: read
|
||||
them back with `vidf_issue inspect FILE --root ROOT.oer`, which applies the checks of
|
||||
clause 6.3.6, or hand them to the station (pki::parse_rca_ctl / parse_crl / apply).
|
||||
The transport is the application's; this is the test tool's version of it.
|
||||
"""
|
||||
import argparse
|
||||
import hashlib
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
parser.add_argument('--dc', required=True, help='DC access point, e.g. http://127.0.0.1:8080/')
|
||||
parser.add_argument('--root', type=Path, required=True, help='root CA certificate (COER)')
|
||||
parser.add_argument('--out', type=Path, required=True)
|
||||
args = parser.parse_args()
|
||||
id8 = hashlib.sha256(args.root.read_bytes()).digest()[-8:].hex().upper()
|
||||
base = args.dc if args.dc.endswith('/') else args.dc + '/'
|
||||
args.out.mkdir(parents=True, exist_ok=True)
|
||||
for kind in ('ctl', 'crl'):
|
||||
url = f'{base}get{kind}/{id8}'
|
||||
with urllib.request.urlopen(url, timeout=10) as response:
|
||||
body = response.read()
|
||||
content_type = response.headers.get('Content-Type', '')
|
||||
if content_type != f'application/x-its-{kind}':
|
||||
print(f'{url}: unexpected Content-Type {content_type!r} (Annex D: application/x-its-{kind})')
|
||||
target = args.out / f'{kind}-{id8}.oer'
|
||||
target.write_bytes(body)
|
||||
print(f'{url} -> {target} ({len(body)} octets)')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,67 @@
|
||||
#!/usr/bin/env python3
|
||||
"""A distribution centre on localhost: TS 102 941 V2.2.1 Annex D, Tables D.1 and D.2.
|
||||
|
||||
Serves the CTL and CRL of one or more root CAs the way an ITS-S fetches them:
|
||||
|
||||
GET /getctl/<HASHEDID8> -> the latest full CTL, application/x-its-ctl
|
||||
GET /getctl/<HASHEDID8>/<sequence> -> a delta CTL of that sequence (404 here: only full CTLs are kept)
|
||||
GET /getcrl/<HASHEDID8> -> the CRL, application/x-its-crl
|
||||
|
||||
The files come from a directory holding ctl-<HASHEDID8>.oer and crl-<HASHEDID8>.oer as
|
||||
vidf_issue ctl / crl write them (HASHEDID8 in upper-case hex). Nothing is verified here:
|
||||
the DC is a file server; the ITS-S verifies the RCA signature (clause 6.3.6).
|
||||
|
||||
local_dc.py --dir lists --port 8080
|
||||
|
||||
Test tool, not part of the library.
|
||||
"""
|
||||
import argparse
|
||||
import http.server
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
PATTERN = re.compile(r'^/get(ctl|crl)/([0-9A-F]{16})(?:/(\d+))?/?$')
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
parser.add_argument('--dir', type=Path, required=True, help='directory with ctl-<HASHEDID8>.oer / crl-<HASHEDID8>.oer')
|
||||
parser.add_argument('--port', type=int, default=8080)
|
||||
parser.add_argument('--bind', default='127.0.0.1')
|
||||
args = parser.parse_args()
|
||||
lists = args.dir.resolve()
|
||||
|
||||
class Handler(http.server.BaseHTTPRequestHandler):
|
||||
def do_GET(self):
|
||||
match = PATTERN.match(self.path)
|
||||
if not match:
|
||||
self.send_error(404, 'not a getctl/getcrl path')
|
||||
return
|
||||
kind, id8, sequence = match.groups()
|
||||
if sequence is not None:
|
||||
self.send_error(404, 'delta CTLs are not kept by this distribution centre')
|
||||
return
|
||||
path = lists / f'{kind}-{id8}.oer'
|
||||
if not path.is_file():
|
||||
self.send_error(404, f'no {kind} for {id8}')
|
||||
return
|
||||
body = path.read_bytes()
|
||||
self.send_response(200)
|
||||
self.send_header('Content-Type', f'application/x-its-{kind}')
|
||||
self.send_header('Content-Length', str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def log_message(self, fmt, *values):
|
||||
print('%s %s' % (self.address_string(), fmt % values), flush=True)
|
||||
|
||||
server = http.server.ThreadingHTTPServer((args.bind, args.port), Handler)
|
||||
print(f'distribution centre on http://{args.bind}:{args.port}/ serving {lists}', flush=True)
|
||||
try:
|
||||
server.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,105 @@
|
||||
#!/usr/bin/env python3
|
||||
"""An EA and AA on localhost: TS 102 941 V2.2.1 clause 6.2.3 enrolment/authorization.
|
||||
|
||||
Serves the two HTTP endpoints an ITS-S POSTs its requests to:
|
||||
|
||||
POST /ea/enrolment body = EnrolmentRequest (EtsiTs103097Data-Encrypted)
|
||||
-> EnrolmentResponse (EtsiTs103097Data-Encrypted), 200
|
||||
POST /aa/authorization body = AuthorizationRequest (EtsiTs103097Data-Encrypted)
|
||||
-> AuthorizationResponse (EtsiTs103097Data-Encrypted), 200
|
||||
|
||||
This tool carries bytes only: every decrypt, signature check, certificate issuance
|
||||
and re-encryption is done by `vidf_issue ea-respond` / `aa-respond` (VIDF_PKI=ON),
|
||||
run once per request as a subprocess -- exactly the division of labour local_dc.py
|
||||
has for the distribution centre, just for clause 6.2.3 instead of clause 6.3. A
|
||||
request `vidf_issue` cannot decrypt/verify/decode gets HTTP 400: a real ITS-S never
|
||||
sends one, so this is tooling clarity, not part of the TS 102 941 message exchange.
|
||||
|
||||
local_pki.py --issue-tool build/vidf_issue \
|
||||
--ea EA.oer --ea-key EA.vkey --ea-enc-key EA.ekey --canonical-key canonical.pem \
|
||||
--aa AA.oer --aa-key AA.vkey --aa-enc-key AA.ekey \
|
||||
--dir issued --port 8090
|
||||
|
||||
--canonical-key is the initial-enrolment case only (clause 6.2.3.2.1): the EA is
|
||||
handed the same canonical private key file the station used, standing in for the
|
||||
manufacturer's out-of-band registry a lab has no other side of. Re-enrolment
|
||||
(--current-ec) is not wired up here; use `vidf_issue ea-respond --current-ec`
|
||||
directly for that.
|
||||
|
||||
Test tool, not part of the library.
|
||||
"""
|
||||
import argparse
|
||||
import http.server
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
parser.add_argument('--issue-tool', type=Path, required=True, help='path to the vidf_issue binary (VIDF_PKI=ON)')
|
||||
parser.add_argument('--ea', type=Path, required=True)
|
||||
parser.add_argument('--ea-key', type=Path, required=True)
|
||||
parser.add_argument('--ea-enc-key', type=Path, required=True)
|
||||
parser.add_argument('--canonical-key', type=Path, required=True, help='initial enrolment only, see above')
|
||||
parser.add_argument('--aa', type=Path, required=True)
|
||||
parser.add_argument('--aa-key', type=Path, required=True)
|
||||
parser.add_argument('--aa-enc-key', type=Path, required=True)
|
||||
parser.add_argument('--dir', type=Path, required=True, help='EC store ea-respond writes to and aa-respond reads from')
|
||||
parser.add_argument('--port', type=int, default=8090)
|
||||
parser.add_argument('--bind', default='127.0.0.1')
|
||||
args = parser.parse_args()
|
||||
args.dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def respond(tool_args, request_body):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
request_path = Path(tmp) / 'request.bin'
|
||||
response_path = Path(tmp) / 'response.bin'
|
||||
request_path.write_bytes(request_body)
|
||||
result = subprocess.run(
|
||||
[str(args.issue_tool), *tool_args, '--request', str(request_path), '--out', str(response_path)],
|
||||
capture_output=True, text=True)
|
||||
if result.returncode != 0:
|
||||
return None, result.stdout + result.stderr
|
||||
return response_path.read_bytes(), result.stdout + result.stderr
|
||||
|
||||
class Handler(http.server.BaseHTTPRequestHandler):
|
||||
def do_POST(self):
|
||||
length = int(self.headers.get('Content-Length', 0))
|
||||
body = self.rfile.read(length)
|
||||
if self.path == '/ea/enrolment':
|
||||
tool_args = ['ea-respond', '--ea', str(args.ea), '--ea-key', str(args.ea_key),
|
||||
'--ea-enc-key', str(args.ea_enc_key), '--canonical-key', str(args.canonical_key),
|
||||
'--dir', str(args.dir)]
|
||||
elif self.path == '/aa/authorization':
|
||||
tool_args = ['aa-respond', '--aa', str(args.aa), '--aa-key', str(args.aa_key),
|
||||
'--aa-enc-key', str(args.aa_enc_key), '--ea', str(args.ea), '--ea-key', str(args.ea_key),
|
||||
'--ea-enc-key', str(args.ea_enc_key), '--ec-dir', str(args.dir)]
|
||||
else:
|
||||
self.send_error(404, 'only /ea/enrolment and /aa/authorization are served')
|
||||
return
|
||||
response, log = respond(tool_args, body)
|
||||
print(log.strip(), flush=True)
|
||||
if response is None:
|
||||
self.send_error(400, 'request did not decrypt/verify/decode')
|
||||
return
|
||||
self.send_response(200)
|
||||
self.send_header('Content-Type', 'application/octet-stream')
|
||||
self.send_header('Content-Length', str(len(response)))
|
||||
self.end_headers()
|
||||
self.wfile.write(response)
|
||||
|
||||
def log_message(self, fmt, *values):
|
||||
print('%s %s' % (self.address_string(), fmt % values), flush=True)
|
||||
|
||||
server = http.server.ThreadingHTTPServer((args.bind, args.port), Handler)
|
||||
print(f'EA/AA on http://{args.bind}:{args.port}/ (ea/enrolment, aa/authorization), EC store {args.dir.resolve()}', flush=True)
|
||||
try:
|
||||
server.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,98 @@
|
||||
#!/usr/bin/env python3
|
||||
"""The ITS-S side of TS 102 941 V2.2.1 clause 6.2.3, over a real HTTP round trip.
|
||||
|
||||
`vanetza_idf::pki` builds and parses the enrolment/authorization messages; per its
|
||||
own doc comment, transport to the EA/AA is "supplied by the application" (clause 6.1
|
||||
reference points S3/S4, HTTP in practice). This is that application: it shells out
|
||||
to `vidf_issue enrol-request`/`enrol-response` (or `authorize-*`) for the message
|
||||
building/parsing and does only the POST in between, the way fetch_trust_lists.py
|
||||
does only the GETs for the distribution centre.
|
||||
|
||||
pki_client.py --issue-tool build/vidf_issue --pki http://127.0.0.1:8090/ enrol \
|
||||
--ea EA.oer --canonical-key canonical.pem --its-id my-station \
|
||||
--out EC.oer --out-key EC.vkey
|
||||
pki_client.py --issue-tool build/vidf_issue --pki http://127.0.0.1:8090/ authorize \
|
||||
--ea EA.oer --aa AA.oer --ec EC.oer --ec-key EC.vkey \
|
||||
--out AT.oer --out-key AT.vkey
|
||||
|
||||
Test tool, not part of the library.
|
||||
"""
|
||||
import argparse
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def run(issue_tool, *tool_args):
|
||||
result = subprocess.run([str(issue_tool), *tool_args], capture_output=True, text=True)
|
||||
if result.stdout:
|
||||
print(result.stdout.strip(), flush=True)
|
||||
if result.returncode != 0:
|
||||
sys.exit(result.stderr.strip() or f'{tool_args[0]} failed')
|
||||
|
||||
|
||||
def post(url, body):
|
||||
request = urllib.request.Request(url, data=body, method='POST', headers={'Content-Type': 'application/octet-stream'})
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=10) as response:
|
||||
return response.read()
|
||||
except urllib.error.HTTPError as error:
|
||||
sys.exit(f'{url}: HTTP {error.code} {error.reason}: {error.read().decode(errors="replace")}')
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
parser.add_argument('--issue-tool', type=Path, required=True)
|
||||
parser.add_argument('--pki', required=True, help='EA/AA access point, e.g. http://127.0.0.1:8090/')
|
||||
sub = parser.add_subparsers(dest='command', required=True)
|
||||
|
||||
enrol = sub.add_parser('enrol')
|
||||
enrol.add_argument('--ea', required=True)
|
||||
enrol.add_argument('--canonical-key', required=True)
|
||||
enrol.add_argument('--its-id', required=True)
|
||||
enrol.add_argument('--permission', action='append', default=[])
|
||||
enrol.add_argument('--out', required=True)
|
||||
enrol.add_argument('--out-key', required=True)
|
||||
|
||||
authorize = sub.add_parser('authorize')
|
||||
authorize.add_argument('--ea', required=True)
|
||||
authorize.add_argument('--aa', required=True)
|
||||
authorize.add_argument('--ec', required=True)
|
||||
authorize.add_argument('--ec-key', required=True)
|
||||
authorize.add_argument('--permission', action='append', default=[])
|
||||
authorize.add_argument('--hours', default='24')
|
||||
authorize.add_argument('--out', required=True)
|
||||
authorize.add_argument('--out-key', required=True)
|
||||
|
||||
args = parser.parse_args()
|
||||
base = args.pki if args.pki.endswith('/') else args.pki + '/'
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
request_path, context_path, response_path = (Path(tmp) / name for name in ('request.bin', 'context.bin', 'response.bin'))
|
||||
if args.command == 'enrol':
|
||||
build_args = ['enrol-request', '--ea', args.ea, '--canonical-key', args.canonical_key, '--its-id', args.its_id,
|
||||
'--out', str(request_path), '--context', str(context_path), '--out-key', args.out_key]
|
||||
for permission in args.permission:
|
||||
build_args += ['--permission', permission]
|
||||
run(args.issue_tool, *build_args)
|
||||
response = post(base + 'ea/enrolment', request_path.read_bytes())
|
||||
response_path.write_bytes(response)
|
||||
run(args.issue_tool, 'enrol-response', '--ea', args.ea, '--context', str(context_path),
|
||||
'--response', str(response_path), '--out', args.out)
|
||||
else:
|
||||
build_args = ['authorize-request', '--ea', args.ea, '--aa', args.aa, '--ec', args.ec, '--ec-key', args.ec_key,
|
||||
'--hours', args.hours, '--out', str(request_path), '--context', str(context_path),
|
||||
'--out-key', args.out_key]
|
||||
for permission in args.permission:
|
||||
build_args += ['--permission', permission]
|
||||
run(args.issue_tool, *build_args)
|
||||
response = post(base + 'aa/authorization', request_path.read_bytes())
|
||||
response_path.write_bytes(response)
|
||||
run(args.issue_tool, 'authorize-response', '--aa', args.aa, '--context', str(context_path),
|
||||
'--response', str(response_path), '--out', args.out)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,130 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Two-C5 radio integration test, not an ETSI ATS or RF certification test.
|
||||
|
||||
Requires the test firmware on both boards and separately installed pyserial.
|
||||
Explicit --allow-transmission enables the laboratory transmitter. Capture and
|
||||
result files retain independent receiver observations.
|
||||
|
||||
FCS is not independently re-validated here, and that is not a gap: the
|
||||
receiver's promiscuous filter (vanetza_idf::C5Radio, ports/esp_idf/src/c5_radio.cpp)
|
||||
does not set WIFI_PROMIS_FILTER_MASK_FCSFAIL, so the ESP32-C5 WiFi hardware
|
||||
already discards any frame that fails FCS before this callback ever sees it
|
||||
(Espressif's own doc comment on that flag: "do not open it in general") --
|
||||
any frame reaching this test already passed a real over-the-air FCS check.
|
||||
The chip's rx_ctrl.sig_len is documented (esp_wifi_he_types.h, this chip's
|
||||
802.11ax/HE RX descriptor) as "the length of the reception MPDU", not
|
||||
"MPDU + FCS" as older ESP32 documentation for earlier chips states; measured
|
||||
directly on this hardware, sig_len is consistently 4 bytes longer than the
|
||||
actual frame content, and those 4 trailing bytes are identical
|
||||
(observed: 00 00 99 00) across captures with completely different random
|
||||
payloads -- proof they are not a content-derived FCS at all, just a fixed
|
||||
trailer this chip's HE RX descriptor appends. Recomputing a software CRC32
|
||||
over them and requiring a match, as an earlier version of this test did, can
|
||||
never succeed and was not actually checking anything real; the trailing
|
||||
bytes are still captured into the pcap for inspection, just not treated as
|
||||
a verifiable FCS.
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import secrets
|
||||
import struct
|
||||
import time
|
||||
from pathlib import Path
|
||||
from serial_sut import SerialSut, records
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--dut', required=True)
|
||||
parser.add_argument('--receiver', required=True)
|
||||
parser.add_argument('--channel', type=int, default=180)
|
||||
parser.add_argument('--out', type=Path, required=True)
|
||||
parser.add_argument('--allow-transmission', action='store_true', required=True)
|
||||
args = parser.parse_args()
|
||||
if args.dut.upper() == args.receiver.upper():
|
||||
parser.error('DUT and receiver must be different devices')
|
||||
if args.channel not in range(172, 185, 2):
|
||||
parser.error('Channel must be an even ITS-G5 channel from 172 through 184')
|
||||
args.out.mkdir(parents=True, exist_ok=False)
|
||||
result = {'kind': 'radio-integration', 'dut': args.dut, 'receiver': args.receiver,
|
||||
'channel': args.channel, 'status': 'error', 'captures': [], 'cleanup_errors': []}
|
||||
devices = []
|
||||
def accepted(device, command):
|
||||
status, observed = records(device.execute(command))
|
||||
if status:
|
||||
raise RuntimeError(f'Diagnostic command {command[0]} returned {status}')
|
||||
return observed
|
||||
try:
|
||||
dut = SerialSut(args.dut); devices.append(dut)
|
||||
receiver = SerialSut(args.receiver); devices.append(receiver)
|
||||
accepted(dut, b'\x00')
|
||||
payload = b'VIDF-RF-' + secrets.token_bytes(16)
|
||||
generated = accepted(dut, b'\x01\x01' + struct.pack('>HH', 2018, 0) + payload)
|
||||
packets = [data for kind, data in generated if kind == 1]
|
||||
if len(packets) != 1:
|
||||
raise RuntimeError('Expected exactly one real GeoNetworking packet')
|
||||
gn = packets[0]
|
||||
# Locally administered source address; independent receiver checks it.
|
||||
source = bytes.fromhex('020000000001')
|
||||
destination = b'\xff' * 6
|
||||
transmit = b'\x04' + source + destination + bytes([3, 2]) + gn
|
||||
channel = struct.pack('>H', args.channel)
|
||||
accepted(receiver, b'\x03' + channel + b'\x00')
|
||||
status, _ = records(receiver.execute(transmit))
|
||||
if status == 0:
|
||||
raise RuntimeError('Receive-only radio incorrectly accepted transmission')
|
||||
result['receive_only_tx_rejected'] = True
|
||||
accepted(dut, b'\x03' + channel + b'\x01')
|
||||
accepted(receiver, b'\x05') # Drain observations from before this transmission.
|
||||
accepted(dut, transmit)
|
||||
result['submission_accepted'] = True
|
||||
matched = False
|
||||
deadline = time.monotonic() + 5
|
||||
with (args.out / 'receiver.pcap').open('wb') as capture:
|
||||
# LINKTYPE_IEEE802_11 (105). Captured frames carry rx_ctrl.sig_len bytes,
|
||||
# which is 4 bytes longer than the actual frame content on this chip (see
|
||||
# the module docstring); the trailing 4 bytes are not a real FCS.
|
||||
capture.write(struct.pack('<IHHIIII', 0xa1b2c3d4, 2, 4, 0, 0, 65535, 105))
|
||||
while time.monotonic() < deadline and not matched:
|
||||
for kind, data in accepted(receiver, b'\x05'):
|
||||
if kind != 3 or len(data) < 5:
|
||||
raise RuntimeError('Malformed independent capture record')
|
||||
rssi, timestamp = struct.unpack('>bI', data[:5])
|
||||
frame = data[5:]
|
||||
now = time.time(); sec = int(now)
|
||||
capture.write(struct.pack('<IIII', sec, int((now-sec)*1e6), len(frame), len(frame)))
|
||||
capture.write(frame)
|
||||
# Reception through this promiscuous filter (no FCSFAIL bit) already
|
||||
# means the hardware validated FCS; content match against exactly
|
||||
# what the DUT submitted is what "independently received correctly"
|
||||
# actually means here. The trailing 4 bytes are retained for
|
||||
# inspection only, not treated as a verifiable FCS (see docstring).
|
||||
exact = (len(frame) == 38 + len(gn) and frame[:2] == b'\x88\x00'
|
||||
and frame[4:10] == destination and frame[10:16] == source
|
||||
and frame[16:22] == destination and frame[24:26] == b'\x03\x00'
|
||||
and frame[26:34] == bytes.fromhex('aaaa030000008947')
|
||||
and frame[34:-4] == gn)
|
||||
result['captures'].append({'rssi_dbm': rssi, 'device_timestamp_us': timestamp,
|
||||
'trailing_bytes': frame[-4:].hex() if len(frame) >= 4 else None,
|
||||
'exact_match': exact})
|
||||
matched |= exact
|
||||
result['status'] = 'pass' if matched else 'fail'
|
||||
if not matched:
|
||||
result['reason'] = 'No exact independent frame observed'
|
||||
except Exception as error:
|
||||
result['reason'] = str(error)
|
||||
finally:
|
||||
for device in devices:
|
||||
try:
|
||||
accepted(device, b'\x06')
|
||||
except Exception as error:
|
||||
result['cleanup_errors'].append(str(error))
|
||||
result['status'] = 'error'
|
||||
device.close()
|
||||
(args.out / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
|
||||
print(json.dumps(result, indent=2))
|
||||
raise SystemExit(0 if result['status'] == 'pass' else 1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,94 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Execute an externally installed TITAN suite and retain real verdicts.
|
||||
|
||||
The supplied configuration owns PICS/PIXIT, test ports and testcase selection.
|
||||
This runner does not install TTCN, rewrite testcases or treat process exit as PASS.
|
||||
"""
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import signal
|
||||
import subprocess
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
for name in ('titan', 'binary', 'config', 'out'):
|
||||
parser.add_argument('--' + name, type=Path, required=True)
|
||||
parser.add_argument('--sut', type=Path)
|
||||
parser.add_argument('--sut-script')
|
||||
parser.add_argument('--port')
|
||||
parser.add_argument('--sut-args', help='extra SUT arguments (VIDF_SUT_ARGS), e.g. "--security-pool ./certificates"')
|
||||
parser.add_argument('--pool', type=Path, help='certificate pool directory copied to <out>/certificates')
|
||||
parser.add_argument('--timeout', type=int, default=300)
|
||||
parser.add_argument('--expected-cases', type=Path, required=True,
|
||||
help='JSON array of the exact testcase names expected in this campaign')
|
||||
args = parser.parse_args()
|
||||
expected = json.loads(args.expected_cases.read_text())
|
||||
if (not isinstance(expected, list) or not expected or
|
||||
any(not isinstance(case, str) or not case for case in expected) or
|
||||
len(expected) != len(set(expected))):
|
||||
parser.error('--expected-cases must contain a nonempty array of unique testcase names')
|
||||
titan, binary, config, out = [p.resolve() for p in (args.titan, args.binary, args.config, args.out)]
|
||||
out.mkdir(parents=True, exist_ok=False)
|
||||
shutil.copy2(config, out / 'test.cfg')
|
||||
shutil.copy2(args.expected_cases, out / 'expected-cases.json')
|
||||
env = dict(os.environ)
|
||||
env.update(TTCN3_DIR=str(titan), LD_LIBRARY_PATH=str(titan / 'lib'),
|
||||
PATH=str(titan / 'bin') + ':' + env.get('PATH', ''))
|
||||
if args.sut:
|
||||
env['VIDF_SUT_EXECUTABLE'] = str(args.sut.resolve())
|
||||
if args.sut_script:
|
||||
env['VIDF_SUT_SCRIPT'] = args.sut_script
|
||||
if args.port:
|
||||
env['VIDF_SUT_PORT'] = args.port
|
||||
if args.sut_args:
|
||||
env['VIDF_SUT_ARGS'] = args.sut_args
|
||||
if args.pool:
|
||||
shutil.copytree(args.pool, out / 'certificates')
|
||||
sha = lambda p: hashlib.sha256(p.read_bytes()).hexdigest()
|
||||
metadata = {'started_utc': datetime.now(timezone.utc).isoformat(),
|
||||
'binary_sha256': sha(binary), 'config_sha256': sha(config),
|
||||
'sut_sha256': sha(args.sut) if args.sut else None,
|
||||
'sut_args': args.sut_args,
|
||||
'pool_sha256': {p.name: sha(p) for p in sorted((out / 'certificates').iterdir())} if args.pool else None,
|
||||
'expected_cases_sha256': sha(args.expected_cases),
|
||||
'verdicts': [], 'status': 'running'}
|
||||
with (out / 'console.txt').open('w') as log:
|
||||
process = subprocess.Popen([str(titan / 'bin/ttcn3_start'), str(binary), 'test.cfg'],
|
||||
cwd=out, env=env, stdout=log, stderr=subprocess.STDOUT,
|
||||
start_new_session=True)
|
||||
try:
|
||||
metadata['exit_code'] = process.wait(timeout=args.timeout)
|
||||
except subprocess.TimeoutExpired:
|
||||
os.killpg(process.pid, signal.SIGTERM)
|
||||
process.wait(timeout=15)
|
||||
metadata['exit_code'] = None
|
||||
metadata['status'] = 'runner-timeout'
|
||||
logs = sorted(out.glob('*mtc*.log'))
|
||||
text = '\n'.join(p.read_text(errors='replace') for p in logs) if logs else (out / 'console.txt').read_text(errors='replace')
|
||||
metadata['verdicts'] = [{'testcase': case, 'verdict': verdict.lower()} for case, verdict in
|
||||
re.findall(r'Test case (\S+) finished\. Verdict: (\w+)', text)]
|
||||
observed = [v['testcase'] for v in metadata['verdicts']]
|
||||
metadata['missing_cases'] = sorted(set(expected) - set(observed))
|
||||
metadata['unexpected_cases'] = sorted(set(observed) - set(expected))
|
||||
metadata['duplicate_cases'] = sorted({case for case in observed if observed.count(case) > 1})
|
||||
metadata['campaign_complete'] = not any(metadata[key] for key in
|
||||
('missing_cases', 'unexpected_cases', 'duplicate_cases'))
|
||||
if metadata['status'] == 'running':
|
||||
metadata['status'] = 'completed' if metadata['exit_code'] == 0 else 'runner-error'
|
||||
metadata['finished_utc'] = datetime.now(timezone.utc).isoformat()
|
||||
(out / 'result.json').write_text(json.dumps(metadata, indent=2) + '\n')
|
||||
print(json.dumps(metadata, indent=2))
|
||||
passed = metadata['status'] == 'completed' and metadata['campaign_complete'] and all(
|
||||
v['verdict'] == 'pass' for v in metadata['verdicts'])
|
||||
raise SystemExit(0 if passed else 1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,112 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Hex-line SUT process for external TTCN adapters; USB carries VID1 frames.
|
||||
|
||||
Requires separately installed pyserial. This is a test application transport,
|
||||
not a TTCN runtime, ETSI UT codec or source of test verdicts.
|
||||
"""
|
||||
import argparse
|
||||
import struct
|
||||
import sys
|
||||
import time
|
||||
import zlib
|
||||
|
||||
|
||||
class SerialSut:
|
||||
def __init__(self, port, timeout=8):
|
||||
import serial
|
||||
self.port = serial.Serial(port=None, baudrate=115200, timeout=0.1, write_timeout=timeout)
|
||||
self.port.dtr = False
|
||||
self.port.rts = False
|
||||
self.port.port = port
|
||||
self.port.open()
|
||||
self.timeout = timeout
|
||||
self.sequence = 0
|
||||
self.buffer = bytearray()
|
||||
|
||||
def close(self):
|
||||
self.port.close()
|
||||
|
||||
def execute(self, payload):
|
||||
if not 0 < len(payload) <= 4096:
|
||||
raise ValueError('Diagnostic payload must contain 1..4096 octets')
|
||||
self.sequence = (self.sequence + 1) & 0xffffffff
|
||||
frame = b'VID1' + struct.pack('>BIH', 3, self.sequence, len(payload)) + payload
|
||||
frame += struct.pack('>I', zlib.crc32(frame))
|
||||
if self.port.write(frame) != len(frame):
|
||||
raise IOError('Incomplete SUT write')
|
||||
deadline = time.monotonic() + self.timeout
|
||||
while time.monotonic() < deadline:
|
||||
self.buffer.extend(self.port.read(4096))
|
||||
while len(self.buffer) >= 4:
|
||||
if self.buffer[:4] != b'VID1':
|
||||
del self.buffer[0]
|
||||
continue
|
||||
if len(self.buffer) < 11:
|
||||
break
|
||||
channel, sequence, size = struct.unpack('>BIH', self.buffer[4:11])
|
||||
if channel not in (1, 2, 3) or not 0 < size <= 4096:
|
||||
del self.buffer[0]
|
||||
continue
|
||||
if len(self.buffer) < size + 15:
|
||||
break
|
||||
raw = bytes(self.buffer[:size + 15])
|
||||
if zlib.crc32(raw[:-4]) != struct.unpack('>I', raw[-4:])[0]:
|
||||
del self.buffer[0]
|
||||
continue
|
||||
del self.buffer[:size + 15]
|
||||
if channel == 3 and sequence == self.sequence:
|
||||
return raw[11:-4]
|
||||
raise TimeoutError('No matching device response; no verdict inferred')
|
||||
|
||||
|
||||
def records(response):
|
||||
if len(response) < 2:
|
||||
raise ValueError('Truncated SUT response')
|
||||
count = response[1]
|
||||
offset = 2
|
||||
decoded = []
|
||||
for _ in range(count):
|
||||
if offset + 3 > len(response):
|
||||
raise ValueError('Truncated record header')
|
||||
kind, size = struct.unpack_from('>BH', response, offset)
|
||||
offset += 3
|
||||
if offset + size > len(response):
|
||||
raise ValueError('Truncated record data')
|
||||
decoded.append((kind, response[offset:offset + size]))
|
||||
offset += size
|
||||
if offset != len(response):
|
||||
raise ValueError('Trailing response data')
|
||||
return response[0], decoded
|
||||
|
||||
|
||||
def provision(sut, bundle_path):
|
||||
"""Diagnostic command 9: a credential bundle (credentials.hpp) for the next reset."""
|
||||
bundle = open(bundle_path, 'rb').read()
|
||||
if len(bundle) > 4095:
|
||||
raise ValueError('The bundle exceeds one diagnostic frame (4095 octets)')
|
||||
result, _ = records(sut.execute(b' ' + bundle))
|
||||
if result != 0:
|
||||
raise RuntimeError('The device refused the credential bundle (result %d)' % result)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--port', required=True)
|
||||
parser.add_argument('--bundle', help='credential bundle sent as diagnostic command 9 before relaying stdin')
|
||||
args = parser.parse_args()
|
||||
sut = SerialSut(args.port)
|
||||
try:
|
||||
if args.bundle:
|
||||
provision(sut, args.bundle)
|
||||
for line in sys.stdin:
|
||||
if len(line) > 8193:
|
||||
raise ValueError('Oversized input line')
|
||||
response = sut.execute(bytes.fromhex(line.strip()))
|
||||
records(response)
|
||||
print(response.hex(), flush=True)
|
||||
finally:
|
||||
sut.close()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,121 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Deterministic allowlists for the IDF port, including ASN.1 dependency closure.
|
||||
|
||||
Run after an upstream merge, then review the generated diff. Unselected ITS
|
||||
services, RPC, host tools, GNSS drivers, PQC, the dummy/null crypto backend and
|
||||
the upstream PKI client tool are not compiled.
|
||||
"""
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
PORT = ROOT / 'ports/esp_idf'
|
||||
|
||||
|
||||
def write_changed(path, text):
|
||||
# Avoid forcing an IDF reconfigure when the allowlist is unchanged.
|
||||
if not path.exists() or path.read_text(encoding='utf-8') != text:
|
||||
path.write_text(text, encoding='utf-8', newline=chr(10))
|
||||
|
||||
|
||||
def module(name):
|
||||
text = (ROOT / 'vanetza' / name / 'CMakeLists.txt').read_text()
|
||||
match = re.search(r'set\(CXX_SOURCES\s+(.*?)\)', text, re.S)
|
||||
if not match:
|
||||
match = re.search(r'add_vanetza_component\(' + name + r'\s+(.*?)\)', text, re.S)
|
||||
return [f'vanetza/{name}/{s}' for s in match[1].split() if s.endswith('.cpp')]
|
||||
|
||||
|
||||
def asn_closure(directory, roots):
|
||||
base = ROOT / 'vanetza/asn1' / directory
|
||||
pending = [base / f'{name}.h' for name in roots]
|
||||
seen = set()
|
||||
sources = set()
|
||||
while pending:
|
||||
path = pending.pop()
|
||||
if path in seen:
|
||||
continue
|
||||
seen.add(path)
|
||||
source = path.with_suffix('.c')
|
||||
if source.is_file():
|
||||
sources.add(source.relative_to(ROOT).as_posix())
|
||||
pending.append(source)
|
||||
for include in re.findall(r'#include\s+"([^"]+)"', path.read_text()):
|
||||
dep = path.parent / include
|
||||
if dep.is_file() and dep.parent == base:
|
||||
pending.append(dep)
|
||||
return sorted(sources)
|
||||
|
||||
|
||||
def main():
|
||||
groups = {'base': module('access') + module('common') + module('net')}
|
||||
# Router still needs both security envelope parsers through its upstream
|
||||
# SecuredMessage variant. No dummy signer, crypto factory or PKI is linked.
|
||||
security = module('security')
|
||||
exclude = ('backend', 'persistence', 'certificate_provider', 'certificate_validator',
|
||||
'sign_service', 'sign_header_policy', 'verify_service', 'verification',
|
||||
'trust_store', 'hmac', 'sha.cpp', 'delegating_security_entity',
|
||||
'certificate_cache', 'revocation_lookup', 'issuer_memory_lookup',
|
||||
'location_checker', 'distance.cpp', 'geometry.cpp', 'hash.cpp',
|
||||
'peer_request_tracker', 'cam_ssp', 'issuer_memory')
|
||||
security = [p for p in security if not any(x in p for x in exclude)]
|
||||
groups['network'] = module('geonet') + module('btp') + module('dcc') + security + [
|
||||
'vanetza/geodesy/geodesy.cpp', 'vanetza/geodesy/haversine.cpp',
|
||||
'vanetza/security/v3/distance.cpp', 'vanetza/security/v3/geometry.cpp',
|
||||
'vanetza/security/v3/boost_geometry.cpp']
|
||||
# Signing security entity (TS 103 097 V2.2.1 / IEEE 1609.2): the upstream v3
|
||||
# sign service, certificate provider base, validator, trust anchors and
|
||||
# message-hash helpers. Backends are supplied by the port (PSA on device,
|
||||
# OpenSSL on host); the upstream backend factory is deliberately absent
|
||||
# because it registers the dummy BackendNull. DummySignService and
|
||||
# DefaultSignHeaderPolicy are compiled with their translation units but never
|
||||
# instantiated by the port; --gc-sections drops them.
|
||||
groups['security_provider'] = [
|
||||
'vanetza/security/peer_request_tracker.cpp',
|
||||
'vanetza/security/v3/certificate_cache.cpp',
|
||||
'vanetza/security/v3/certificate_validator.cpp',
|
||||
'vanetza/security/v3/hash.cpp',
|
||||
'vanetza/security/v3/issuer_memory_lookup.cpp',
|
||||
'vanetza/security/v3/location_checker.cpp',
|
||||
'vanetza/security/v3/persistence.cpp',
|
||||
'vanetza/security/v3/revocation_lookup.cpp',
|
||||
'vanetza/security/v3/sign_header_policy.cpp',
|
||||
'vanetza/security/v3/sign_service.cpp',
|
||||
'vanetza/security/v3/trust_store.cpp',
|
||||
'vanetza/geodesy/country_data_reader.cpp',
|
||||
'vanetza/geodesy/country_database.cpp']
|
||||
# Host-only OpenSSL backend and PEM/DER key loading (VANETZA_WITH_OPENSSL).
|
||||
groups['security_host'] = [
|
||||
'vanetza/security/backend_openssl.cpp',
|
||||
'vanetza/security/openssl_wrapper.cpp',
|
||||
'vanetza/security/persistence.cpp']
|
||||
groups['asn_support'] = ['vanetza/asn1/' + s for s in
|
||||
(ROOT / 'vanetza/asn1/asn1c_support_sources.txt').read_text().splitlines()
|
||||
if s and not s.startswith('#')]
|
||||
groups['asn_support'] += ['vanetza/asn1/memory.c', 'vanetza/asn1/asn1c_wrapper.cpp',
|
||||
'vanetza/asn1/type_traits.cpp']
|
||||
groups['security_asn'] = asn_closure('security', ['EtsiTs103097Data', 'Certificate', 'EtsiTs103097Certificate'])
|
||||
# TS 102 941 V2.2.1 Annex A.2 request/response messages (VIDF_PKI), minus what security_asn has.
|
||||
groups['pki_asn'] = sorted(set(asn_closure('security', ['EtsiTs102941Data', 'InnerEcRequest'])) - set(groups['security_asn']))
|
||||
for service in ('cam', 'denm', 'vam'):
|
||||
groups[service] = asn_closure('its/r2', [service.upper()])
|
||||
cmake = '# Generated by tools/source_manifest.py. Review changes after upstream updates.\n'
|
||||
for name, paths in groups.items():
|
||||
cmake += f'set(VIDF_{name.upper()}_SOURCES\n'
|
||||
cmake += ''.join(f' "${{VIDF_ROOT}}/{p}"\n' for p in sorted(set(paths)))
|
||||
cmake += ')\n'
|
||||
write_changed(PORT / 'sources.cmake', cmake)
|
||||
manifest = {'upstream': 'https://github.com/riebl/vanetza',
|
||||
'commit': 'a7cacc1879f3e2124dfb8ef2d83886b1e82d3b36',
|
||||
'license': 'LGPL-3.0-or-later; see LICENSE.md and per-file notices',
|
||||
'groups': {k: sorted(set(v)) for k, v in groups.items()},
|
||||
'sha256': {p: hashlib.sha256((ROOT / p).read_bytes()).hexdigest()
|
||||
for p in sorted({p for v in groups.values() for p in v})}}
|
||||
write_changed(PORT / 'source-manifest.json', json.dumps(manifest, indent=2) + '\n')
|
||||
print({name: len(paths) for name, paths in groups.items()})
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user