audit-labs/evidence-seal

  1"""Optional RFC 3161 trusted timestamping of manifests.
  2
  3A signature proves *who* sealed the evidence; a timestamp proves the seal
  4existed *by* a certain time — attested by an independent Time-Stamp Authority
  5(TSA), not by the sealer's own clock. The TSA timestamps the manifest's ``id``
  6(itself the hash of every file and all metadata), so one token vouches for the
  7whole package at a point in time.
  8
  9This module builds RFC 3161 requests, submits them to a TSA, and binds the
 10returned token into the manifest. It needs the ``asn1crypto`` package (install
 11``evidence-seal[timestamp]``); the core seal/verify path never imports it.
 12
 13The bound token is stored under a ``timestamp`` key, which — like ``signature``
 14— is excluded from the manifest id, so timestamping never invalidates the id or
 15an existing signature.
 16"""
 17
 18from __future__ import annotations
 19
 20import base64
 21import secrets
 22import urllib.request
 23from datetime import timezone
 24from pathlib import Path
 25
 26_TSA_CONTENT_TYPE = "application/timestamp-query"
 27_TSA_ACCEPT = "application/timestamp-reply"
 28
 29
 30def _require_asn1():
 31    try:
 32        from asn1crypto import algos, cms, core, tsp
 33    except ImportError as exc:  # pragma: no cover - exercised via a clear message
 34        raise RuntimeError(
 35            "timestamping requires the 'asn1crypto' package — "
 36            "install evidence-seal[timestamp]"
 37        ) from exc
 38    return algos, cms, core, tsp
 39
 40
 41def _require_crypto():
 42    try:
 43        import cryptography  # noqa: F401
 44    except ImportError as exc:  # pragma: no cover - exercised via a clear message
 45        raise RuntimeError(
 46            "verifying a TSA signature requires the 'cryptography' package — "
 47            "install evidence-seal[sign,timestamp]"
 48        ) from exc
 49
 50
 51def build_request(manifest_id_hex: str, cert_req: bool = True) -> bytes:
 52    """Return a DER-encoded RFC 3161 TimeStampReq over a manifest id.
 53
 54    The message imprint is the manifest id (a SHA-256 digest) carried directly
 55    as the hashed message, so the TSA timestamps exactly what the id commits to.
 56    """
 57    algos, _cms, core, tsp = _require_asn1()
 58    request = tsp.TimeStampReq(
 59        {
 60            "version": 1,
 61            "message_imprint": tsp.MessageImprint(
 62                {
 63                    "hash_algorithm": algos.DigestAlgorithm({"algorithm": "sha256"}),
 64                    "hashed_message": bytes.fromhex(manifest_id_hex),
 65                }
 66            ),
 67            "nonce": core.Integer(secrets.randbits(64)),
 68            "cert_req": cert_req,
 69        }
 70    )
 71    return request.dump()
 72
 73
 74def submit(request_der: bytes, tsa_url: str, timeout: float = 30.0) -> bytes:
 75    """POST a TimeStampReq to a TSA and return the raw TimeStampResp bytes.
 76
 77    Network call — the caller is responsible for choosing a trusted TSA URL.
 78    """
 79    req = urllib.request.Request(
 80        tsa_url,
 81        data=request_der,
 82        headers={"Content-Type": _TSA_CONTENT_TYPE, "Accept": _TSA_ACCEPT},
 83        method="POST",
 84    )
 85    with urllib.request.urlopen(req, timeout=timeout) as response:
 86        return response.read()
 87
 88
 89def _extract(token_or_response_der: bytes):
 90    """Return ``(token_contentinfo, tst_info)`` from a token or a TimeStampResp.
 91
 92    Accepts either a bare RFC 3161 token (a CMS ContentInfo) or a full
 93    TimeStampResp (what a TSA returns and a ``.tsr`` file holds).
 94    """
 95    _algos, cms, _core, tsp = _require_asn1()
 96
 97    def _tst_of(content_info):
 98        return content_info["content"]["encap_content_info"]["content"].parsed
 99
