#!/usr/bin/env python3 from __future__ import annotations import argparse import base64 import hashlib import os import re import shutil import ssl import subprocess import sys import tempfile import zipfile from collections import defaultdict from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from functools import lru_cache from pathlib import Path from typing import Any, Iterable from urllib.parse import urlparse from xml.etree import ElementTree as ET SAFE_ENTRY_PATTERN = re.compile(r"^[A-Za-z0-9._-]+$") HEX_PAIR_PATTERN = re.compile(r"%([0-9A-Fa-f]{2})") OPENSSL_TEXT_FIELD_PATTERN = re.compile(r"^(Policy OID|Hash Algorithm|Serial number|Time stamp):\s*(.*)$") OPENSSL_HEXDUMP_LINE_PATTERN = re.compile(r"^\s*[0-9A-Fa-f]{4,}\s*-\s*(.+)$") OPENSSL_SECTION_PATTERN = re.compile(r"^\[\s*([^\]]+)\s*\]$") OPENSSL_ASSIGNMENT_PATTERN = re.compile(r"^([A-Za-z][A-Za-z0-9_.-]*)\s*=\s*(.*?)\s*$") OID_DOTTED_PATTERN = re.compile(r"^\d+(?:\.\d+)+$") MAX_ENTRY_NAME_LENGTH = 180 NETLOG_PRE_IGNITION_WARNING_TOLERANCE = timedelta(minutes=1) LAYOUT_GRAPHIC_RESOURCE_EXTENSIONS = { ".apng", ".avif", ".bmp", ".css", ".eot", ".gif", ".ico", ".jpeg", ".jpg", ".otf", ".png", ".svg", ".ttf", ".webp", ".woff", ".woff2", } @dataclass class ValidationMessage: level: str text: str @dataclass class TsaValidationOutcome: ok: bool label: str message: str gen_time: datetime | None = None # Root certificates of the RFC 3161 timestamp authorities Pemarks uses (freeTSA, DFN, Microsoft, Certum, # Sectigo/USERTrust, DigiCert). A TSA token must chain to one of these, to a root of the operating-system CA # bundle, or to a root given with --tsa-trusted-roots. The chain stored inside the extract report is never # trusted by itself (it could have been forged together with the report) unless --tsa-trust-report-chain. EMBEDDED_TSA_ROOTS_PEM = """ -----BEGIN CERTIFICATE----- MIIH/zCCBeegAwIBAgIJAMHphhYNqOmAMA0GCSqGSIb3DQEBDQUAMIGVMREwDwYD VQQKEwhGcmVlIFRTQTEQMA4GA1UECxMHUm9vdCBDQTEYMBYGA1UEAxMPd3d3LmZy ZWV0c2Eub3JnMSIwIAYJKoZIhvcNAQkBFhNidXNpbGV6YXNAZ21haWwuY29tMRIw EAYDVQQHEwlXdWVyemJ1cmcxDzANBgNVBAgTBkJheWVybjELMAkGA1UEBhMCREUw HhcNMTYwMzEzMDE1MjEzWhcNNDEwMzA3MDE1MjEzWjCBlTERMA8GA1UEChMIRnJl ZSBUU0ExEDAOBgNVBAsTB1Jvb3QgQ0ExGDAWBgNVBAMTD3d3dy5mcmVldHNhLm9y ZzEiMCAGCSqGSIb3DQEJARYTYnVzaWxlemFzQGdtYWlsLmNvbTESMBAGA1UEBxMJ V3VlcnpidXJnMQ8wDQYDVQQIEwZCYXllcm4xCzAJBgNVBAYTAkRFMIICIjANBgkq hkiG9w0BAQEFAAOCAg8AMIICCgKCAgEAtgKODjAy8REQ2WTNqUudAnjhlCrpE6ql mQfNppeTmVvZrH4zutn+NwTaHAGpjSGv4/WRpZ1wZ3BRZ5mPUBZyLgq0YrIfQ5Fx 0s/MRZPzc1r3lKWrMR9sAQx4mN4z11xFEO529L0dFJjPF9MD8Gpd2feWzGyptlel b+PqT+++fOa2oY0+NaMM7l/xcNHPOaMz0/2olk0i22hbKeVhvokPCqhFhzsuhKsm q4Of/o+t6dI7sx5h0nPMm4gGSRhfq+z6BTRgCrqQG2FOLoVFgt6iIm/BnNffUr7V DYd3zZmIwFOj/H3DKHoGik/xK3E82YA2ZulVOFRW/zj4ApjPa5OFbpIkd0pmzxzd EcL479hSA9dFiyVmSxPtY5ze1P+BE9bMU1PScpRzw8MHFXxyKqW13Qv7LWw4sbk3 SciB7GACbQiVGzgkvXG6y85HOuvWNvC5GLSiyP9GlPB0V68tbxz4JVTRdw/Xn/XT FNzRBM3cq8lBOAVt/PAX5+uFcv1S9wFE8YjaBfWCP1jdBil+c4e+0tdywT2oJmYB BF/kEt1wmGwMmHunNEuQNzh1FtJY54hbUfiWi38mASE7xMtMhfj/C4SvapiDN837 gYaPfs8x3KZxbX7C3YAsFnJinlwAUss1fdKar8Q/YVs7H/nU4c4Ixxxz4f67fcVq M2ITKentbCMCAwEAAaOCAk4wggJKMAwGA1UdEwQFMAMBAf8wDgYDVR0PAQH/BAQD AgHGMB0GA1UdDgQWBBT6VQ2MNGZRQ0z357OnbJWveuaklzCBygYDVR0jBIHCMIG/ gBT6VQ2MNGZRQ0z357OnbJWveuakl6GBm6SBmDCBlTERMA8GA1UEChMIRnJlZSBU U0ExEDAOBgNVBAsTB1Jvb3QgQ0ExGDAWBgNVBAMTD3d3dy5mcmVldHNhLm9yZzEi MCAGCSqGSIb3DQEJARYTYnVzaWxlemFzQGdtYWlsLmNvbTESMBAGA1UEBxMJV3Vl cnpidXJnMQ8wDQYDVQQIEwZCYXllcm4xCzAJBgNVBAYTAkRFggkAwemGFg2o6YAw MwYDVR0fBCwwKjAooCagJIYiaHR0cDovL3d3dy5mcmVldHNhLm9yZy9yb290X2Nh LmNybDCBzwYDVR0gBIHHMIHEMIHBBgorBgEEAYHyJAEBMIGyMDMGCCsGAQUFBwIB FidodHRwOi8vd3d3LmZyZWV0c2Eub3JnL2ZyZWV0c2FfY3BzLmh0bWwwMgYIKwYB BQUHAgEWJmh0dHA6Ly93d3cuZnJlZXRzYS5vcmcvZnJlZXRzYV9jcHMucGRmMEcG CCsGAQUFBwICMDsaOUZyZWVUU0EgdHJ1c3RlZCB0aW1lc3RhbXBpbmcgU29mdHdh cmUgYXMgYSBTZXJ2aWNlIChTYWFTKTA3BggrBgEFBQcBAQQrMCkwJwYIKwYBBQUH MAGGG2h0dHA6Ly93d3cuZnJlZXRzYS5vcmc6MjU2MDANBgkqhkiG9w0BAQ0FAAOC AgEAaK9+v5OFYu9M6ztYC+L69sw1omdyli89lZAfpWMMh9CRmJhM6KBqM/ipwoLt nxyxGsbCPhcQjuTvzm+ylN6VwTMmIlVyVSLKYZcdSjt/eCUN+41K7sD7GVmxZBAF ILnBDmTGJmLkrU0KuuIpj8lI/E6Z6NnmuP2+RAQSHsfBQi6sssnXMo4HOW5gtPO7 gDrUpVXID++1P4XndkoKn7Svw5n0zS9fv1hxBcYIHPPQUze2u30bAQt0n0iIyRLz aWuhtpAtd7ffwEbASgzB7E+NGF4tpV37e8KiA2xiGSRqT5ndu28fgpOY87gD3ArZ DctZvvTCfHdAS5kEO3gnGGeZEVLDmfEsv8TGJa3AljVa5E40IQDsUXpQLi8G+UC4 1DWZu8EVT4rnYaCw1VX7ShOR1PNCCvjb8S8tfdudd9zhU3gEB0rxdeTy1tVbNLXW 99y90xcwr1ZIDUwM/xQ/noO8FRhm0LoPC73Ef+J4ZBdrvWwauF3zJe33d4ibxEcb 8/pz5WzFkeixYM2nsHhqHsBKw7JPouKNXRnl5IAE1eFmqDyC7G/VT7OF669xM6hb Ut5G21JE4cNK6NNucS+fzg1JPX0+3VhsYZjj7D5uljRvQXrJ8iHgr/M6j2oLHvTA I2MLdq2qjZFDOCXsxBxJpbmLGBx9ow6ZerlUxzws2AWv2pk= -----END CERTIFICATE----- -----BEGIN CERTIFICATE----- MIIGBDCCA+ygAwIBAgIBATANBgkqhkiG9w0BAQsFADCBkjELMAkGA1UEBhMCREUx RTBDBgNVBAoMPFZlcmVpbiB6dXIgRm9lcmRlcnVuZyBlaW5lcyBEZXV0c2NoZW4g Rm9yc2NodW5nc25ldHplcyBlLiBWLjEQMA4GA1UECwwHREZOLVBLSTEqMCgGA1UE AwwhREZOLVZlcmVpbiBDb21tdW5pdHkgUm9vdCBDQSAyMDIyMB4XDTIyMDEyNjE0 MDg0MVoXDTQyMDEyMTE0MDg0MVowgZIxCzAJBgNVBAYTAkRFMUUwQwYDVQQKDDxW ZXJlaW4genVyIEZvZXJkZXJ1bmcgZWluZXMgRGV1dHNjaGVuIEZvcnNjaHVuZ3Nu ZXR6ZXMgZS4gVi4xEDAOBgNVBAsMB0RGTi1QS0kxKjAoBgNVBAMMIURGTi1WZXJl aW4gQ29tbXVuaXR5IFJvb3QgQ0EgMjAyMjCCAiIwDQYJKoZIhvcNAQEBBQADggIP ADCCAgoCggIBAMF5dV4jPdSCX8uV5h3T9HnSQzfoSK7dw0F7xqa8RvM9sqT8le/C rrgHKx96DXtYLCiqXtRIy7Pt/5oePzJ0K6UkmvFpoW0Fg5MxDKKCE7rvWSvk9U8q 3OBYky+X5bz5+OjdLQF6LPBcNfBDrLdO+DM71/F2uDXDjwd0XCCAMTGDz2cwGbtF 447MsWdDUu3VdP9YMSYwWBi+Pj2qric6X8J6s7XTKTBnSxSNulAHtOlUjSY9Pgy/ YWkn95BUpMT/jVfeKbQrWCpCptfG7KeaM08D11VhaRIfU/LVDCQef57qdkKDh2iB wCWvEXQMqMn18uPGgh50XCuDdlyEoKAlxWc74UQ615Q2iaLghvKCnYq5wzFOaIu9 /zLyfKURALE2c7HNY0To6CejbLRKCWF3Kkr6qZy3EtmcYxrFpEIBqJ9QYFDX4k8h k7KIqR2/vYUFS5ZX1GLBpCqaf/YkHAUaj1iqqYc8/l5SHXquwgmQK1zhpB8zwubw 5qJWFkZGO0HeExKlJosSRWhMg471mevYVm5WjIMbF7Pl6jXeXyfGTuwbZj7D3CCk Nyr52iviwQWHNs1KV9YOtE11H17R/H4dWXt1OuVEHkBUx4Vjde3V3rA5MivbDL/Y XpIflBIBMzuXhbzYril4zoPnkVKONvW2JARWxUnWFJ496jroYeGg42TfAgMBAAGj YzBhMB0GA1UdDgQWBBQQ9q+j+vYWkNDqIUESlxH4m6ValDAfBgNVHSMEGDAWgBQQ 9q+j+vYWkNDqIUESlxH4m6ValDAPBgNVHRMBAf8EBTADAQH/MA4GA1UdDwEB/wQE AwIBBjANBgkqhkiG9w0BAQsFAAOCAgEAS+nQQGCAkg1LphkSTZO89y5s4ORNAjc6 C9B12l3bJIvzJXZlaDzjXDGDgd/bjXrMwIprxRo/LZSSrwWcHa+HPb+RTEKRzyI4 NLUwaJUvTyxA320OugZf+x8qlyj2HHeX46M/s6ocxrU8OX+Qjf1YhkgZQH9fyMSq CucD/BsYrDCCPATg8iU2/VaRHJ5y4ICEzAjxFDyTioKeSneTfhciYMa1COL8okEX 0T0keGx+hO9IL7UYoDdqQGpsRMQzv9OgcgJvP8CSqalE/zG84eU11awnxQHB7bz7 HV3eTVnyNg0TdFU+SMutiWa80Kf006aj5Q1Nf94kmOjN5yXe7fUniMScwaEv74oZ hcWPJhIjLbDdwImBHB6gI5c1t4uEGZj4o0OryzVKGJeA1BvyfF2ryROG2eJN45mE g2mIVLwy6OI/3fKxBsGzLoz/7YrnTSc4W5UtM34LZAZLVua3brTDHu2iiapiXFiE O89xT8m1mcQJwBYH3c7j6OUwTfHKe1EpCoJNMhdRuy7HSDO39gKKI04y/pa/qzBj 2f0FlXbzD10nv8u3hBorxoEhqA3tYi3FQQwPuo6m7qOXkfPD13PRUY9/Rv/skpRY cMJ/7jZQk6CYR7Tl9i/ipXT5B01j8HQEejv4B1ir/CqC156QweVbDbJ0Kv1rPCWC gUikjM/huUU= -----END CERTIFICATE----- -----BEGIN CERTIFICATE----- MIIFzDCCA7SgAwIBAgIQVJjS0dRbGZVIE3nIEcCHmTANBgkqhkiG9w0BAQwFADB3 MQswCQYDVQQGEwJVUzEeMBwGA1UEChMVTWljcm9zb2Z0IENvcnBvcmF0aW9uMUgw RgYDVQQDEz9NaWNyb3NvZnQgSWRlbnRpdHkgVmVyaWZpY2F0aW9uIFJvb3QgQ2Vy dGlmaWNhdGUgQXV0aG9yaXR5IDIwMjAwHhcNMjAwNDE2MTgzNjE2WhcNNDUwNDE2 MTg0NDQwWjB3MQswCQYDVQQGEwJVUzEeMBwGA1UEChMVTWljcm9zb2Z0IENvcnBv cmF0aW9uMUgwRgYDVQQDEz9NaWNyb3NvZnQgSWRlbnRpdHkgVmVyaWZpY2F0aW9u IFJvb3QgQ2VydGlmaWNhdGUgQXV0aG9yaXR5IDIwMjAwggIiMA0GCSqGSIb3DQEB AQUAA4ICDwAwggIKAoICAQCzkSoHgwZn/Z6d4MfAt6TmQgR/D6bbX/vVWtdFoPt3 C/CA86ZtWk15U9ighoRXRSDHolT7x6K/isduNfOiFcQvTuNKhZZJDf++mdgU9rwn B+5Cmyv1C5IG5P1pE2WokXLymITrgz0O5NdxEkghyw3t9kdJt5v5yccXtoRP/7is mtdzZ0mF44a9N0DQJYbU3rXCbWJq1al4vC1vSfnlbBQU/RTH02UWN97LbrxeKY39 YpsVLNYF5rmJMjOjYsfX1lJnCMQu9FYrnguHzOyntKaq6wXNGVelOgsEJxyRZ54t Yi0vHr7awCDLBBnKM/uJvpjicqByNb554ZyDb+RtF2+Q8z0AhnU4jtDgSZq729P4 MMrVV4hoTXLTv21/cdj9vQ2ukmRIt1tveSa1zZuVIYTR7w8yPXtXjPNFB0x84F4Y DjV2i22eyzZ0qwX44HNdMlaUZ5clCsY1PZSX58FEi4D9wfj0dBnlMPYG+yFXPgYc i2sVhidJe4KTylnodUfoPzj0x1N5oLa04lxR771fOMET5ngMlVouxUBZKMwPJMDs ugl3I5k4prYc2se6ILbXN9h/N68I4ztx225zG32ZcrDkhjNZdLUWAHtQbcaGE9r9 xDmCPSQAmmDaupTABVEsNKxQmROHu7MFgLJNMAJcuCaDXbRjc++uI5VPYCi+N9Vb pQIDAQABo1QwUjAOBgNVHQ8BAf8EBAMCAYYwDwYDVR0TAQH/BAUwAwEB/zAdBgNV HQ4EFgQUyH7SaoUqG8oZmAQHJ89QEE9oqKIwEAYJKwYBBAGCNxUBBAMCAQAwDQYJ KoZIhvcNAQEMBQADggIBAK9q3eYZ5y2UQxlOy+lQlWSlA5ECi+I2gDsVolLCFhm2 alpddEMw9Jv/YHQJsSEekBZtxSSPXGaIY/RPzH3yEkxAEIsBn9qpyK7ylRvPnQXr ST50oGhb5VYsZRyCflPaVtlGF3mSRcQQNghSKRfLL6byftRpJIoej7BzDcwcSquy qu2nkWMBZCKoMrh+MiizZ3MtkbTcMQEL90cKpvHXSu1WYMQsCKN7QLC8dCdSh9a+ iN03ioluZ4gd9cldoP62qzqA1xqXPBc2IkEerE3Vg+Y8OL1PMOlUqdO2BMMydmG7 sBjFKxizwIDVt5WwXlFNIvzsWKro2JS0pS7tkt7nGHwhV91VY/e/bc0f0qZ3KHDH 4ls6WwjSW07IAJaz4YM2r4YKZVx09ursemp0oPBL7u+Uo6xQ8oft1zowg8n7fVe+ 5eP4QcrlZK6zo+xY7IWazO+56vNWGLlcc5qvxXcXg1nbNxoYclSlQdK2I3WjQ5rl d3yWebdBjb/s3ICgn9F3dVhfNRPgJRpnC33OJfoHCuRhIdjUHOUHxjaZ9JbQxhX+ Ts3Xroud2xb9BMaSvdSI5qmjqrv3ZDg7X8wM0DW+dBkDpsWqTKJhNoI+HfMrvJdd 20t4Oy31O+9gI+j17AsjNpWvmGa/U9N7uGlKKpZmacSUxvRfbqyYeIiABlyisu2i -----END CERTIFICATE----- -----BEGIN CERTIFICATE----- MIIDuzCCAqOgAwIBAgIDBETAMA0GCSqGSIb3DQEBBQUAMH4xCzAJBgNVBAYTAlBM MSIwIAYDVQQKExlVbml6ZXRvIFRlY2hub2xvZ2llcyBTLkEuMScwJQYDVQQLEx5D ZXJ0dW0gQ2VydGlmaWNhdGlvbiBBdXRob3JpdHkxIjAgBgNVBAMTGUNlcnR1bSBU cnVzdGVkIE5ldHdvcmsgQ0EwHhcNMDgxMDIyMTIwNzM3WhcNMjkxMjMxMTIwNzM3 WjB+MQswCQYDVQQGEwJQTDEiMCAGA1UEChMZVW5pemV0byBUZWNobm9sb2dpZXMg Uy5BLjEnMCUGA1UECxMeQ2VydHVtIENlcnRpZmljYXRpb24gQXV0aG9yaXR5MSIw IAYDVQQDExlDZXJ0dW0gVHJ1c3RlZCBOZXR3b3JrIENBMIIBIjANBgkqhkiG9w0B AQEFAAOCAQ8AMIIBCgKCAQEA4/t9o3K6wvDJFIf1awFO4W5AB7ptJ11/91sts1rH UV+rpDKmYYe2bg+G0jACl/jXaVehGDldamR5xgFZrDwxSjh80gTSSyjoIF87B6LM TXPb865Px1bVWqeWifrzq2jUI4ZZJ88JJ7ysbnKDHDBy3+Ci6dLhdHUZvSqeexVU BBvXQzmtVSjF4hq79MDkrjhJM8x2hZ85RdKknvISjFH4fOQtf/WsX+sWn7Et0brM kUJ3TCXJkDhv2/DM+44el1k+1WBO5gUo7Ul5E0u6SNsv+XLTOcr+H9g0cvW0QM8x AcPs3hEtF10fuFDRXhmnad4HMyjKUJX5p1TLVIZQRan5SQIDAQABo0IwQDAPBgNV HRMBAf8EBTADAQH/MB0GA1UdDgQWBBQIds3LB/8k9sXN7buQvOKEN0Z19zAOBgNV HQ8BAf8EBAMCAQYwDQYJKoZIhvcNAQEFBQADggEBAKaorSLOAT2mo/9i0Eidi15y sHhE49wcrwn9I0j6vSrEuVUEtRCjjSfeC4Jj0O7eDDd5QVsisrCaQVymcODU0HfL I9MA4GxWL+FpDQ3Zqr8hgVDZBqWo/5U30Kr+4rP1mS1FhIrlQgnXdAIv94nYmem8 J9RHjboNRhx3zxSkHLmkMcScKHQDNP8zGSal6Q10tz6XxnboJ5ajZt3hrvJBW8qY VoNzcOSGGtIxQbovvi0TWnZvTuhOgQ4/WwMioBK+ZlgRSssDxLQqKi2WF+A5VLxI 03YnnZotBqbJ7DnSq9ufmgsnAjUpsUCV5/nonFWIGUbWtzT1fs45mtk48VH3Tyw= -----END CERTIFICATE----- -----BEGIN CERTIFICATE----- MIIF3jCCA8agAwIBAgIQAf1tMPyjylGoG7xkDjUDLTANBgkqhkiG9w0BAQwFADCB iDELMAkGA1UEBhMCVVMxEzARBgNVBAgTCk5ldyBKZXJzZXkxFDASBgNVBAcTC0pl cnNleSBDaXR5MR4wHAYDVQQKExVUaGUgVVNFUlRSVVNUIE5ldHdvcmsxLjAsBgNV BAMTJVVTRVJUcnVzdCBSU0EgQ2VydGlmaWNhdGlvbiBBdXRob3JpdHkwHhcNMTAw MjAxMDAwMDAwWhcNMzgwMTE4MjM1OTU5WjCBiDELMAkGA1UEBhMCVVMxEzARBgNV BAgTCk5ldyBKZXJzZXkxFDASBgNVBAcTC0plcnNleSBDaXR5MR4wHAYDVQQKExVU aGUgVVNFUlRSVVNUIE5ldHdvcmsxLjAsBgNVBAMTJVVTRVJUcnVzdCBSU0EgQ2Vy dGlmaWNhdGlvbiBBdXRob3JpdHkwggIiMA0GCSqGSIb3DQEBAQUAA4ICDwAwggIK AoICAQCAEmUXNg7D2wiz0KxXDXbtzSfTTK1Qg2HiqiBNCS1kCdzOiZ/MPans9s/B 3PHTsdZ7NygRK0faOca8Ohm0X6a9fZ2jY0K2dvKpOyuR+OJv0OwWIJAJPuLodMkY tJHUYmTbf6MG8YgYapAiPLz+E/CHFHv25B+O1ORRxhFnRghRy4YUVD+8M/5+bJz/ Fp0YvVGONaanZshyZ9shZrHUm3gDwFA66Mzw3LyeTP6vBZY1H1dat//O+T23LLb2 VN3I5xI6Ta5MirdcmrS3ID3KfyI0rn47aGYBROcBTkZTmzNg95S+UzeQc0PzMsNT 79uq/nROacdrjGCT3sTHDN/hMq7MkztReJVni+49Vv4M0GkPGw/zJSZrM233bkf6 c0Plfg6lZrEpfDKEY1WJxA3Bk1QwGROs0303p+tdOmw1XNtB1xLaqUkL39iAigmT Yo61Zs8liM2EuLE/pDkP2QKe6xJMlXzzawWpXhaDzLhn4ugTncxbgtNMs+1b/97l c6wjOy0AvzVVdAlJ2ElYGn+SNuZRkg7zJn0cTRe8yexDJtC/QV9AqURE9JnnV4ee UB9XVKg+/XRjL7FQZQnmWEIuQxpMtPAlR1n6BB6T1CZGSlCBst6+eLf8ZxXhyVeE Hg9j1uliutZfVS7qXMYoCAQlObgOK6nyTJccBz8NUvXt7y+CDwIDAQABo0IwQDAd BgNVHQ4EFgQUU3m/WqorSs9UgOHYm8Cd8rIDZsswDgYDVR0PAQH/BAQDAgEGMA8G A1UdEwEB/wQFMAMBAf8wDQYJKoZIhvcNAQEMBQADggIBAFzUfA3P9wF9QZllDHPF Up/L+M+ZBn8b2kMVn54CVVeWFPFSPCeHlCjtHzoBN6J2/FNQwISbxmtOuowhT6KO VWKR82kV2LyI48SqC/3vqOlLVSoGIG1VeCkZ7l8wXEskEVX/JJpuXior7gtNn3/3 ATiUFJVDBwn7YKnuHKsSjKCaXqeYalltiz8I+8jRRa8YFWSQEg9zKC7F4iRO/Fjs 8PRF/iKz6y+O0tlFYQXBl2+odnKPi4w2r78NBc5xjeambx9spnFixdjQg3IM8WcR iQycE0xyNN+81XHfqnHd4blsjDwSXWXavVcStkNr/+XeTWYRUc+ZruwXtuhxkYze Sf7dNXGiFSeUHM9h4ya7b6NnJSFd5t0dCy5oGzuCr+yDZ4XUmFF0sbmZgIn/f3gZ XHlKYC6SQK5MNyosycdiyA5d9zZbyuAlJQG03RoHnHcAP9Dc1ew91Pq7P8yF1m9/ qS3fuQL39ZeatTXaw2ewh0qpKJ4jjv9cJ2vhsE/zB+4ALtRZh8tSQZXq9EfX7mRB VXyNWQKV3WKdwrnuWih0hKWbt5DHDAff9Yk2dDLWKMGwsAvgnEzDHNb842m1R0aB L6KCq9NjRHDEjf8tM7qtj3u1cIiuPhnPQCjY/MiQu12ZIvVS5ljFH4gxQ+6IHdfG jjxDah2nGN59PRbxYvnKkKj9 -----END CERTIFICATE----- -----BEGIN CERTIFICATE----- MIIFkDCCA3igAwIBAgIQBZsbV56OITLiOQe9p3d1XDANBgkqhkiG9w0BAQwFADBi MQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRGlnaUNlcnQgSW5jMRkwFwYDVQQLExB3 d3cuZGlnaWNlcnQuY29tMSEwHwYDVQQDExhEaWdpQ2VydCBUcnVzdGVkIFJvb3Qg RzQwHhcNMTMwODAxMTIwMDAwWhcNMzgwMTE1MTIwMDAwWjBiMQswCQYDVQQGEwJV UzEVMBMGA1UEChMMRGlnaUNlcnQgSW5jMRkwFwYDVQQLExB3d3cuZGlnaWNlcnQu Y29tMSEwHwYDVQQDExhEaWdpQ2VydCBUcnVzdGVkIFJvb3QgRzQwggIiMA0GCSqG SIb3DQEBAQUAA4ICDwAwggIKAoICAQC/5pBzaN675F1KPDAiMGkz7MKnJS7JIT3y ithZwuEppz1Yq3aaza57G4QNxDAf8xukOBbrVsaXbR2rsnnyyhHS5F/WBTxSD1If xp4VpX6+n6lXFllVcq9ok3DCsrp1mWpzMpTREEQQLt+C8weE5nQ7bXHiLQwb7iDV ySAdYyktzuxeTsiT+CFhmzTrBcZe7FsavOvJz82sNEBfsXpm7nfISKhmV1efVFiO DCu3T6cw2Vbuyntd463JT17lNecxy9qTXtyOj4DatpGYQJB5w3jHtrHEtWoYOAMQ jdjUN6QuBX2I9YI+EJFwq1WCQTLX2wRzKm6RAXwhTNS8rhsDdV14Ztk6MUSaM0C/ CNdaSaTC5qmgZ92kJ7yhTzm1EVgX9yRcRo9k98FpiHaYdj1ZXUJ2h4mXaXpI8OCi EhtmmnTK3kse5w5jrubU75KSOp493ADkRSWJtppEGSt+wJS00mFt6zPZxd9LBADM fRyVw4/3IbKyEbe7f/LVjHAsQWCqsWMYRJUadmJ+9oCw++hkpjPRiQfhvbfmQ6QY uKZ3AeEPlAwhHbJUKSWJbOUOUlFHdL4mrLZBdd56rF+NP8m800ERElvlEFDrMcXK chYiCd98THU/Y+whX8QgUWtvsauGi0/C1kVfnSD8oR7FwI+isX4KJpn15GkvmB0t 9dmpsh3lGwIDAQABo0IwQDAPBgNVHRMBAf8EBTADAQH/MA4GA1UdDwEB/wQEAwIB hjAdBgNVHQ4EFgQU7NfjgtJxXWRM3y5nP+e6mK4cD08wDQYJKoZIhvcNAQEMBQAD ggIBALth2X2pbL4XxJEbw6GiAI3jZGgPVs93rnD5/ZpKmbnJeFwMDF/k5hQpVgs2 SV1EY+CtnJYYZhsjDT156W1r1lT40jzBQ0CuHVD1UvyQO7uYmWlrx8GnqGikJ9yd +SeuMIW59mdNOj6PWTkiU0TryF0Dyu1Qen1iIQqAyHNm0aAFYF/opbSnr6j3bTWc fFqK1qI4mfN4i/RN0iAL3gTujJtHgXINwBQy7zBZLq7gcfJW5GqXb5JQbZaNaHqa sjYUegbyJLkJEVDXCLG4iXqEI2FCKeWjzaIgQdfRnGTZ6iahixTXTBmyUEFxPT9N cCOGDErcgdLMMpSEDQgJlxxPwO5rIHQw0uA5NBCFIRUBCOhVMt5xSdkoF1BN5r5N 0XWs0Mr7QbhDparTwwVETyw2m+L64kW4I1NsBm9nVX9GtUw/bihaeSbSpKhil9Ie 4u1Ki7wb/UdKDd9nZn6yW0HQO+T0O/QEY+nvwlQAUaCKKsnOeMzV6ocEGLPOr0mI r/OSmbaz5mEP0oUA51Aa5BuVnRmhuZyxm7EAHu/QD09CbMkKvO5D+jpxpchNJqU1 /YldvIViHTLSoCtU7ZpXwdv6EM8Zt4tKG48BtieVU+i2iW1bvGjUI+iLUaJW+fCm gKDWHrO8Dw9TdSmq6hN35N6MgSGtBxBHEa2HPQfRdbzP82Z+ -----END CERTIFICATE----- """ @dataclass(frozen=True) class TsaTrustPolicy: trust_report_chain: bool = False extra_root_files: tuple[Path, ...] = () TSA_TRUST_POLICY = TsaTrustPolicy() def split_pem_certificates(pem_text: str) -> list[str]: return re.findall(r"-----BEGIN CERTIFICATE-----.+?-----END CERTIFICATE-----", pem_text, flags=re.DOTALL) def trusted_tsa_roots_pem(policy: TsaTrustPolicy) -> str: parts = [EMBEDDED_TSA_ROOTS_PEM] system_bundle = ssl.get_default_verify_paths().cafile if system_bundle and Path(system_bundle).is_file(): parts.append(Path(system_bundle).read_text(encoding="utf-8", errors="ignore")) for root_file in policy.extra_root_files: parts.append(Path(root_file).read_text(encoding="utf-8")) return "\n".join(split_pem_certificates("\n".join(parts))) + "\n" @dataclass class ReportEntry: kind: str identifier: str url: str original_hash: str @property def fallback_name(self) -> str: prefix = "page" if self.kind == "Page" else "resource" return f"{prefix}-{self.identifier or 'unknown'}" @property def display_name(self) -> str: target = self.url or self.fallback_name return f"{self.kind} {self.identifier} {target}".strip() @property def is_layout_graphic_resource(self) -> bool: if self.kind != "Resource" or not self.url: return False try: parsed = urlparse(self.url) suffix = Path(parsed.path).suffix.lower() except Exception: # noqa: BLE001 return False return suffix in LAYOUT_GRAPHIC_RESOURCE_EXTENSIONS @dataclass class WorkspaceAssets: base_dir: Path extract_report: Path original_archive_zip: Path | None netlog: Path | None @dataclass class VerificationContext: messages: list[ValidationMessage] = field(default_factory=list) errors: int = 0 warnings: int = 0 def ok(self, text: str) -> None: self.messages.append(ValidationMessage("OK", text)) def info(self, text: str) -> None: self.messages.append(ValidationMessage("INFO", text)) def warning(self, text: str) -> None: self.messages.append(ValidationMessage("WARNING", text)) self.warnings += 1 def error(self, text: str) -> None: self.messages.append(ValidationMessage("ERROR", text)) self.errors += 1 def emit(self) -> None: for message in self.messages: print(f"[{message.level}] {message.text}") class ArchiveAccessor: def __init__(self, workspace: WorkspaceAssets) -> None: self.workspace = workspace self._temp_dir: tempfile.TemporaryDirectory[str] | None = None self._search_root = self._prepare_search_root() self._files_by_name = defaultdict(list) for path in self._search_root.rglob("*"): if path.is_file(): self._files_by_name[path.name].append(path) def _prepare_search_root(self) -> Path: if self.workspace.original_archive_zip is None: return self.workspace.base_dir temp_dir = tempfile.TemporaryDirectory(prefix="pemarks-original-archive-") self._temp_dir = temp_dir with zipfile.ZipFile(self.workspace.original_archive_zip) as archive: archive.extractall(temp_dir.name) return Path(temp_dir.name) def close(self) -> None: if self._temp_dir is not None: self._temp_dir.cleanup() self._temp_dir = None def read_entry(self, entry: ReportEntry) -> bytes: encoded_name = encode_entry_name(entry.url, entry.fallback_name) candidates = self._files_by_name.get(encoded_name, []) if not candidates: raise FileNotFoundError(f"No archived file named {encoded_name} for {entry.display_name}") if len(candidates) > 1: candidates = sorted(candidates, key=lambda current: (len(current.parts), str(current))) return candidates[0].read_bytes() class ExtractReportVerifier: def __init__(self, workspace: WorkspaceAssets) -> None: self.workspace = workspace self.context = VerificationContext() self.report_bytes = workspace.extract_report.read_bytes() self.report_text = self.report_bytes.decode("utf-8") self.root: ET.Element | None = None self.parse_error: ET.ParseError | None = None try: self.root = ET.fromstring(self.report_bytes) except ET.ParseError as exc: self.parse_error = exc self.archive_accessor = ArchiveAccessor(workspace) def close(self) -> None: self.archive_accessor.close() def verify(self) -> int: ignition_outcome: TsaValidationOutcome | None = None validation_outcome: TsaValidationOutcome | None = None try: report_valid = self.validate_report_structure() if report_valid: ignition_outcome = self.validate_ignition_tsa() validation_outcome = self.validate_validation_tsa() self.validate_original_archive_hashes() self.validate_netlog_window(ignition_outcome.gen_time if ignition_outcome else None, validation_outcome.gen_time if validation_outcome else None) else: self.context.info("skipping TSA, archive-hash, and netlog validation because the extract report is invalid") if report_valid and self.context.errors == 0: self.context.info("extract report verification completed successfully") else: self.context.info("extract report verification completed with errors") self.context.emit() return 0 if self.context.errors == 0 else 1 finally: self.close() def validate_report_structure(self) -> bool: failures: list[str] = [] if self.parse_error is not None: self.context.error(f"extract report validation failed: invalid XML: {self.parse_error}") return False if self.root is None: self.context.error("extract report validation failed: XML document could not be loaded") return False if self.root.tag != "report": failures.append(f"Root element must be , found <{self.root.tag}>") ignition = self.root.find("ignition") content = self.root.find("content") validation = self.root.find("validation") if ignition is None: failures.append("Missing section") if content is None: failures.append("Missing section") if validation is None: failures.append("Missing section") if ignition is not None: for tag in ("targetUrl", "user", "pemarksVersion", "extractionStartUtc", "urlHash", "urlTsa"): if ignition.find(tag) is None: failures.append(f"Missing /<${tag}> section".replace("$", "")) if content is not None: if "url" not in content.attrib: failures.append("Missing content@url attribute") if "ignitionTSA" not in content.attrib: failures.append("Missing content@ignitionTSA attribute") if content.find("ignitionTSA") is None: failures.append("Missing / section") if content.find("context") is None: failures.append("Missing / section") if validation is not None: for tag in ("contentHash", "contentTsa", "archivingCompletedUtc"): if validation.find(tag) is None: failures.append(f"Missing /<${tag}> section".replace("$", "")) if content is not None and validation is not None: content_hash = validation.findtext("contentHash", default="").strip() if not content_hash: failures.append("Validation contentHash is blank") content_ignition_tsa = content.attrib.get("ignitionTSA", "").strip() if not content_ignition_tsa: failures.append("content@ignitionTSA is blank") if failures: self.context.error("extract report validation failed: " + "; ".join(failures)) return False self.context.ok("extract report is valid") return True def validate_ignition_tsa(self) -> TsaValidationOutcome: if self.root is None: message = "ignition TSA cannot be checked because the report XML is invalid" self.context.error(message) return TsaValidationOutcome(False, "ignition TSA", message) ignition = self.root.find("ignition") target_url = text_of(ignition, "targetUrl") outcome = self._validate_tsa_element( tsa_parent=ignition, tsa_tag="urlTsa", label="ignition TSA", expected_payload=target_url.encode("utf-8"), expected_payload_id="target-url", expected_payload_type="URL", expected_stage="IGNITION", expected_source_url=target_url, ) content = self.root.find("content") content_ignition = (content.attrib.get("ignitionTSA") if content is not None else "") or "" embedded_ignition = condensed_tsa_value(ignition.find("urlTsa") if ignition is not None else None) if content is not None and content_ignition.strip() != embedded_ignition: message = ( "ignition TSA validation failed: content@ignitionTSA does not match " f"/ ({content_ignition.strip()} != {embedded_ignition})" ) self.context.error(message) return TsaValidationOutcome(False, "ignition TSA", message, outcome.gen_time) content_ignition_summary = content.find("ignitionTSA") if content is not None else None summary_failures = validate_embedded_ignition_summary( full_tsa=ignition.find("urlTsa") if ignition is not None else None, embedded_summary=content_ignition_summary, ) if summary_failures: message = "ignition TSA validation failed: " + "; ".join(summary_failures) self.context.error(message) return TsaValidationOutcome(False, "ignition TSA", message, outcome.gen_time) return outcome def validate_validation_tsa(self) -> TsaValidationOutcome: if self.root is None: message = "validation TSA cannot be checked because the report XML is invalid" self.context.error(message) return TsaValidationOutcome(False, "validation TSA", message) validation = self.root.find("validation") if validation is None: message = "validation TSA cannot be checked because is missing" self.context.error(message) return TsaValidationOutcome(False, "validation TSA", message) expected_content_hash = text_of(validation, "contentHash") reconstructed_hashes = self.compute_content_hashes() if normalize_hex(expected_content_hash) not in {normalize_hex(value) for value in reconstructed_hashes.values()}: computed_summary = ", ".join( f"{name}={value}" for name, value in sorted(reconstructed_hashes.items()) ) message = ("validation content hash mismatch: report=" f"{expected_content_hash} computed=[{computed_summary}]") self.context.error(message) return TsaValidationOutcome(False, "validation TSA", message) self.context.ok("validation content hash matches the serialized subtree") return self._validate_tsa_element( tsa_parent=validation, tsa_tag="contentTsa", label="validation TSA", expected_payload=bytes.fromhex(expected_content_hash), expected_payload_id="content-hash", expected_payload_type="CONTENT_HASH", expected_stage="VALIDATION", expected_source_url=self.root.find("content").attrib.get("url", "") if self.root.find("content") is not None else None, ) def compute_content_hashes(self) -> dict[str, str]: content = self.root.find("content") if content is None: raise ValueError("Missing element") candidates = { "canonical": hashlib.sha512(serialize_element_for_hash(content)).hexdigest(), "elementtree": hashlib.sha512(serialize_element_preserving_parser_whitespace(content)).hexdigest(), "elementtree_with_trailing_newline": hashlib.sha512( serialize_element_preserving_parser_whitespace(content, trailing_newline=True) ).hexdigest(), } raw_slice = extract_content_xml_slice(self.report_text) if raw_slice is not None: candidates["raw_report_slice"] = hashlib.sha512(raw_slice.encode("utf-8")).hexdigest() candidates["raw_report_slice_with_trailing_newline"] = hashlib.sha512( (raw_slice + "\n").encode("utf-8") ).hexdigest() return candidates def validate_original_archive_hashes(self) -> None: grouped_entries: dict[tuple[str, str, str], list[ReportEntry]] = defaultdict(list) for entry in self.iter_report_entries(): grouped_entries[(entry.kind, entry.identifier, entry.original_hash)].append(entry) for entries in grouped_entries.values(): payload: bytes | None = None resolved_entry: ReportEntry | None = None last_error: Exception | None = None for entry in entries: try: payload = self.archive_accessor.read_entry(entry) resolved_entry = entry break except Exception as exc: # noqa: BLE001 last_error = exc representative = entries[0] if payload is None: alias_urls = ", ".join(entry.url for entry in entries if entry.url) self.context.error( f"{representative.display_name}: original archive lookup failed: {last_error}; aliases=[{alias_urls}]" ) continue actual_hash = java_bigint_sha512_decimal(payload) if actual_hash != representative.original_hash: self.context.error( f"{representative.display_name}: original hash mismatch (report={representative.original_hash}, actual={actual_hash})" ) continue if len(entries) > 1 and resolved_entry is not None: self.context.ok( f"{representative.kind} {representative.identifier}: original archive hash matches extract report " f"for {len(entries)} aliased URLs via {resolved_entry.url}" ) else: self.context.ok(f"{representative.display_name}: original archive hash matches extract report") def validate_netlog_window(self, ignition_gen_time: datetime | None, validation_gen_time: datetime | None) -> None: if self.workspace.netlog is None: self.context.info("netlog report not provided; timestamp correlation skipped") return if ignition_gen_time is None or validation_gen_time is None: self.context.error("netlog timestamp validation cannot run because TSA timestamps are unavailable") return netlog_object = load_json_file(self.workspace.netlog) events = netlog_object.get("events", []) if not isinstance(events, list): self.context.error("netlog report is invalid: top-level events array is missing") return window_start = ignition_gen_time.astimezone(timezone.utc) window_end = validation_gen_time.astimezone(timezone.utc) if window_start > window_end: self.context.error( f"netlog timestamp validation failed: ignition TSA time {window_start.isoformat()} " f"is after validation TSA time {window_end.isoformat()}" ) return precision_tolerance = timedelta(seconds=1) effective_window_start = window_start - precision_tolerance effective_window_end = window_end + precision_tolerance events_by_url: dict[str, list[datetime]] = defaultdict(list) for event in events: if not isinstance(event, dict): continue url = first_non_blank(event.get("url"), nested_dict_value(event, "params", "url")) timestamp = extract_netlog_timestamp(event) if url and timestamp is not None: events_by_url[url].append(timestamp) for entry in self.iter_report_entries(): timestamps = events_by_url.get(entry.url, []) if not timestamps: self.context.info(f"{entry.display_name}: no browser netlog timestamp found; ignored") continue matching = [timestamp for timestamp in timestamps if effective_window_start <= timestamp <= effective_window_end] if matching: first_match = min(matching) self.context.ok( f"{entry.display_name}: netlog timestamp {first_match.isoformat()} is within TSA bounds" ) continue if ( entry.is_layout_graphic_resource and timestamps and max(timestamps) < effective_window_start and max(timestamps) >= (window_start - NETLOG_PRE_IGNITION_WARNING_TOLERANCE) ): seen = ", ".join(timestamp.isoformat() for timestamp in sorted(timestamps)) self.context.warning( f"{entry.display_name}: netlog timestamps are before ignition TSA but within the tolerated " f"pre-ignition window for layout/graphic resources " f"[{(window_start - NETLOG_PRE_IGNITION_WARNING_TOLERANCE).isoformat()} .. {window_start.isoformat()}], " f"seen={seen}" ) continue seen = ", ".join(timestamp.isoformat() for timestamp in sorted(timestamps)) self.context.error( f"{entry.display_name}: netlog timestamps are outside the TSA window " f"[{window_start.isoformat()} .. {window_end.isoformat()}] " f"(effective [{effective_window_start.isoformat()} .. {effective_window_end.isoformat()}]), seen={seen}" ) def iter_report_entries(self) -> Iterable[ReportEntry]: content = self.root.find("content") if content is None: return [] entries: list[ReportEntry] = [] for kind in ("Page", "Resource"): for element in content.findall(kind): entry = ReportEntry( kind=kind, identifier=text_of(element, "ID"), url=text_of(element, "URL"), original_hash=text_of(element, "OriginalHash"), ) if entry.identifier and entry.original_hash: entries.append(entry) return entries def _validate_tsa_element( self, tsa_parent: ET.Element | None, tsa_tag: str, label: str, expected_payload: bytes, expected_payload_id: str, expected_payload_type: str, expected_stage: str, expected_source_url: str | None, ) -> TsaValidationOutcome: tsa_element = tsa_parent.find(tsa_tag) if tsa_parent is not None else None if tsa_element is None: message = f"{label} is missing" self.context.error(message) return TsaValidationOutcome(False, label, message) if text_of(tsa_element, "status"): message = f"{label} validation failed: {text_of(tsa_element, 'status')} ({text_of(tsa_element, 'reason')})" self.context.error(message) return TsaValidationOutcome(False, label, message) tsa_data = xml_to_data(tsa_element) payload_data = as_dict(tsa_data.get("payload")) tsa_root = as_dict(tsa_data.get("tsa")) validation_data = as_dict(tsa_root.get("validation")) embedded_hash = as_dict(payload_data.get("hash")).get("hex", "") embedded_hash_algorithm = as_dict(payload_data.get("hash")).get("algorithm", "") expected_hash = hashlib.sha512(expected_payload).hexdigest() failures: list[str] = [] if payload_data.get("id") != expected_payload_id: failures.append(f"payload.id mismatch ({payload_data.get('id')} != {expected_payload_id})") if payload_data.get("type") != expected_payload_type: failures.append(f"payload.type mismatch ({payload_data.get('type')} != {expected_payload_type})") if payload_data.get("stage") != expected_stage: failures.append(f"payload.stage mismatch ({payload_data.get('stage')} != {expected_stage})") if expected_source_url and payload_data.get("sourceUrl") not in (None, "", expected_source_url): failures.append( f"payload.sourceUrl mismatch ({payload_data.get('sourceUrl')} != {expected_source_url})" ) if embedded_hash_algorithm not in ("", "SHA-512"): failures.append(f"payload.hash.algorithm mismatch ({embedded_hash_algorithm} != SHA-512)") if embedded_hash.lower() != expected_hash.lower(): failures.append(f"payload.hash.hex mismatch ({embedded_hash} != {expected_hash})") if validation_data.get("status") != "VALID": failures.append(f"tsa.validation.status is {validation_data.get('status')}, expected VALID") checks = as_dict(validation_data.get("checks")) failed_checks = [name for name, value in checks.items() if str(value).lower() != "true"] if failed_checks: failures.append("tsa.validation.checks contains non-true values for " + ", ".join(sorted(failed_checks))) token_base64 = as_dict(tsa_root.get("token")).get("base64", "") if not token_base64: failures.append("tsa.token.base64 is missing") if failures: message = f"{label} validation failed: " + "; ".join(failures) self.context.error(message) return TsaValidationOutcome(False, label, message) try: openssl_result = verify_tsa_token( token_base64=token_base64, payload=expected_payload, certificates=extract_certificate_chain(tsa_root), embedded_gen_time=tsa_root_path(tsa_root, "tstInfo", "genTimeUtc"), embedded_serial=tsa_root_path(tsa_root, "tstInfo", "serialNumberHex"), embedded_policy=tsa_root_path(tsa_root, "tstInfo", "policyOid"), ) except Exception as exc: # noqa: BLE001 message = f"{label} validation failed: {exc}" self.context.error(message) return TsaValidationOutcome(False, label, message) if not openssl_result.ok: message = f"{label} validation failed: {openssl_result.message}" self.context.error(message) return TsaValidationOutcome(False, label, message, openssl_result.gen_time) gen_time_text = openssl_result.gen_time.isoformat() if openssl_result.gen_time is not None else "unknown" message = f"{label} is valid (genTime={gen_time_text})" self.context.ok(message) return TsaValidationOutcome(True, label, message, openssl_result.gen_time) def encode_entry_name(value: str | None, fallback: str) -> str: source = (value or "").strip() or fallback if not source.strip(): source = "entry" encoded_parts: list[str] = [] for current in source.encode("utf-8"): if is_safe_entry_character(current): encoded_parts.append(chr(current)) else: encoded_parts.append(f"%{current:02X}") result = "".join(encoded_parts) if result in {"", ".", ".."}: result = fallback if len(result) > MAX_ENTRY_NAME_LENGTH: digest = hashlib.sha256(source.encode("utf-8")).hexdigest()[:16].upper() result = result[: min(140, len(result))] + "__" + digest return result def is_safe_entry_character(current: int) -> bool: return ( ord("a") <= current <= ord("z") or ord("A") <= current <= ord("Z") or ord("0") <= current <= ord("9") or current in (ord("."), ord("-"), ord("_")) ) def java_bigint_sha512_decimal(payload: bytes) -> str: digest = hashlib.sha512(payload).digest() return str(int.from_bytes(digest, byteorder="big", signed=True)) def serialize_element_for_hash(element: ET.Element) -> bytes: return _serialize_element(element, level=0).encode("utf-8") def serialize_element_preserving_parser_whitespace( element: ET.Element, *, trailing_newline: bool = False, ) -> bytes: serialized = ET.tostring(element, encoding="utf-8", short_empty_elements=True) if trailing_newline and not serialized.endswith(b"\n"): serialized += b"\n" return serialized def extract_content_xml_slice(report_text: str) -> str | None: match = re.search(r"", report_text, re.DOTALL) return match.group(0) if match else None def _serialize_element(element: ET.Element, level: int) -> str: indent = " " * level attributes = "".join( f' {key}="{escape_xml_attribute(value)}"' for key, value in element.attrib.items() ) children = list(element) text = element.text or "" if not children and text == "": return f"{indent}<{element.tag}{attributes}/>\n" if not children: if element.tag == "Diff": body = f"" else: body = escape_xml_text(text) return f"{indent}<{element.tag}{attributes}>{body}\n" output = [f"{indent}<{element.tag}{attributes}>"] if text and text.strip(): output.append(escape_xml_text(text)) output.append("\n") for child in children: output.append(_serialize_element(child, level + 1)) if child.tail and child.tail.strip(): output.append((" " * (level + 1)) + escape_xml_text(child.tail) + "\n") output.append(f"{indent}\n") return "".join(output) def split_cdata(text: str) -> str: return text.replace("]]>", "]]]]>") def escape_xml_text(text: str) -> str: return ( text.replace("&", "&") .replace("<", "<") .replace(">", ">") ) def escape_xml_attribute(text: str) -> str: return ( escape_xml_text(text) .replace('"', """) .replace("'", "'") ) def xml_to_data(element: ET.Element) -> Any: children = list(element) if not children: if element.attrib.get("null") == "true": return None return (element.text or "").strip() if all(child.tag == "item" for child in children): return [xml_to_data(child) for child in children] result: dict[str, Any] = {} for child in children: value = xml_to_data(child) if child.tag in result: current = result[child.tag] if not isinstance(current, list): result[child.tag] = [current] result[child.tag].append(value) else: result[child.tag] = value return result def verify_tsa_token( token_base64: str, payload: bytes, certificates: list[bytes], embedded_gen_time: str | None, embedded_serial: str | None, embedded_policy: str | None, trust_policy: TsaTrustPolicy | None = None, ) -> TsaValidationOutcome: policy = trust_policy if trust_policy is not None else TSA_TRUST_POLICY if shutil.which("openssl") is None: return TsaValidationOutcome(False, "TSA", "openssl is required but was not found on PATH") token = base64.b64decode(token_base64) if not certificates: return TsaValidationOutcome(False, "TSA", "TSA certificate chain is missing from the extract report") embedded_gen_dt = parse_datetime_utc(embedded_gen_time) if embedded_gen_time else None with tempfile.TemporaryDirectory(prefix="pemarks-tsa-") as temp_dir_name: temp_dir = Path(temp_dir_name) token_path = temp_dir / "token.tst" payload_path = temp_dir / "payload.bin" token_path.write_bytes(token) payload_path.write_bytes(payload) if policy.trust_report_chain: # Explicit opt-in (test / simulated TSAs): trust the chain stored in the report itself. ca_path, untrusted_path = write_certificate_chain(temp_dir, certificates) trust_args = ["-CAfile", str(ca_path), "-partial_chain"] else: ca_path = temp_dir / "trusted-tsa-roots.pem" ca_path.write_text(trusted_tsa_roots_pem(policy), encoding="utf-8") untrusted_path = temp_dir / "report-chain.pem" untrusted_path.write_text("".join(ssl.DER_cert_to_PEM_cert(c) for c in certificates), encoding="utf-8") trust_args = ["-CAfile", str(ca_path)] verify_command = [ "openssl", "ts", "-verify", "-token_in", "-in", str(token_path), "-data", str(payload_path), *trust_args, ] if embedded_gen_dt is not None: verify_command.extend(["-attime", str(int(embedded_gen_dt.timestamp()))]) if untrusted_path is not None: verify_command.extend(["-untrusted", str(untrusted_path)]) verify_result = subprocess.run( verify_command, text=True, capture_output=True, check=False, ) if verify_result.returncode != 0: details = " ".join(part.strip() for part in (verify_result.stdout, verify_result.stderr) if part.strip()) if not policy.trust_report_chain: return TsaValidationOutcome( False, "TSA", "TSA token verification failed: the TSA certificate does not chain to a trusted TSA root " "(embedded Pemarks TSA roots, system CA bundle, --tsa-trusted-roots). " f"openssl: {details}", ) return TsaValidationOutcome(False, "TSA", f"TSA token verification failed: {details}") inspect_result = subprocess.run( ["openssl", "ts", "-reply", "-token_in", "-in", str(token_path), "-text"], text=True, capture_output=True, check=False, ) if inspect_result.returncode != 0: details = " ".join(part.strip() for part in (inspect_result.stdout, inspect_result.stderr) if part.strip()) return TsaValidationOutcome(False, "TSA", f"TSA token inspection failed: {details}") parsed = parse_openssl_ts_text(inspect_result.stdout) expected_imprint = hashlib.sha512(payload).hexdigest() failures: list[str] = [] message_imprint = normalize_hex(parsed.get("message_imprint_hex")) if message_imprint != normalize_hex(expected_imprint): failures.append(f"message imprint mismatch ({message_imprint} != {expected_imprint})") parsed_gen_time = parsed.get("gen_time") if embedded_gen_dt is not None and parsed_gen_time is not None and parsed_gen_time != embedded_gen_dt: failures.append( f"genTime mismatch ({parsed_gen_time.isoformat()} != {embedded_gen_dt.isoformat()})" ) parsed_serial = normalize_hex(parsed.get("serial_hex")) if embedded_serial and parsed_serial != normalize_hex(embedded_serial): failures.append(f"serial mismatch ({parsed_serial} != {embedded_serial})") parsed_policy = (parsed.get("policy_oid") or "").strip() parsed_policy_normalized = normalize_policy_identifier(parsed_policy) embedded_policy_normalized = normalize_policy_identifier(embedded_policy) if embedded_policy and parsed_policy_normalized != embedded_policy_normalized: failures.append(f"policy mismatch ({parsed_policy} != {embedded_policy})") if failures: return TsaValidationOutcome(False, "TSA", "TSA token verification failed: " + "; ".join(failures), parsed_gen_time) gen_time_text = parsed_gen_time.isoformat() if parsed_gen_time is not None else "unknown" return TsaValidationOutcome(True, "TSA", f"TSA token is valid (genTime={gen_time_text})", parsed_gen_time) def parse_openssl_ts_text(output: str) -> dict[str, Any]: result: dict[str, Any] = {} lines = output.splitlines() index = 0 while index < len(lines): line = lines[index].rstrip() match = OPENSSL_TEXT_FIELD_PATTERN.match(line.strip()) if match: field, value = match.groups() if field == "Policy OID": result["policy_oid"] = value.strip() elif field == "Hash Algorithm": result["hash_algorithm"] = value.strip() elif field == "Serial number": result["serial_hex"] = value.strip() elif field == "Time stamp": result["gen_time"] = parse_openssl_ts_datetime(value.strip()) index += 1 continue if line.strip() == "Message data:": hex_bytes: list[str] = [] index += 1 while index < len(lines): candidate_line = lines[index].rstrip() candidate = candidate_line.strip() if not candidate: index += 1 continue if OPENSSL_TEXT_FIELD_PATTERN.match(candidate): break match = OPENSSL_HEXDUMP_LINE_PATTERN.match(candidate_line) if not match: break payload_section = match.group(1) payload_section = re.split(r"\s{2,}", payload_section, maxsplit=1)[0] hex_bytes.extend(re.findall(r"\b[0-9A-Fa-f]{2}\b", payload_section)) index += 1 result["message_imprint_hex"] = "".join(hex_bytes) continue index += 1 return result def write_certificate_chain(directory: Path, certificates: list[bytes]) -> tuple[Path, Path | None]: pem_entries = [ssl.DER_cert_to_PEM_cert(certificate) for certificate in certificates] ca_path = directory / "ca.pem" untrusted_path = directory / "untrusted.pem" if len(pem_entries) == 1: ca_path.write_text(pem_entries[0], encoding="utf-8") return ca_path, None ca_path.write_text(pem_entries[-1], encoding="utf-8") untrusted_path.write_text("".join(pem_entries[:-1]), encoding="utf-8") return ca_path, untrusted_path def extract_certificate_chain(tsa_root: dict[str, Any]) -> list[bytes]: certificates_root = as_dict(tsa_root.get("certificates")) entries = certificates_root.get("entries") certificates: list[bytes] = [] if isinstance(entries, list): for entry in entries: entry_dict = as_dict(entry) der_base64 = entry_dict.get("derBase64") if der_base64: certificates.append(base64.b64decode(der_base64)) if certificates: return certificates chain_entries = certificates_root.get("chainDerBase64") if isinstance(chain_entries, list): for value in chain_entries: if value: certificates.append(base64.b64decode(str(value))) return certificates def tsa_root_path(root: dict[str, Any], *path: str) -> str | None: current: Any = root for key in path: if not isinstance(current, dict): return None current = current.get(key) return current if isinstance(current, str) else None def extract_netlog_timestamp(event: dict[str, Any]) -> datetime | None: for key in ("capturedAtEpochMs", "time"): value = event.get(key) if value is None: continue if isinstance(value, (int, float)): return datetime.fromtimestamp(float(value) / 1000.0, tz=timezone.utc) if isinstance(value, str): stripped = value.strip() if not stripped: continue try: return datetime.fromtimestamp(float(stripped) / 1000.0, tz=timezone.utc) except ValueError: try: return parse_datetime_utc(stripped) except ValueError: continue return None def load_json_file(path: Path) -> dict[str, Any]: import json return json.loads(path.read_text(encoding="utf-8")) def parse_datetime_utc(value: str) -> datetime: if not value: raise ValueError("datetime value is empty") candidate = value.strip() if candidate.endswith("Z"): candidate = candidate[:-1] + "+00:00" parsed = datetime.fromisoformat(candidate) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) def parse_openssl_ts_datetime(value: str) -> datetime: candidate = value.strip() for pattern in ("%b %d %H:%M:%S %Y %Z", "%b %e %H:%M:%S %Y %Z"): try: parsed = datetime.strptime(candidate, pattern) return parsed.replace(tzinfo=timezone.utc) except ValueError: continue raise ValueError(f"Unsupported OpenSSL TSA timestamp format: {value}") def candidate_openssl_config_paths() -> list[Path]: candidates: list[Path] = [] environment_value = os.environ.get("OPENSSL_CONF", "").strip() if environment_value: candidates.append(Path(environment_value)) candidates.extend( [ Path("/etc/ssl/openssl.cnf"), Path("/usr/lib/ssl/openssl.cnf"), Path("/etc/pki/tls/openssl.cnf"), ] ) resolved: list[Path] = [] seen: set[Path] = set() for candidate in candidates: if not candidate.exists() or candidate in seen: continue seen.add(candidate) resolved.append(candidate) return resolved def load_openssl_oid_aliases_from_config(config_path: Path) -> dict[str, str]: root_assignments: dict[str, str] = {} sections: dict[str, dict[str, str]] = defaultdict(dict) current_section: str | None = None for raw_line in config_path.read_text(encoding="utf-8", errors="ignore").splitlines(): line = raw_line.split("#", 1)[0].split(";", 1)[0].strip() if not line: continue section_match = OPENSSL_SECTION_PATTERN.match(line) if section_match: current_section = section_match.group(1).strip() continue assignment_match = OPENSSL_ASSIGNMENT_PATTERN.match(line) if not assignment_match: continue key = assignment_match.group(1).strip().lower() value = assignment_match.group(2).strip() if current_section is None: root_assignments[key] = value else: sections[current_section][key] = value oid_section_name = root_assignments.get("oid_section", "").strip() if not oid_section_name: return {} aliases: dict[str, str] = {} for alias, value in sections.get(oid_section_name, {}).items(): if OID_DOTTED_PATTERN.fullmatch(value): aliases[alias.lower()] = value return aliases @lru_cache(maxsize=1) def get_openssl_oid_aliases() -> dict[str, str]: aliases: dict[str, str] = {} for config_path in candidate_openssl_config_paths(): aliases.update(load_openssl_oid_aliases_from_config(config_path)) return aliases def normalize_policy_identifier(value: str | None, aliases: dict[str, str] | None = None) -> str: if value is None: return "" normalized = value.strip() if not normalized: return "" if OID_DOTTED_PATTERN.fullmatch(normalized): return normalized alias_map = aliases if aliases is not None else get_openssl_oid_aliases() return alias_map.get(normalized.lower(), normalized) def normalize_hex(value: str | None) -> str: if value is None: return "" normalized = value.strip().lower() if normalized.startswith("0x"): normalized = normalized[2:] normalized = normalized.replace(":", "").replace(" ", "") return normalized.lstrip("0") or "0" def condensed_tsa_value(element: ET.Element | None) -> str: if element is None: return "" data = xml_to_data(element) if not isinstance(data, dict): return text_of(element, "status") reference = str(data.get("reference", "")).strip() if reference: return reference status = str(data.get("status", "")).strip() authority_id = first_non_blank(data.get("authorityId"), data.get("authorityName")) gen_time = first_non_blank(data.get("genTimeUtc")) if authority_id and gen_time: return f"{authority_id}@{gen_time}" tsa_root = as_dict(data.get("tsa")) authority = as_dict(tsa_root.get("authority")) authority_id = first_non_blank(authority.get("id"), authority.get("name")) gen_time = tsa_root_path(tsa_root, "tstInfo", "genTimeUtc") if authority_id and gen_time: return f"{authority_id}@{gen_time}" token = str(as_dict(tsa_root.get("token")).get("base64", "")).strip() if token: return token return status def business_ignition_summary_from_full_tsa(element: ET.Element | None) -> dict[str, str]: if element is None: return {} status = text_of(element, "status") reason = text_of(element, "reason") if status: summary: dict[str, str] = {"status": status} if reason: summary["reason"] = reason return summary data = xml_to_data(element) if not isinstance(data, dict): return {} tsa_root = as_dict(data.get("tsa")) authority = as_dict(tsa_root.get("authority")) summary = { "status": first_non_blank(tsa_root_path(tsa_root, "validation", "status"), "AVAILABLE") or "AVAILABLE", "reference": condensed_tsa_value(element), "authorityId": str(authority.get("id", "")).strip(), "authorityName": str(authority.get("name", "")).strip(), "authorityDomain": str(authority.get("domain", "")).strip(), "genTimeUtc": str(tsa_root_path(tsa_root, "tstInfo", "genTimeUtc")).strip(), } return {key: value for key, value in summary.items() if value} def validate_embedded_ignition_summary( *, full_tsa: ET.Element | None, embedded_summary: ET.Element | None, ) -> list[str]: expected = business_ignition_summary_from_full_tsa(full_tsa) if not expected: return [] if embedded_summary is None: return ["/ summary is missing"] actual = xml_to_data(embedded_summary) if not isinstance(actual, dict): return ["/ summary is invalid"] failures: list[str] = [] for key, expected_value in expected.items(): actual_value = str(actual.get(key, "")).strip() if actual_value != expected_value: failures.append( f"/ {key} mismatch ({actual_value} != {expected_value})" ) return failures def as_dict(value: Any) -> dict[str, Any]: return value if isinstance(value, dict) else {} def text_of(parent: ET.Element | None, tag: str) -> str: if parent is None: return "" child = parent.find(tag) if child is None or child.text is None: return "" return child.text.strip() def first_non_blank(*values: Any) -> str | None: for value in values: if isinstance(value, str) and value.strip(): return value.strip() return None def nested_dict_value(data: dict[str, Any], *keys: str) -> Any: current: Any = data for key in keys: if not isinstance(current, dict): return None current = current.get(key) return current def detect_workspace_assets( base_dir: Path, extract_report_override: Path | None = None, archive_override: Path | None = None, netlog_override: Path | None = None, ) -> WorkspaceAssets: if not base_dir.exists(): raise FileNotFoundError(f"Workspace folder does not exist: {base_dir}") extract_report = extract_report_override or choose_single_candidate( [ path for path in base_dir.iterdir() if path.is_file() and path.suffix.lower() == ".xml" and looks_like_extract_report(path) ], "extract report XML", ) original_archive_zip = archive_override or choose_optional_candidate( [ path for path in base_dir.iterdir() if path.is_file() and path.suffix.lower() == ".zip" ], preferred_substrings=("original-resources", "pemarks-original-resources"), ) netlog = netlog_override or choose_optional_candidate( [ path for path in base_dir.iterdir() if path.is_file() and path.suffix.lower() == ".json" and looks_like_netlog(path) ], preferred_substrings=("netlog",), ) return WorkspaceAssets( base_dir=base_dir, extract_report=extract_report, original_archive_zip=original_archive_zip, netlog=netlog, ) def looks_like_extract_report(path: Path) -> bool: try: text = path.read_text(encoding="utf-8") except UnicodeDecodeError: return False return " bool: try: text = path.read_text(encoding="utf-8") except UnicodeDecodeError: return False return '"events"' in text def choose_single_candidate(candidates: list[Path], description: str) -> Path: if not candidates: raise FileNotFoundError(f"No {description} found") if len(candidates) == 1: return candidates[0] preferred = choose_optional_candidate(candidates, preferred_substrings=("extract", "report")) if preferred is not None: return preferred joined = ", ".join(str(candidate.name) for candidate in sorted(candidates)) raise FileExistsError(f"Multiple {description} files found: {joined}") def choose_optional_candidate(candidates: list[Path], preferred_substrings: tuple[str, ...]) -> Path | None: if not candidates: return None lowered = [] for candidate in candidates: score = sum(substring in candidate.name.lower() for substring in preferred_substrings) lowered.append((score, candidate.name.lower(), candidate)) lowered.sort(key=lambda current: (-current[0], current[1])) return lowered[0][2] def parse_arguments(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Validate a Pemarks extract report against the original-resources archive and optional netlog." ) parser.add_argument( "folder", nargs="?", default=".", help="Workspace folder containing the extract report, the original-resources archive zip or extracted tree, and optionally the netlog report.", ) parser.add_argument("--extract-report", dest="extract_report", help="Explicit path to the extract report XML.") parser.add_argument("--archive", dest="archive", help="Explicit path to the original-resources ZIP archive.") parser.add_argument("--netlog", dest="netlog", help="Explicit path to the netlog JSON file.") parser.add_argument( "--tsa-trusted-roots", dest="tsa_trusted_roots", action="append", default=[], help="Additional PEM file of trusted TSA root certificates (repeatable).", ) parser.add_argument( "--tsa-trust-report-chain", dest="tsa_trust_report_chain", action="store_true", help="Trust the TSA certificate chain stored in the extract report itself (only for test / simulated TSAs).", ) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_arguments(sys.argv[1:] if argv is None else argv) global TSA_TRUST_POLICY TSA_TRUST_POLICY = TsaTrustPolicy( trust_report_chain=bool(args.tsa_trust_report_chain), extra_root_files=tuple(Path(path).resolve() for path in args.tsa_trusted_roots), ) try: workspace = detect_workspace_assets( base_dir=Path(args.folder).resolve(), extract_report_override=Path(args.extract_report).resolve() if args.extract_report else None, archive_override=Path(args.archive).resolve() if args.archive else None, netlog_override=Path(args.netlog).resolve() if args.netlog else None, ) verifier = ExtractReportVerifier(workspace) return verifier.verify() except Exception as exc: # noqa: BLE001 print(f"[ERROR] verifier failed to start: {exc}") return 1 if __name__ == "__main__": sys.exit(main())