"""Verify the signature of secured GeoNetworking packets against configured trust anchors. `secured.py` only unwraps the TS 103 097 envelope and says "signed, not verified". With `--trust` the bridge also checks, per packet, whether the signature holds and whether the signer chains to a trusted root, and the dashboard shows which. Trust comes from files the operator names: a `VCR1` credential bundle (vanetza_idf/credentials.hpp, e.g. the MicrOBU demo chain) or COER certificates (`.oer`). Nothing is trusted by default, so without `--trust` nothing changes. What is checked (IEEE Std 1609.2 clause 5.3.1, ECDSA NIST P-256 with SHA-256): * the message signature over Hash(tbsData) || Hash(signer certificate); * a signer given as a full certificate must be signed by a trusted authority, and is then remembered, so later messages that only carry its digest can be checked too; * the configured chain itself at start-up: roots self-signed, authorities signed by a root, tickets by an authority. A certificate whose signature fails is not trusted. Not checked: validity periods, regions, permissions against the PSID, revocation. A packet this says is "verified" was signed by a key the configured chain vouches for, nothing more. ASN.1: IEEE1609dot2.asn / IEEE1609dot2BaseTypes.asn in asn1/, from vanetza-idf. ECDSA: the `cryptography` package (OpenSSL). """ from __future__ import annotations import hashlib import logging import struct from pathlib import Path VERIFIED, FAILED, UNKNOWN = True, False, None def hashed_id8(octets: bytes) -> bytes: return hashlib.sha256(octets).digest()[-8:] def _bundle_certificates(data: bytes) -> list[bytes]: """Certificates of a VCR1 bundle: records [type 1][length 2 BE][payload]; types 1-3 are certificates.""" out, i = [], 4 if data[:4] == b"VCR1" else 0 while i + 3 <= len(data): kind, length = data[i], struct.unpack(">H", data[i + 1:i + 3])[0] if kind in (1, 2, 3): out.append(data[i + 3:i + 3 + length]) i += 3 + length return out class TrustStore: def __init__(self, paths: list[str]): import asn1tools directory = Path(__file__).resolve().parent / "asn1" spec = asn1tools.compile_files([str(directory / "IEEE1609dot2.asn"), str(directory / "IEEE1609dot2BaseTypes.asn")], "oer") self._m = spec.modules["IEEE1609dot2"] certificates = [] for path in paths: data = Path(path).read_bytes() certificates += _bundle_certificates(data) if data[:4] == b"VCR1" else [data] self.issuers: dict[bytes, bytes] = {} # digest -> roots and authorities whose signature held self.signers: dict[bytes, bytes] = {} # digest -> tickets (and certificates learned on air) pending = list(certificates) # Roots first, then whatever they vouch for, until nothing more can be placed. while pending: placed = False for cert in list(pending): verdict = self._place(cert) if verdict is not None: pending.remove(cert) placed = True if not placed: break for cert in pending: logging.warning("trust: %s not trusted (issuer unknown or signature fails)", hashed_id8(cert).hex().upper()) logging.info("trust: %d root/authority certificate(s), %d signer(s): %s", len(self.issuers), len(self.signers), ", ".join(d.hex().upper() for d in list(self.issuers) + list(self.signers))) # ---- certificates -------------------------------------------------------------------- def _decode(self, cert: bytes) -> dict: return self._m["Certificate"].decode(cert) def _is_authority(self, decoded: dict) -> bool: return "certIssuePermissions" in decoded["toBeSigned"] def _place(self, cert: bytes): """Trusts cert if its signature holds under a trusted issuer (or itself, for a root). None if not yet.""" decoded = self._decode(cert) kind, issuer = decoded["issuer"] if kind == "self": issuer_cert = None elif kind in ("sha256AndDigest",): issuer_cert = self.issuers.get(bytes(issuer)) if issuer_cert is None: return None else: return None if not self._certificate_signature_ok(cert, decoded, issuer_cert): return False target = self.issuers if kind == "self" or self._is_authority(decoded) else self.signers target[hashed_id8(cert)] = cert return True def _public_key(self, decoded: dict): from cryptography.hazmat.primitives.asymmetric import ec kind, key = decoded["toBeSigned"]["verifyKeyIndicator"] if kind != "verificationKey" or key[0] != "ecdsaNistP256": raise ValueError("verification key is not ECDSA NIST P-256") form, point = key[1] if form == "compressed-y-0": encoded = b"\x02" + point elif form == "compressed-y-1": encoded = b"\x03" + point elif form == "uncompressedP256": encoded = b"\x04" + point["x"] + point["y"] else: raise ValueError("unsupported point form " + form) return ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), encoded) @staticmethod def _ecdsa_ok(public_key, message: bytes, signature) -> bool: from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature kind, sig = signature if kind != "ecdsaNistP256Signature": return False _, r = sig["rSig"] r_x = r if isinstance(r, (bytes, bytearray)) else r["x"] # x-only or compressed: r is x der = encode_dss_signature(int.from_bytes(r_x, "big"), int.from_bytes(sig["sSig"], "big")) try: public_key.verify(der, message, ec.ECDSA(hashes.SHA256())) return True except InvalidSignature: return False def _certificate_signature_ok(self, cert: bytes, decoded: dict, issuer_cert: bytes | None) -> bool: if "signature" not in decoded: return False tbs = self._m["ToBeSignedCertificate"].encode(decoded["toBeSigned"]) signer_input = hashlib.sha256(issuer_cert if issuer_cert is not None else b"").digest() key = self._public_key(self._decode(issuer_cert) if issuer_cert is not None else decoded) return self._ecdsa_ok(key, hashlib.sha256(tbs).digest() + signer_input, decoded["signature"]) # ---- messages ------------------------------------------------------------------------ def verify(self, envelope: bytes) -> dict: """{"verified": True | False | None, "reason": str, "signer_id": hex | None} for one Ieee1609Dot2Data.""" try: data = self._m["Ieee1609Dot2Data"].decode(envelope) kind, signed = data["content"] if kind != "signedData": return {"verified": UNKNOWN, "reason": "not signedData"} signer_kind, signer = signed["signer"] if signer_kind == "digest": digest = bytes(signer) cert = self.signers.get(digest) if cert is None: return {"verified": UNKNOWN, "reason": "signer not known here", "signer_id": digest.hex().upper()} elif signer_kind == "certificate": cert = self._m["Certificate"].encode(signer[0]) digest = hashed_id8(cert) if digest not in self.signers: placed = self._place(cert) if placed is None: return {"verified": UNKNOWN, "reason": "issuer not trusted here", "signer_id": digest.hex().upper()} if placed is False: return {"verified": FAILED, "reason": "certificate signature fails", "signer_id": digest.hex().upper()} else: return {"verified": UNKNOWN, "reason": "self-signed message"} tbs = self._m["ToBeSignedData"].encode(signed["tbsData"]) message = hashlib.sha256(tbs).digest() + hashlib.sha256(cert).digest() ok = self._ecdsa_ok(self._public_key(self._decode(cert)), message, signed["signature"]) return {"verified": VERIFIED if ok else FAILED, "reason": "signature valid" if ok else "signature does not verify", "signer_id": digest.hex().upper()} except Exception as exc: # a packet this cannot decode is unknown, never "verified" return {"verified": UNKNOWN, "reason": "not checkable: %s" % exc}