100    # Try a full response first; fall back to a bare token.
101    try:
102        response = tsp.TimeStampResp.load(token_or_response_der)
103        status = response["status"]["status"].native
104        token = response["time_stamp_token"]
105        tst = _tst_of(token)  # forces a parse; raises if this was not a response
106    except Exception:
107        token = cms.ContentInfo.load(token_or_response_der)
108        return token, _tst_of(token), None
109    return token, tst, status
110
111
112def parse_token(token_or_response_der: bytes) -> dict:
113    """Extract the human-facing fields from a timestamp token."""
114    _token, tst, _status = _extract(token_or_response_der)
115    gen_time = tst["gen_time"].native.astimezone(timezone.utc)
116    tsa = None
117    if tst["tsa"].native is not None:
118        tsa = str(tst["tsa"].native)
119    return {
120        "imprint": tst["message_imprint"]["hashed_message"].native.hex(),
121        "imprint_algorithm": tst["message_imprint"]["hash_algorithm"]["algorithm"].native,
122        "gen_time": gen_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
123        "serial_number": str(tst["serial_number"].native),
124        "tsa": tsa,
125    }
126
127
128def apply_timestamp(manifest: dict, token_or_response_der: bytes) -> dict:
129    """Bind a TSA token into *manifest* after checking it timestamps its id.
130
131    Raises ``ValueError`` if the token's message imprint does not equal the
132    manifest id — a token for anything else must never be attached.
133    """
134    token, _tst, _status = _extract(token_or_response_der)
135    fields = parse_token(token_or_response_der)
136
137    if fields["imprint_algorithm"] != "sha256" or fields["imprint"] != manifest.get("id"):
138        raise ValueError("token does not timestamp this manifest's id")
139
140    stamped = dict(manifest)
141    stamped["timestamp"] = {
142        "format": "rfc3161",
143        "imprint_algorithm": fields["imprint_algorithm"],
144        "imprint": fields["imprint"],
145        "gen_time": fields["gen_time"],
146        "serial_number": fields["serial_number"],
147        "tsa": fields["tsa"],
148        "token": base64.b64encode(token.dump()).decode(),
149    }
150    return stamped
151
152
153def verify_timestamp(manifest: dict, tsa_cert: bytes | None = None) -> tuple[bool, str]:
154    """Verify a manifest's embedded timestamp.
155
156    Returns ``(ok, message)``. Always checks that the stored token timestamps the
157    current manifest id. When *tsa_cert* (PEM or DER bytes) is given, the token's
158    RFC 3161 CMS signature is also verified against that certificate — proving
159    the timestamp really was issued by that authority.
160    """
161    block = manifest.get("timestamp")
162    if not block:
163        return False, "manifest is not timestamped"
164    if block.get("format") != "rfc3161":
165        return False, f"unsupported timestamp format: {block.get('format')}"
166
167    try:
168        token_der = base64.b64decode(block["token"])
169        fields = parse_token(token_der)
170    except Exception as exc:
171        return False, f"timestamp token is unreadable: {exc}"
172
173    if fields["imprint_algorithm"] != "sha256" or fields["imprint"] != manifest.get("id"):
174        return False, "timestamp does not match the manifest id"
175
176    tsa = f" by {fields['tsa']}" if fields["tsa"] else ""
177    bound = f"timestamped at {fields['gen_time']}{tsa}"
178
179    if tsa_cert is None:
180        return True, bound
181
182    sig_ok, sig_message = verify_token_signature(token_der, tsa_cert)
183    return sig_ok, f"{bound}; {sig_message}"
184
185
186# --------------------------------------------------------------------------- #
187# Full RFC 3161 CMS signature verification
188# --------------------------------------------------------------------------- #
189
190# asn1crypto hash names -> cryptography hash classes for the algorithms a TSA
191# realistically signs with.
192_HASHES = {
193    "sha1": "SHA1",
194    "sha224": "SHA224",
195    "sha256": "SHA256",
196    "sha384": "SHA384",
197    "sha512": "SHA512",
198}
199
200
201def _load_certificate(cert_bytes: bytes):
202    from cryptography import x509
203
204    try:
205        return x509.load_pem_x509_certificate(cert_bytes)
206    except ValueError:
207        return x509.load_der_x509_certificate(cert_bytes)
208
209
210def _hash_instance(name: str):
211    from cryptography.hazmat.primitives import hashes
212
213    if name not in _HASHES:
214        raise ValueError(f"unsupported digest algorithm: {name}")
215    return getattr(hashes, _HASHES[name])()
216
217
218def _verify_raw(public_key, signature: bytes, data: bytes, sig_algo: str, hash_name: str) -> None:
219    """Verify *signature* over *data*, raising on any failure."""
220    from cryptography.hazmat.primitives.asymmetric import ec, padding
221
222    if sig_algo == "rsassa_pkcs1v15":
223        public_key.verify(signature, data, padding.PKCS1v15(), _hash_instance(hash_name))
224    elif sig_algo == "rsassa_pss":
225        digest = _hash_instance(hash_name)
226        public_key.verify(
227            signature,
228            data,
229            padding.PSS(mgf=padding.MGF1(digest), salt_length=padding.PSS.DIGEST_LENGTH),
230            digest,
231        )
232    elif sig_algo == "ecdsa":
233        public_key.verify(signature, data, ec.ECDSA(_hash_instance(hash_name)))
234    elif sig_algo in ("ed25519", "ed448"):
235        public_key.verify(signature, data)
236    else:
237        raise ValueError(f"unsupported signature algorithm: {sig_algo}")
238
239
240def _signer_matches_cert(signer_info, cert) -> bool:
241    """True if the SignerInfo identifies the supplied certificate."""
242    sid = signer_info["sid"]
243    if sid.name == "issuer_and_serial_number":
244        return sid.chosen["serial_number"].native == cert.serial_number
245    # subject_key_identifier: compare against the cert's SKI extension.
246    from cryptography import x509
247
248    try:
249        ski = cert.extensions.get_extension_for_class(x509.SubjectKeyIdentifier).value
250    except x509.ExtensionNotFound:
251        return False
252    return sid.chosen.native == ski.digest
253
254
255def _has_timestamping_eku(cert) -> bool:
256    from cryptography import x509
257    from cryptography.x509.oid import ExtendedKeyUsageOID
258
259    try:
260        eku = cert.extensions.get_extension_for_class(x509.ExtendedKeyUsage).value
261    except x509.ExtensionNotFound:
262        return False
263    return ExtendedKeyUsageOID.TIME_STAMPING in eku
264
265
266def _signed_attr(signed_attrs, attr_type):
267    for attr in signed_attrs:
268        if attr["type"].native == attr_type:
269            return attr["values"][0].native
270    return None
271
272
273def verify_token_signature(token_or_response_der: bytes, cert_bytes: bytes) -> tuple[bool, str]:
274    """Verify a token's CMS signature against a TSA certificate.
275
276    Checks, in order: the certificate carries the timeStamping extended key
277    usage; it identifies the token's signer; ``gen_time`` falls within its
278    validity window; the signed message digest matches the timestamped content;
279    and the signature verifies. Returns ``(ok, message)``.
280
281    This authenticates the token against the certificate you supply. Establishing
282    that the certificate itself is trusted (chain to a known root) is left to the
283    caller — supply a TSA certificate you already trust.
284    """
285    import hashlib
286
287    _require_asn1()
288    _require_crypto()
289
290    try:
291        token, tst, _status = _extract(token_or_response_der)
292        signed_data = token["content"]
293        signer_info = signed_data["signer_infos"][0]
294    except Exception as exc:
295        return False, f"TSA signature: token is unreadable ({exc})"
296
297    try:
298        cert = _load_certificate(cert_bytes)
299    except ValueError as exc:
300        return False, f"TSA signature: cannot load certificate ({exc})"
301
302    if not _has_timestamping_eku(cert):
303        return False, "TSA signature: certificate lacks the timeStamping extended key usage"
304    if not _signer_matches_cert(signer_info, cert):
305        return False, "TSA signature: certificate does not match the token's signer"
306
307    gen_time = tst["gen_time"].native.astimezone(timezone.utc)
308    if not (cert.not_valid_before_utc <= gen_time <= cert.not_valid_after_utc):
309        return False, "TSA signature: gen_time is outside the certificate validity window"
310
311    econtent = signed_data["encap_content_info"]["content"].parsed.dump()
312    digest_name = signer_info["digest_algorithm"]["algorithm"].native
313    signed_attrs = signer_info["signed_attrs"]
314
315    if signed_attrs.native is not None:
316        recorded = _signed_attr(signed_attrs, "message_digest")
317        if recorded is None or recorded != hashlib.new(digest_name, econtent).digest():
318            return False, "TSA signature: signed message digest does not match the token content"
319        signed_bytes = signed_attrs.untag().dump()
320    else:
321        signed_bytes = econtent
322
323    sig_algo = signer_info["signature_algorithm"].signature_algo
324    # For rsassa_pkcs1v15 the OID carries no hash, and asn1crypto raises rather
325    # than returning None — fall back to the SignerInfo digest algorithm.
326    try:
327        hash_name = signer_info["signature_algorithm"].hash_algo or digest_name
328    except ValueError:
329        hash_name = digest_name
330    try:
331        _verify_raw(
332            cert.public_key(), signer_info["signature"].native, signed_bytes, sig_algo, hash_name
333        )
334    except ValueError as exc:
335        return False, f"TSA signature: {exc}"
336    except Exception as exc:
337        return False, f"TSA signature is INVALID ({type(exc).__name__})"
338
339    return True, f"TSA signature valid ({cert.subject.rfc4514_string()})"
340
341
342def load_der(path: str | Path) -> bytes:
343    """Read a DER file (a ``.tsq`` request or ``.tsr`` response)."""
344    return Path(path).read_bytes()