OTA Security & Secure Update

#!/usr/bin/env python3
# UNECE R156 SUMS: OTA package verification
import hashlib, struct, json
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec

def verify_ota_package(pkg: bytes, expected_sha256: str,
                       signature: bytes, oem_pub_key,
                       current_ecu_version: int) -> dict:
    result = {}

    # 1. Hash integrity
    actual = hashlib.sha256(pkg).hexdigest()
    result["hash_ok"] = (actual == expected_sha256)

    # 2. ECDSA-P256 authenticity
    try:
        oem_pub_key.verify(signature, pkg, ec.ECDSA(hashes.SHA256()))
        result["sig_ok"] = True
    except Exception:
        result["sig_ok"] = False

    # 3. Anti-rollback: package version >= current ECU version
    new_version = struct.unpack(">I", pkg[4:8])[0]
    result["anti_rollback_ok"] = (new_version >= current_ecu_version)
    result["new_version"] = hex(new_version)
    result["current_version"] = hex(current_ecu_version)

    # 4. Final install decision (R156: all three must pass)
    result["install"] = result["hash_ok"] and result["sig_ok"] and result["anti_rollback_ok"]

    return result

# Fail-safe: if any check fails, OTA client must:
# 1. Log DEM event: DEM_EVENT_OTA_VERIFY_FAIL
# 2. Discard the package
# 3. Remain on current firmware
# 4. Report failure to OEM OTA backend for telemetry

#!/usr/bin/env python3
# OTA client: TLS 1.3 + mTLS download with resumption support
import ssl, urllib.request, hashlib

def download_ota_package(url: str,
                         ca_cert:      str = "oem_root_ca.pem",
                         client_cert:  str = "vehicle_ecu_cert.pem",
                         client_key:   str = "vehicle_ecu_key.pem") -> bytes:

    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.minimum_version = ssl.TLSVersion.TLSv1_3  # R156: TLS 1.3 minimum
    ctx.verify_mode = ssl.CERT_REQUIRED
    ctx.load_verify_locations(ca_cert)             # verify OEM backend cert
    ctx.load_cert_chain(client_cert, client_key)  # present vehicle's mTLS cert

    handler = urllib.request.HTTPSHandler(context=ctx)
    opener  = urllib.request.build_opener(handler)

    req = urllib.request.Request(url)
    req.add_header("X-Vehicle-VIN", "WBA12345678901234")
    req.add_header("X-SW-Version",  "0x0302")  # current version; server validates eligibility

    with opener.open(req, timeout=300) as resp:
        package = resp.read()

    print(f"Downloaded: {len(package)} bytes")
    print(f"SHA-256: {hashlib.sha256(package).hexdigest()}")
    return package

# Network failure handling:
# - TLS connection drop: retry with exponential backoff (max 3 attempts)
# - Partial download: verify partial SHA-256 before resuming (Range header)
# - Backend unavailable: remain on current firmware; retry next ignition cycle