Spaces:
Sleeping
Sleeping
import marimo
__generated_with = "0.25.2"
app = marimo.App(
width="medium",
)
@app.cell(hide_code=True)
def _():
import marimo as mo
return (mo,)
@app.cell(hide_code=True)
def _(mo):
mo.md("""
# 🔏 Proof Bundle Verifier
Everything runs locally in this notebook. Nothing is uploaded anywhere; network calls happen only when you switch on **network checks**.
**Two ways to use it**
- **Verify a whole bundle** — upload it as a `.zip` and every artifact is checked and cross-referenced: manifest hashes, detached signatures, CMS/PKCS#7, RFC 3161 timestamp, OpenTimestamps, Rekor transparency log, certificate chains, and CRL / OCSP revocation.
- **Manual check** (further down) — drop one certificate, one file and one signature to verify a single detached signature.
""")
return
@app.cell(hide_code=True)
def _(
BUNDLES_DIR,
DATA_DIR,
LOCAL_PATHS_ALLOWED,
REPORTS_DIR,
mo,
pathlib,
uploaded_bundles,
):
_uploaded = uploaded_bundles()
bundle_zip = mo.ui.file(filetypes=[".zip"], multiple=False, kind="area", label="Upload a proof bundle as .zip")
bundle_choice = mo.ui.dropdown(options={k: str(v) for k, v in _uploaded.items()}, label="…or pick a bundle uploaded earlier") if _uploaded else None
online_switch = mo.ui.switch(label="Network checks — Rekor live lookup & key, Bitcoin block headers, CA issuer download, CRL / OCSP revocation")
if LOCAL_PATHS_ALLOWED:
bundle_picker = mo.ui.file_browser(
initial_path=DATA_DIR if DATA_DIR.is_dir() else pathlib.Path.home(),
selection_mode="directory",
multiple=False,
restrict_navigation=False,
label="Pick a proof-bundle folder on this machine",
)
bundle_path_text = mo.ui.text(placeholder="…or paste a folder path, e.g. ~/Downloads/SW-2026-428978-proof-bundle", full_width=True)
_local_controls = [mo.md("*Local folder access is on (editor mode). It is hidden in the deployed app unless `VERIFIER_ALLOW_LOCAL_PATHS=1`.*"), bundle_picker, bundle_path_text]
else:
bundle_picker = None
bundle_path_text = None
_local_controls = []
_storage_note = (
mo.md(f"Uploaded bundles are kept in `{BUNDLES_DIR}` and PDF reports in `{REPORTS_DIR}`.")
if DATA_DIR.is_dir()
else mo.md(f"*No persistent storage at `{DATA_DIR}`; uploads are extracted to a temporary folder and reports are download-only.*")
)
mo.vstack(
[mo.md("## 1 · Verify a whole bundle"), bundle_zip]
+ ([bundle_choice] if bundle_choice is not None else [])
+ _local_controls
+ [online_switch, _storage_note]
)
return (
bundle_choice,
bundle_path_text,
bundle_picker,
bundle_zip,
online_switch,
)
@app.cell(hide_code=True)
def _(
bundle_choice,
bundle_path_text,
bundle_picker,
bundle_zip,
extract_bundle_zip,
is_allowed_bundle_dir,
mo,
online_switch,
pathlib,
run_bundle_checks,
upload_bytes,
):
bundle_dir = None
bundle_source = None
bundle_error = None
_zip_name, _zip_bytes = upload_bytes(bundle_zip.value)
if _zip_bytes is not None:
try:
bundle_dir = extract_bundle_zip(_zip_name, _zip_bytes)
bundle_source = f"uploaded `{_zip_name}` → `{bundle_dir}`"
except Exception as _e:
bundle_error = f"Could not extract {_zip_name}: {_e}"
elif bundle_choice is not None and bundle_choice.value:
bundle_dir = pathlib.Path(bundle_choice.value)
bundle_source = "previously uploaded bundle"
elif bundle_path_text is not None and bundle_path_text.value.strip():
bundle_dir = pathlib.Path(bundle_path_text.value.strip()).expanduser()
bundle_source = "path"
elif bundle_picker is not None and bundle_picker.value:
bundle_dir = pathlib.Path(bundle_picker.value[0].path)
bundle_source = "folder picker"
bundle_results = None
if bundle_dir is not None and not is_allowed_bundle_dir(bundle_dir):
bundle_error = "Only uploaded bundles can be verified in this deployment."
bundle_dir = None
if bundle_dir is not None:
with mo.status.spinner(title=f"Verifying {bundle_dir.name}…" + (" (with network checks)" if online_switch.value else "")):
try:
bundle_results = run_bundle_checks(bundle_dir, online=online_switch.value)
except Exception as _e:
bundle_error = f"{type(_e).__name__}: {_e}"
return bundle_dir, bundle_error, bundle_results, bundle_source
@app.cell(hide_code=True)
def _(
LOCAL_PATHS_ALLOWED,
STATUS_ICON,
bundle_dir,
bundle_error,
bundle_results,
mo,
):
def _render_bundle():
if bundle_error:
return mo.callout(mo.md(f"**Could not verify `{bundle_dir}`.**\n\n`{bundle_error}`"), kind="danger")
if bundle_results is None:
return mo.callout(mo.md(("Upload a bundle `.zip` above to verify it." if not LOCAL_PATHS_ALLOWED else "Upload a bundle `.zip` or pick a folder above to verify it.") + " The manual single-signature check is further down."), kind="neutral")
_c = bundle_results["counts"]
if _c["fail"]:
_kind, _head = "danger", f"❌ {_c['fail']} check{'s' if _c['fail'] != 1 else ''} failed"
elif _c["warn"]:
_kind, _head = "warn", f"⚠️ Passed with {_c['warn']} warning{'s' if _c['warn'] != 1 else ''}"
else:
_kind, _head = "success", "✅ Every check passed"
_summary = mo.callout(
mo.md(
f"## {_head}\n\n**{bundle_results['serial']}** · {bundle_results['n_files']} files · "
+ " · ".join(f"{STATUS_ICON[k]} {v}" for k, v in _c.items() if v)
+ ("" if bundle_results["online"] else "\n\n*Network checks are off: Rekor signatures, Bitcoin anchoring and AIA fetches were not confirmed.*")
),
kind=_kind,
)
_sections = {}
for _chk in bundle_results["checks"]:
_sections.setdefault(_chk.section, []).append(_chk)
_blocks = [_summary]
for _name, _items in _sections.items():
_worst = "fail" if any(i.status == "fail" for i in _items) else ("warn" if any(i.status == "warn" for i in _items) else ("skip" if all(i.status == "skip" for i in _items) else "pass"))
_rows = "\n".join(f"| {STATUS_ICON[i.status]} | {i.name} | {i.detail.replace('|', '\\|')} |" for i in _items)
_blocks.append(mo.accordion({f"{STATUS_ICON[_worst]} {_name}": mo.md(f"| | Check | Detail |\n|:-:|---|---|\n{_rows}")}, multiple=True, lazy=False))
return mo.vstack(_blocks)
_render_bundle()
return
@app.cell(hide_code=True)
def _(
DATA_DIR,
LOCAL_PATHS_ALLOWED,
REPORTS_DIR,
build_customer_pdf_report,
build_pdf_report,
bundle_results,
bundle_source,
copy_button,
mo,
report_json,
report_markdown,
save_report,
):
if bundle_results is not None:
_stem = f"{bundle_results['serial']}-verification-{bundle_results['ran_at'][:19].replace(':', '')}"
_pdf_customer = build_customer_pdf_report(bundle_results)
_pdf = build_pdf_report(bundle_results)
_saved_customer = save_report(_pdf_customer, f"{_stem}-customer.pdf")
_saved = save_report(_pdf, f"{_stem}.pdf")
_items = [
mo.md("### Share the result"),
mo.hstack(
[
mo.download(data=_pdf_customer, filename=f"{_stem}-customer.pdf", mimetype="application/pdf", label="⬇ Customer report (PDF)"),
mo.download(data=_pdf, filename=f"{_stem}.pdf", mimetype="application/pdf", label="⬇ Technical report (PDF)"),
mo.download(data=lambda: report_markdown(bundle_results).encode(), filename=f"{_stem}.md", mimetype="text/markdown", label="⬇ Markdown"),
mo.download(data=lambda: report_json(bundle_results).encode(), filename=f"{_stem}.json", mimetype="application/json", label="⬇ JSON"),
copy_button(report_markdown(bundle_results), "Copy Markdown"),
],
justify="start",
gap=1,
),
]
if _saved is not None:
_items.append(mo.md(f"PDFs saved to `{_saved_customer}` and `{_saved}`."))
if bundle_source and bundle_source.startswith("uploaded"):
_items.append(mo.md(f"Bundle {bundle_source}."))
if LOCAL_PATHS_ALLOWED and DATA_DIR.is_dir() and REPORTS_DIR.is_dir():
_prev = sorted(REPORTS_DIR.glob("*.pdf"), key=lambda p: p.stat().st_mtime, reverse=True)[:10]
if len(_prev) > 1:
_items.append(mo.accordion({f"Earlier reports in {REPORTS_DIR} ({len(_prev)} most recent)": mo.vstack(
[mo.download(data=p.read_bytes(), filename=p.name, mimetype="application/pdf", label=p.name) for p in _prev]
)}))
_report_ui = mo.vstack(_items)
else:
_report_ui = None
_report_ui
return
@app.cell(hide_code=True)
def _(mo):
mo.md("""
---
## 2 · Manual check: one certificate, one file, one signature
1. **Upload the certificate** (`.pem`, `.crt`, `.cer` or `.der`). Its details appear immediately.
2. **Provide what was signed**: upload the original file and the SHA-256 is computed for you, or paste a hash you already have.
3. **Provide the signature**: upload the raw `.sig` file, or paste it as hex / base64.
4. Optionally add the **issuer / CA certificates** (PEM bundle, DER or `.p7s`) to validate the chain.
5. Click **Verify signature**.
""")
return
@app.cell(hide_code=True)
def _(mo):
cert_upload = mo.ui.file(
filetypes=[".pem", ".crt", ".cer", ".der"],
multiple=False,
kind="area",
label="Step 1 · Drop the certificate here (PEM or DER)",
)
cert_upload
return (cert_upload,)
@app.cell(hide_code=True)
def _(
cert_upload,
certificate_summary,
copy_button,
fields_table,
load_certificate,
mo,
serialization,
upload_bytes,
):
_name, _bytes = upload_bytes(cert_upload.value)
cert = None
cert_error = None
if _bytes is not None:
try:
cert = load_certificate(_bytes)
except Exception as _e:
cert_error = f"{_e}"
if cert is None and cert_error is None:
_view = mo.callout(mo.md("Waiting for a certificate. Nothing is sent anywhere; parsing happens locally."), kind="neutral")
elif cert is None:
_view = mo.callout(mo.md(f"**Could not read `{_name}` as a certificate.**\n\n`{cert_error}`\n\nMake sure it is a single X.509 certificate in PEM (`-----BEGIN CERTIFICATE-----`) or DER form, not a private key or a PKCS#7 bundle."), kind="danger")
else:
_fields, _warnings = certificate_summary(cert)
_blocks = [
mo.md(f"### 📜 `{_name}`"),
fields_table(_fields),
mo.hstack(
[
copy_button(_fields["SHA-256 fingerprint"], "Copy SHA-256 fingerprint"),
copy_button(_fields["Serial"], "Copy serial"),
copy_button(_fields["Subject"], "Copy subject"),
copy_button(cert.public_bytes(serialization.Encoding.PEM).decode(), "Copy PEM"),
],
justify="start",
gap=0.5,
),
]
if _warnings:
_blocks.append(mo.callout(mo.md("\n".join(f"- {w}" for w in _warnings)), kind="warn"))
else:
_blocks.append(mo.callout(mo.md("Certificate loaded. It is within its validity period and allowed to sign data."), kind="success"))
_view = mo.vstack(_blocks)
_view
return (cert,)
@app.cell(hide_code=True)
def _(mo):
verify_form = (
mo.md(
"""
### Step 2 · What was signed?
{data_file}
*…or paste the SHA-256 hash if you already have it:*
{sha256_hash}
### Step 3 · The signature
{signature_file}
*…or paste it (hex or base64):*
{signature_text}
### Step 4 · Issuer certificates *(optional, for chain validation)*
{issuer_file}
{check_revocation}
"""
)
.batch(
data_file=mo.ui.file(multiple=True, kind="area", label="Drop the original file(s) — the hash is computed for you"),
sha256_hash=mo.ui.text(
placeholder="e.g. 9dc6d54d8349543fa004fffcd4a36e8fc654bcf2a8217b189f5090bbd1d260fe",
full_width=True,
),
signature_file=mo.ui.file(multiple=True, kind="area", label="Drop the .sig file(s)"),
signature_text=mo.ui.text_area(placeholder="MEUCIBja9Z3OzxHv/UZP3Zg3pI6y9uKmsNG4gevEOWg1hJ1L…", full_width=True),
issuer_file=mo.ui.file(multiple=True, kind="area", label="Drop CA / intermediate certificates (PEM, DER or .p7s)"),
check_revocation=mo.ui.checkbox(label="Also check revocation via CRL and OCSP (needs network and the issuer certificate)"),
)
.form(submit_button_label="Verify signature", bordered=True)
)
verify_form
return (verify_form,)
@app.cell(hide_code=True)
def _(
Check,
STATUS_ICON,
build_chain,
build_pdf_report,
cert,
certificate_summary,
certificate_validity,
check_crl,
check_ocsp,
decode_sha256,
decode_signature,
dt,
fields_table,
hashes,
hashlib,
issued_by,
load_certs_any,
mo,
save_report,
validate_sha256_signature,
verify_form,
):
def _render_result():
if not verify_form.value:
return mo.callout(mo.md("Fill in steps 1–3 and click **Verify signature**."), kind="neutral")
if cert is None:
return mo.callout(mo.md("**Upload a valid certificate in Step 1 first.**"), kind="warn")
_v = verify_form.value
_problems = []
# --- candidate data (one or more files, or a pasted hash) -----------------
_data_cands = [] # (digest, source)
for _f in _v["data_file"] or []:
_data_cands.append((hashlib.sha256(_f.contents).digest(), f"computed from `{_f.name}` ({len(_f.contents):,} bytes)"))
if _v["sha256_hash"].strip():
try:
_pasted = decode_sha256(_v["sha256_hash"])
if _data_cands and all(d != _pasted for d, _ in _data_cands):
_problems.append("The pasted hash does **not** match any uploaded file. The uploaded file(s) were used.")
elif not _data_cands:
_data_cands.append((_pasted, "pasted"))
except ValueError as _e:
if not _data_cands:
return mo.callout(mo.md(f"**Invalid SHA-256 hash.** {_e}"), kind="danger")
_problems.append(f"Ignored the pasted hash: {_e}")
if not _data_cands:
return mo.callout(mo.md("**Step 2 is empty.** Upload the signed file or paste its SHA-256 hash."), kind="warn")
# --- candidate signatures (one or more .sig files, or pasted text) --------
_sig_cands = [] # (bytes, encoding, source)
for _f in _v["signature_file"] or []:
try:
_s, _enc = decode_signature(_f.contents)
if _s:
_sig_cands.append((_s, _enc, f"`{_f.name}`"))
except ValueError as _e:
_problems.append(f"Ignored `{_f.name}`: {_e}")
if _v["signature_text"].strip():
try:
_s, _enc = decode_signature(_v["signature_text"])
_sig_cands.append((_s, _enc, "pasted"))
except ValueError as _e:
if not _sig_cands:
return mo.callout(mo.md(f"**Invalid signature.** {_e}"), kind="danger")
_problems.append(f"Ignored the pasted signature: {_e}")
if not _sig_cands:
return mo.callout(mo.md("**Step 3 is empty.** Upload the `.sig` file or paste the signature."), kind="warn")
# --- verify every (data, signature) pair; first match wins ----------------
_matched, _tried = None, []
_digest, _hash_source = _data_cands[0]
_sig, _sig_enc, _sig_source = _sig_cands[0]
for _d, _dsrc in _data_cands:
for _s, _enc, _ssrc in _sig_cands:
_m, _t = validate_sha256_signature(cert, _d, _s)
if _m:
_matched, _tried = _m, _t
_digest, _hash_source, _sig, _sig_enc, _sig_source = _d, _dsrc, _s, _enc, _ssrc
break
if (_d, _s) == (_data_cands[0][0], _sig_cands[0][0]):
_tried = _t
if _matched:
break
_n_pairs = len(_data_cands) * len(_sig_cands)
if _n_pairs > 1:
_problems.append(f"Tried {_n_pairs} file × signature combinations; " + ("showing the one that verified." if _matched else "none verified."))
_fields, _warnings = certificate_summary(cert)
_summary = {
"Signed data": f"SHA-256 {_digest.hex()} ({_hash_source})",
"Signature": f"{len(_sig)} bytes, {_sig_enc} ({_sig_source})",
"Signer": _fields["Subject"],
"Signer key": _fields["Public key"],
"Scheme": _matched or "no scheme matched",
}
if _matched:
_banner = mo.callout(
mo.md("## ✅ Signature is valid\n\nThe signature was produced by the private key belonging to this certificate, over exactly this data."),
kind="success",
)
else:
_banner = mo.callout(
mo.md(
"## ❌ Signature is NOT valid\n\n"
"Common causes:\n\n"
"- **Wrong certificate** – a proof bundle usually has several; try the *signing* certificate (e.g. `item-signing-certificate.pem`).\n"
"- **Wrong file** – the signature covers a specific file (e.g. `provenance.json`), not the PDF or manifest.\n"
"- **Modified data** – even a one-byte change (line endings, re-saving) changes the hash.\n"
"- **Wrong signature file** – e.g. `provenance.workshop.sig` vs `provenance.item.sig`."
),
kind="danger",
)
# --- chain (optional) -------------------------------------------------
_chain_block = None
_pool = []
for _f in _v["issuer_file"] or []:
try:
_pool += load_certs_any(_f.contents)
except Exception as _e:
_problems.append(f"Could not read issuer file `{_f.name}`: {_e}")
if _pool:
_hops = build_chain(cert, _pool)
_rows = []
for _c, _st in _hops:
_icon = {"ok": "✅", "self-signed": "🔒", "missing": "❌"}[_st]
_nb, _na = certificate_validity(_c)
_rows.append(f"| {_icon} | `{_c.subject.rfc4514_string()}` | {_st} | {_nb:%Y-%m-%d} → {_na:%Y-%m-%d} | `{_c.fingerprint(hashes.SHA256()).hex()[:16]}…` |")
_rooted = _hops[-1][1] == "self-signed"
_chain_block = mo.vstack(
[
mo.md("### Certificate chain"),
mo.md("| | Certificate | Link | Validity | SHA-256 |\n|:-:|---|---|---|---|\n" + "\n".join(_rows)),
mo.callout(
mo.md(
"Chain reaches a self-signed root. Pin that root's fingerprint out of band to complete trust."
if _rooted
else "**Chain is incomplete.** No uploaded certificate signs the last link; add the missing CA certificate."
),
kind="success" if _rooted else "warn",
),
]
)
# --- revocation (optional, online) -----------------------------------------
_rev_block = None
_rev_checks = []
if _v.get("check_revocation"):
_issuer = next((c for c in _pool if c.subject == cert.issuer and issued_by(cert, c)), None) if _pool else None
if _issuer is None:
_rev_block = mo.callout(mo.md("**Revocation not checked.** Add the certificate's issuer (CA) in Step 4 so a CRL / OCSP query can be built."), kind="warn")
else:
with mo.status.spinner(title="Querying CRL and OCSP…"):
_rev_checks = [Check("Revocation (CRL / OCSP)", "CRL", *check_crl(cert, _issuer)), Check("Revocation (CRL / OCSP)", "OCSP", *check_ocsp(cert, _issuer))]
_rev_rows = "\n".join(f"| {STATUS_ICON[c.status]} | {c.name} | {c.detail} |" for c in _rev_checks)
_rev_kind = "danger" if any(c.status == "fail" for c in _rev_checks) else ("warn" if any(c.status == "warn" for c in _rev_checks) else "success")
_rev_block = mo.vstack([mo.md("### Revocation"), mo.callout(mo.md(f"| | Check | Detail |\n|:-:|---|---|\n{_rev_rows}"), kind=_rev_kind)])
_notes = ["The signature only attests to the SHA-256 hash; make sure the hash was computed from the intended data."] + _warnings
# --- PDF report for this single verification -----------------------------
_checks = [Check("Signature", "Signature verifies", "pass" if _matched else "fail", f"scheme: {_matched}" if _matched else "no scheme matched"),
Check("Signature", "Signed data", "info", _summary["Signed data"]),
Check("Signature", "Signature", "info", _summary["Signature"]),
Check("Signature", "Signer", "info", f"{_summary['Signer']} · {_summary['Signer key']}")]
_checks += [Check("Certificate", "Warning", "warn", w) for w in _warnings]
_checks += _rev_checks
_checks += [Check("Input", "Note", "warn", p) for p in _problems]
_cert_records = [{"name": "signer certificate", **_fields, "warnings": _warnings}]
if _pool:
for _c, _st in _hops[1:]:
_f2, _w2 = certificate_summary(_c)
_cert_records.append({"name": _c.subject.rfc4514_string(), "source": "uploaded issuer", **_f2, "warnings": _w2})
_checks += [Check("Certificate chain", _c.subject.rfc4514_string().split(",")[0], "pass" if _st != "missing" else "fail", _st) for _c, _st in _hops]
_counts = {k: sum(1 for c in _checks if c.status == k) for k in STATUS_ICON}
_results = {"serial": _fields["Subject"].split(",")[0].replace("CN=", ""), "path": f"manual check · {_hash_source} · signature {_sig_source}",
"ran_at": dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"), "online": False, "counts": _counts, "checks": _checks,
"certificates": _cert_records, "subtitle": f"Detached signature check for **{_fields['Subject']}**"}
_stem = f"signature-check-{_results['serial']}-{_results['ran_at'][:19].replace(':', '')}"
_pdf = build_pdf_report(_results, title="X.509 Signature Verification Report")
_saved = save_report(_pdf, f"{_stem}.pdf")
_pdf_row = mo.hstack([mo.download(data=_pdf, filename=f"{_stem}.pdf", mimetype="application/pdf", label="⬇ PDF report")]
+ ([mo.md(f"saved to `{_saved}`")] if _saved else []), justify="start", gap=1)
_blocks = [_banner, _pdf_row]
if _problems:
_blocks.append(mo.callout(mo.md("\n".join(f"- ⚠️ {p}" for p in _problems)), kind="warn"))
_blocks.append(fields_table(_summary))
if _chain_block is not None:
_blocks.append(_chain_block)
if _rev_block is not None:
_blocks.append(_rev_block)
_blocks.append(
mo.accordion(
{
"Notes": mo.md("\n".join(f"- {n}" for n in _notes)),
"Certificate details": fields_table(_fields),
f"Schemes tried ({len(_tried) + (1 if _matched else 0)})": mo.md(
"\n".join([f"- ✅ {_matched}"] if _matched else []) + "\n" + "\n".join(f"- ❌ {t}" for t in _tried)
),
}
)
)
return mo.vstack(_blocks)
_render_result()
return
@app.cell
def _(
InvalidSignature,
base64,
build_chain,
certificate_summary,
certificate_validity,
decode_signature,
dsa,
dt,
ec,
hashes,
hashlib,
issued_by,
load_certificate,
load_certs_any,
mo,
padding,
pathlib,
rsa,
serialization,
validate_sha256_signature,
x509,
):
import io
import json
import os
import re
import tempfile
import zipfile
import urllib.request
import urllib.error
from dataclasses import dataclass, asdict
from asn1crypto import cms as a_cms
from asn1crypto import tsp as a_tsp
from opentimestamps.core.timestamp import DetachedTimestampFile
from opentimestamps.core.serialize import BytesDeserializationContext
from opentimestamps.core.notary import BitcoinBlockHeaderAttestation, PendingAttestation
DATA_DIR = pathlib.Path(os.environ.get("VERIFIER_DATA_DIR", "/data"))
BUNDLES_DIR = DATA_DIR / "bundles"
REPORTS_DIR = DATA_DIR / "reports"
MAX_ZIP_BYTES = 200 * 1024 * 1024
# Browsing or typing arbitrary server paths is only offered in the local editor (or when explicitly enabled).
# In the deployed app visitors can only verify bundles they upload as .zip.
LOCAL_PATHS_ALLOWED = mo.app_meta().mode == "edit" or os.environ.get("VERIFIER_ALLOW_LOCAL_PATHS") == "1"
REKOR_URL = "https://rekor.sigstore.dev"
BLOCK_API = "https://blockstream.info/api"
@dataclass
class Check:
section: str
name: str
status: str # pass | fail | warn | info | skip
detail: str = ""
STATUS_ICON = {"pass": "✅", "fail": "❌", "warn": "⚠️", "info": "ℹ️", "skip": "⏭️"}
def http_get(url, timeout=12):
_req = urllib.request.Request(url, headers={"User-Agent": "marimo-proof-verifier/1.0"})
with urllib.request.urlopen(_req, timeout=timeout) as _r:
return _r.read()
def _cert_label(cert):
return cert.subject.rfc4514_string()
def _short_fp(cert):
return cert.fingerprint(hashes.SHA256()).hex()[:16]
def load_bundle(path):
"""Read every file in a proof-bundle folder. Returns a dict used by the check functions."""
path = pathlib.Path(path).expanduser().resolve()
if not path.is_dir():
raise NotADirectoryError(f"{path} is not a folder")
files = {p.name: p.read_bytes() for p in sorted(path.iterdir()) if p.is_file() and not p.name.startswith(".")}
digests = {n: hashlib.sha256(d).hexdigest() for n, d in files.items()}
by_hash = {}
for n, h in digests.items():
by_hash.setdefault(h, []).append(n)
def _json(name):
try:
return json.loads(files[name]) if name in files else None
except Exception:
return None
certs = {}
for n, d in files.items():
if n.lower().endswith((".pem", ".crt", ".cer", ".der")):
try:
certs[n] = load_certificate(d)
except Exception:
pass
embedded = {}
for n, d in files.items():
if n.lower().endswith((".p7s", ".p7b", ".tsr")):
try:
for c in load_certs_any(d):
embedded.setdefault(c.fingerprint(hashes.SHA256()), (c, n))
except Exception:
pass
pool = list(certs.values()) + [c for c, _ in embedded.values()]
return {
"path": path, "files": files, "digests": digests, "by_hash": by_hash,
"manifest": _json("manifest.json"), "provenance": _json("provenance.json"),
"certs": certs, "embedded": embedded, "pool": pool,
}
def _files_for_hash(bundle, hex_digest):
_names = bundle["by_hash"].get(hex_digest.lower(), [])
return ", ".join(f"`{n}`" for n in _names) if _names else None
def extract_bundle_zip(name, data):
"""Safely extract an uploaded bundle .zip into BUNDLES_DIR (or a temp dir) and return the folder.
Rejects path traversal and oversized archives; strips a single top-level folder if the zip has one."""
stem = re.sub(r"[^A-Za-z0-9._-]+", "-", pathlib.Path(name).stem).strip("-") or "bundle"
root = BUNDLES_DIR if DATA_DIR.is_dir() else pathlib.Path(tempfile.gettempdir()) / "verifier-bundles"
root.mkdir(parents=True, exist_ok=True)
target = root / stem
with zipfile.ZipFile(io.BytesIO(data)) as zf:
members = [m for m in zf.infolist() if not m.is_dir() and not m.filename.startswith("__MACOSX/") and not pathlib.Path(m.filename).name.startswith(".")]
total = sum(m.file_size for m in members)
if total > MAX_ZIP_BYTES:
raise ValueError(f"archive expands to {total:,} bytes (limit {MAX_ZIP_BYTES:,})")
tops = {pathlib.PurePosixPath(m.filename).parts[0] for m in members if len(pathlib.PurePosixPath(m.filename).parts) > 1}
strip = tops.pop() if len(tops) == 1 and all(len(pathlib.PurePosixPath(m.filename).parts) > 1 for m in members) else None
plan = []
for m in members:
rel = pathlib.PurePosixPath(m.filename)
if strip and rel.parts[0] == strip:
rel = pathlib.PurePosixPath(*rel.parts[1:])
if rel.is_absolute() or ".." in rel.parts or not rel.parts:
raise ValueError(f"unsafe path in archive: {m.filename}")
dest = (target / pathlib.Path(*rel.parts)).resolve()
if target.resolve() not in dest.parents:
raise ValueError(f"unsafe path in archive: {m.filename}")
plan.append((m, dest))
if target.exists():
for p in sorted(target.rglob("*"), reverse=True):
p.unlink() if p.is_file() else p.rmdir()
target.mkdir(parents=True, exist_ok=True)
for m, dest in plan:
dest.parent.mkdir(parents=True, exist_ok=True)
with zf.open(m) as src, open(dest, "wb") as out:
out.write(src.read())
return target
def uploaded_bundles():
"""{folder name: path} for bundles previously extracted into BUNDLES_DIR (or the temp fallback)."""
root = BUNDLES_DIR if DATA_DIR.is_dir() else pathlib.Path(tempfile.gettempdir()) / "verifier-bundles"
if not root.is_dir():
return {}
return {p.name: p for p in sorted(root.iterdir(), key=lambda p: p.stat().st_mtime, reverse=True) if p.is_dir()}
def is_allowed_bundle_dir(path):
"""In locked mode only folders inside the upload area may be verified."""
if LOCAL_PATHS_ALLOWED:
return True
root = (BUNDLES_DIR if DATA_DIR.is_dir() else pathlib.Path(tempfile.gettempdir()) / "verifier-bundles").resolve()
try:
return root in pathlib.Path(path).resolve().parents
except OSError:
return False
def certificate_records(bundle):
"""Certificate summaries for the report appendix: bundle PEM/DER files plus certs embedded in CMS/TSR blobs."""
records = []
for n, c in bundle["certs"].items():
fields, warnings = certificate_summary(c)
records.append({"name": n, "source": "bundle file", **fields, "warnings": warnings})
seen = {c.fingerprint(hashes.SHA256()) for c in bundle["certs"].values()}
for fp, (c, src) in bundle["embedded"].items():
if fp in seen:
continue
fields, warnings = certificate_summary(c)
records.append({"name": _cert_label(c), "source": f"embedded in {src}", **fields, "warnings": warnings})
return records
# ---------------------------------------------------------------- manifest + cross references
def check_manifest(bundle):
S = "Manifest"
m = bundle["manifest"]
if m is None:
return [Check(S, "manifest.json", "fail", "Missing or not valid JSON")]
out = [Check(S, "Serial", "info", f"`{m.get('serial')}` · generated {m.get('generated_at')} · access tier **{m.get('access_tier')}**")]
listed = set()
for f in m.get("files", []):
name = f.get("filename"); listed.add(name)
if name not in bundle["files"]:
out.append(Check(S, name, "fail", "listed in manifest but missing from folder"))
continue
ok_hash = bundle["digests"][name] == (f.get("sha256") or "").lower()
ok_size = len(bundle["files"][name]) == f.get("bytes")
if ok_hash and ok_size:
out.append(Check(S, name, "pass", f"{f.get('bytes'):,} bytes · sha256 matches · {f.get('access')}"))
else:
out.append(Check(S, name, "fail", ("size mismatch " if not ok_size else "") + ("sha256 mismatch" if not ok_hash else "")))
for name in m.get("unavailable", []) or []:
out.append(Check(S, name, "info", "declared unavailable by the issuer"))
extra = sorted(n for n in bundle["files"] if n not in listed and n != "manifest.json")
if extra:
out.append(Check(S, "Unlisted files", "warn", ", ".join(f"`{n}`" for n in extra)))
return out
def check_cross_refs(bundle):
S = "Provenance record"
p = bundle["provenance"]
if p is None:
return [Check(S, "provenance.json", "skip", "not present in this bundle")]
out = []
item = p.get("item", {})
out.append(Check(S, "Record", "info", f"**{item.get('title')}** ({item.get('type')}) · issuer **{p.get('issuer')}** · registered {p.get('registered_at')} · format `{p.get('format')}`"))
serials = {"provenance.json": p.get("serial"), "manifest.json": (bundle["manifest"] or {}).get("serial")}
for n, c in bundle["certs"].items():
try:
serials[n] = c.subject.get_attributes_for_oid(x509.oid.NameOID.COMMON_NAME)[0].value
except Exception:
pass
vals = {v for v in serials.values() if v}
out.append(Check(S, "Serial agrees across files", "pass" if len(vals) == 1 else "fail", " · ".join(f"`{n}`: {v}" for n, v in serials.items())))
art = p.get("artifacts", {})
coa = art.get("coa_sha256")
if coa:
hit = _files_for_hash(bundle, coa)
out.append(Check(S, "COA PDF hash", "pass" if hit else "fail", f"`{coa[:16]}…` → {hit or 'no file in the bundle has this hash'}"))
ics = art.get("item_cert_serial")
if ics:
hit = [n for n, c in bundle["certs"].items() if format(c.serial_number, "x") == ics.lower()]
out.append(Check(S, "Item certificate serial", "pass" if hit else "fail", f"`{ics}` → " + (", ".join(f"`{n}`" for n in hit) or "no certificate with this serial")))
serial = p.get("serial")
for n, c in bundle["certs"].items():
try:
san = c.extensions.get_extension_for_class(x509.SubjectAlternativeName).value
uris = san.get_values_for_type(x509.UniformResourceIdentifier)
ok = any(serial in u for u in uris)
out.append(Check(S, f"SAN of {n}", "pass" if ok else "warn", ", ".join(f"`{u}`" for u in uris) or "no URIs"))
except x509.ExtensionNotFound:
pass
if p.get("verify_url"):
out.append(Check(S, "Verify URL", "info", f"<{p['verify_url']}>"))
return out
# ---------------------------------------------------------------- detached signatures
def check_detached_signatures(bundle):
S = "Detached signatures"
sigs = {n: d for n, d in bundle["files"].items() if n.lower().endswith(".sig")}
if not sigs:
return [Check(S, "No .sig files", "skip", "")]
out = []
cand_files = {n: d for n, d in bundle["files"].items() if not n.lower().endswith((".sig", ".p7s", ".ots", ".tsr"))}
cert_items = list(bundle["certs"].items()) + [(f"{src} (embedded: {_cert_label(c)})", c) for c, src in bundle["embedded"].values()]
for sname, sdata in sigs.items():
try:
sig, enc = decode_signature(sdata)
except ValueError as e:
out.append(Check(S, sname, "fail", str(e))); continue
found = None
for cname, cert in cert_items:
for fname, fdata in cand_files.items():
digest = hashlib.sha256(fdata).digest()
matched, _ = validate_sha256_signature(cert, digest, sig)
if matched:
found = (cname, fname, matched); break
if found:
break
if found:
out.append(Check(S, sname, "pass", f"signs `{found[1]}` with **{found[0]}** · {found[2]} · {len(sig)} bytes ({enc})"))
else:
out.append(Check(S, sname, "fail", f"no (certificate, file) pair in the bundle verifies this {len(sig)}-byte signature"))
return out
# ---------------------------------------------------------------- CMS / PKCS#7
def cms_verify(p7s_bytes, content=None):
"""Verify a CMS SignedData. For detached signatures pass `content`. Returns (results, certs)."""
ci = a_cms.ContentInfo.load(p7s_bytes)
if ci["content_type"].native != "signed_data":
raise ValueError(f"not SignedData ({ci['content_type'].native})")
sd = ci["content"]
certs = [x509.load_der_x509_certificate(c.chosen.dump()) for c in sd["certifi" "cates"]] if sd["certificates"] else []
eci = sd["encap_content_info"]
if content is None and eci["content"].native is not None:
content = eci["content"].native
results = []
for si in sd["signer_infos"]:
sid = si["sid"]
if sid.name == "issuer_and_serial_number":
serial = sid.chosen["serial_number"].native
signer = next((c for c in certs if c.serial_number == serial), None)
else:
ski = sid.chosen.native
signer = next((c for c in certs if c.extensions.get_extension_for_class(x509.SubjectKeyIdentifier).value.digest == ski), None)
dalg = si["digest_algorithm"]["algorithm"].native
hcls = {"sha256": hashes.SHA256, "sha384": hashes.SHA384, "sha512": hashes.SHA512, "sha1": hashes.SHA1}[dalg]
sattrs = si["signed_attrs"]
md = next((a["values"][0].native for a in sattrs if a["type"].native == "message_digest"), None)
stime = next((a["values"][0].native for a in sattrs if a["type"].native == "signing_time"), None)
md_ok = (content is not None and md is not None and hashlib.new(dalg, content).digest() == md)
tbs = b"\x31" + sattrs.dump()[1:]
sig = si["signature"].native
sig_ok, err = None, None
if signer is not None:
pub = signer.public_key()
try:
if isinstance(pub, ec.EllipticCurvePublicKey):
pub.verify(sig, tbs, ec.ECDSA(hcls()))
elif isinstance(pub, rsa.RSAPublicKey):
pub.verify(sig, tbs, padding.PKCS1v15(), hcls())
else:
pub.verify(sig, tbs)
sig_ok = True
except InvalidSignature:
sig_ok = False
except Exception as e:
err = str(e)
results.append({"signer": signer, "digest_alg": dalg, "message_digest": md.hex() if md else None,
"content_matches": md_ok, "signature_ok": sig_ok, "error": err, "signing_time": stime,
"content_type": eci["content_type"].native})
return results, certs
def check_cms(bundle):
S = "CMS / PKCS#7 signatures"
p7s = {n: d for n, d in bundle["files"].items() if n.lower().endswith(".p7s")}
if not p7s:
return [Check(S, "No .p7s files", "skip", "")]
out = []
for name, data in p7s.items():
try:
ci = a_cms.ContentInfo.load(data)
sd = ci["content"]
md_hex = None
for si in sd["signer_infos"]:
md_hex = next((a["values"][0].native.hex() for a in si["signed_attrs"] if a["type"].native == "message_digest"), None)
target = bundle["by_hash"].get(md_hex, [None])[0] if md_hex else None
content = bundle["files"].get(target) if target else None
results, certs = cms_verify(data, content)
except Exception as e:
out.append(Check(S, name, "fail", f"could not parse: {e}")); continue
for r in results:
signer = r["signer"]
label = _cert_label(signer) if signer else "unknown signer (certificate not embedded)"
if target is None:
out.append(Check(S, name, "fail", f"messageDigest `{(r['message_digest'] or '')[:16]}…` matches no file in the bundle"))
continue
ok = r["content_matches"] and r["signature_ok"]
out.append(Check(S, name, "pass" if ok else "fail",
f"signs `{target}` · signer **{label}** · {r['digest_alg']} · "
f"content {'✓' if r['content_matches'] else '✗'} · signature {'✓' if r['signature_ok'] else ('✗' if r['signature_ok'] is False else r['error'])}"
+ (f" · signed at {r['signing_time']:%Y-%m-%d %H:%M} UTC" if r["signing_time"] else "")))
if signer is not None:
hops = build_chain(signer, certs + bundle["pool"])
path = " → ".join(f"{_cert_label(c).split(',')[0]} [{st}]" for c, st in hops)
ok_chain = hops[-1][1] == "self-signed"
out.append(Check(S, f"{name} · embedded chain", "pass" if ok_chain else "warn", path + (" · root fingerprint `" + _short_fp(hops[-1][0]) + "…`" if ok_chain else "")))
return out
# ---------------------------------------------------------------- RFC 3161 timestamp
def parse_timestamp_token(data):
"""Accept a TimeStampResp or a bare TimeStampToken. Returns (token ContentInfo, TSTInfo)."""
try:
resp = a_tsp.TimeStampResp.load(data)
status = resp["status"]["status"].native
if status not in ("granted", "granted_with_mods"):
raise ValueError(f"TSA status: {status}")
token = resp["time_stamp_token"]
token.native # force parse
except Exception:
token = a_cms.ContentInfo.load(data)
tst = token["content"]["encap_content_info"]["content"].parsed
return token, tst
def check_timestamp(bundle):
S = "RFC 3161 timestamp"
tsrs = {n: d for n, d in bundle["files"].items() if n.lower().endswith((".tsr", ".tst", ".tsq"))}
if not tsrs:
return [Check(S, "No .tsr files", "skip", "")]
out = []
for name, data in tsrs.items():
try:
token, tst = parse_timestamp_token(data)
except Exception as e:
out.append(Check(S, name, "fail", f"could not parse: {e}")); continue
imprint = tst["message_imprint"]["hashed_message"].native.hex()
halg = tst["message_imprint"]["hash_algorithm"]["algorithm"].native
gen = tst["gen_time"].native
hit = _files_for_hash(bundle, imprint) if halg == "sha256" else None
out.append(Check(S, f"{name} · covers", "pass" if hit else "fail", f"{halg} `{imprint[:16]}…` → {hit or 'no file in the bundle has this hash'}"))
acc = tst["accuracy"].native if tst["accuracy"].native else None
out.append(Check(S, f"{name} · time", "info", f"**{gen:%Y-%m-%d %H:%M:%S} UTC** · policy `{tst['policy'].native}` · serial `{format(tst['serial_number'].native, 'x')[:20]}…`" + (f" · accuracy {acc}" if acc else "")))
try:
results, certs = cms_verify(token.dump(), tst.dump())
r = results[0]
signer = r["signer"]
out.append(Check(S, f"{name} · TSA signature", "pass" if (r["content_matches"] and r["signature_ok"]) else "fail",
f"TSA **{_cert_label(signer) if signer else 'unknown'}** · {r['digest_alg']} · content {'✓' if r['content_matches'] else '✗'} · signature {'✓' if r['signature_ok'] else '✗'}"))
if signer is not None:
hops = build_chain(signer, certs + bundle["pool"])
ok_chain = hops[-1][1] == "self-signed"
last = hops[-1][0]
out.append(Check(S, f"{name} · TSA chain", "pass" if ok_chain else "warn",
" → ".join(_cert_label(c).split(",")[0] for c, _ in hops)
+ (f" · root `{_short_fp(last)}…` (trust this root out of band)" if ok_chain
else f" · ends at a certificate issued by **{last.issuer.rfc4514_string().split(',')[0]}**, which is not in the bundle (cross-signed root?) — enable network checks to fetch it")))
try:
eku = signer.extensions.get_extension_for_class(x509.ExtendedKeyUsage).value
ok_eku = x509.oid.ExtendedKeyUsageOID.TIME_STAMPING in eku
out.append(Check(S, f"{name} · TSA key usage", "pass" if ok_eku else "fail", "timeStamping EKU " + ("present" if ok_eku else "missing")))
except x509.ExtensionNotFound:
out.append(Check(S, f"{name} · TSA key usage", "warn", "no EKU extension"))
nb, na = certificate_validity(signer)
out.append(Check(S, f"{name} · TSA cert valid at signing", "pass" if nb <= gen <= na else "fail", f"{nb:%Y-%m-%d} → {na:%Y-%m-%d}"))
except Exception as e:
out.append(Check(S, f"{name} · TSA signature", "fail", f"{e}"))
return out
# ---------------------------------------------------------------- OpenTimestamps
def check_ots(bundle, online):
S = "OpenTimestamps"
otss = {n: d for n, d in bundle["files"].items() if n.lower().endswith(".ots")}
if not otss:
return [Check(S, "No .ots files", "skip", "")]
out = []
for name, data in otss.items():
try:
dtf = DetachedTimestampFile.deserialize(BytesDeserializationContext(data))
except Exception as e:
out.append(Check(S, name, "fail", f"could not parse: {e}")); continue
digest = dtf.file_digest.hex()
hit = _files_for_hash(bundle, digest)
out.append(Check(S, f"{name} · covers", "pass" if hit else "fail", f"{dtf.file_hash_op} `{digest[:16]}…` → {hit or 'no file in the bundle has this hash'}"))
atts = list(dtf.timestamp.all_attestations())
if not atts:
out.append(Check(S, f"{name} · attestations", "fail", "none"))
for msg, att in atts:
if isinstance(att, BitcoinBlockHeaderAttestation):
detail = f"Bitcoin block **{att.height}** · expected merkle root `{msg[::-1].hex()[:16]}…`"
if not online:
out.append(Check(S, f"{name} · Bitcoin", "info", detail + " · enable network checks to confirm against the chain"))
continue
try:
bh = http_get(f"{BLOCK_API}/block-height/{att.height}").decode()
blk = json.loads(http_get(f"{BLOCK_API}/block/{bh}"))
ok = blk.get("merkle_root") == msg[::-1].hex()
t = dt.datetime.fromtimestamp(blk["timestamp"], dt.timezone.utc)
out.append(Check(S, f"{name} · Bitcoin", "pass" if ok else "fail", detail + f" · block `{bh[:12]}…` mined **{t:%Y-%m-%d %H:%M} UTC** · merkle root {'matches' if ok else 'DOES NOT match'}"))
except Exception as e:
out.append(Check(S, f"{name} · Bitcoin", "warn", detail + f" · lookup failed: {e}"))
elif isinstance(att, PendingAttestation):
out.append(Check(S, f"{name} · pending", "warn", f"not yet anchored · calendar `{att.uri}` · run `ots upgrade {name}` later"))
else:
out.append(Check(S, f"{name} · attestation", "info", type(att).__name__))
return out
# ---------------------------------------------------------------- Rekor transparency log
def _rfc6962_leaf(data):
return hashlib.sha256(b"\x00" + data).digest()
def _rfc6962_node(left, right):
return hashlib.sha256(b"\x01" + left + right).digest()
def verify_inclusion_proof(leaf_hash, index, tree_size, proof, root):
"""RFC 9162 §2.1.3.2 inclusion-proof verification."""
fn, sn, r = index, tree_size - 1, leaf_hash
for p in proof:
if sn == 0:
return False
if fn & 1 or fn == sn:
r = _rfc6962_node(p, r)
while not (fn & 1 or fn == 0):
fn >>= 1; sn >>= 1
else:
r = _rfc6962_node(r, p)
fn >>= 1; sn >>= 1
return sn == 0 and r == root
def check_rekor(bundle, online):
S = "Rekor transparency log"
rk = {n: d for n, d in bundle["files"].items() if n.lower().startswith("rekor") and n.lower().endswith(".json")}
if not rk:
return [Check(S, "No rekor.*.json files", "skip", "")]
out = []
rekor_pub = None
if online:
try:
rekor_pub = serialization.load_pem_public_key(http_get(f"{REKOR_URL}/api/v1/log/publicKey"))
except Exception as e:
out.append(Check(S, "Rekor public key", "warn", f"could not fetch: {e}"))
sig_files = {n: d for n, d in bundle["files"].items() if n.lower().endswith(".sig")}
for name, data in rk.items():
try:
doc = json.loads(data)
uuid, e = next(iter(doc["entry"].items()))
body_b64 = e["body"]; body = base64.b64decode(body_b64); bj = json.loads(body)
except Exception as ex:
out.append(Check(S, name, "fail", f"could not parse: {ex}")); continue
kind = bj.get("kind"); spec = bj.get("spec", {})
t = dt.datetime.fromtimestamp(e["integratedTime"], dt.timezone.utc)
out.append(Check(S, f"{name} · entry", "info", f"kind `{kind}` · log index **{e['logIndex']}** · integrated **{t:%Y-%m-%d %H:%M:%S} UTC** · <{doc.get('entry_url', '')}>"))
if kind == "hashedrekord":
hv = spec["data"]["hash"]["value"]
hit = _files_for_hash(bundle, hv)
out.append(Check(S, f"{name} · logged hash", "pass" if hit else "fail", f"`{hv[:16]}…` → {hit or 'no file in the bundle has this hash'}"))
sig_b = base64.b64decode(spec["signature"]["content"])
sig_hit = [n for n, d in sig_files.items() if d == sig_b]
out.append(Check(S, f"{name} · logged signature", "pass" if sig_hit else "warn", "equals " + ", ".join(f"`{n}`" for n in sig_hit) if sig_hit else "does not match any .sig file byte-for-byte"))
try:
pk_pem = base64.b64decode(spec["signature"]["publicKey"]["content"])
lc = load_certs_any(pk_pem)[0]
fp = lc.fingerprint(hashes.SHA256())
known = [n for n, c in bundle["certs"].items() if c.fingerprint(hashes.SHA256()) == fp]
emb = bundle["embedded"].get(fp)
src = ", ".join(f"`{n}`" for n in known) or (f"embedded in `{emb[1]}`" if emb else None)
out.append(Check(S, f"{name} · logged certificate", "pass" if src else "warn", f"**{_cert_label(lc)}** · " + (src or "not found in this bundle")))
if hit:
fname = bundle["by_hash"][hv][0]
matched, _ = validate_sha256_signature(lc, hashlib.sha256(bundle["files"][fname]).digest(), sig_b)
out.append(Check(S, f"{name} · logged signature verifies", "pass" if matched else "fail", f"over `{fname}` with logged certificate" + (f" · {matched}" if matched else "")))
except Exception as ex:
out.append(Check(S, f"{name} · logged certificate", "warn", f"could not load public key: {ex}"))
leaf = _rfc6962_leaf(body)
out.append(Check(S, f"{name} · entry UUID", "pass" if uuid.endswith(leaf.hex()) else "fail", f"`{uuid[:16]}…` ends with the Merkle leaf hash" if uuid.endswith(leaf.hex()) else "UUID does not match sha256(0x00 ‖ body)"))
ip = e.get("verification", {}).get("inclusionProof")
if ip:
ok = verify_inclusion_proof(leaf, ip["logIndex"], ip["treeSize"], [bytes.fromhex(h) for h in ip["hashes"]], bytes.fromhex(ip["rootHash"]))
out.append(Check(S, f"{name} · inclusion proof", "pass" if ok else "fail", f"{len(ip['hashes'])} hashes · tree size {ip['treeSize']:,} · root `{ip['rootHash'][:16]}…`"))
cp = ip.get("checkpoint", "")
try:
note, sigline = cp.split("\n\n", 1); note += "\n"
origin, size, root_b64 = note.split("\n")[:3]
cp_ok = base64.b64decode(root_b64).hex() == ip["rootHash"] and int(size) == ip["treeSize"]
out.append(Check(S, f"{name} · checkpoint", "pass" if cp_ok else "fail", f"`{origin}` · size {size} · root {'matches proof' if cp_ok else 'MISMATCH'}"))
if rekor_pub is not None:
sigb = base64.b64decode(sigline.strip().split(" ", 2)[2])
try:
rekor_pub.verify(sigb[4:], note.encode(), ec.ECDSA(hashes.SHA256())); cps = "pass"; cpd = "signed by rekor.sigstore.dev"
except InvalidSignature:
cps = "fail"; cpd = "checkpoint signature INVALID"
out.append(Check(S, f"{name} · checkpoint signature", cps, cpd))
except Exception as ex:
out.append(Check(S, f"{name} · checkpoint", "warn", f"could not parse: {ex}"))
else:
out.append(Check(S, f"{name} · inclusion proof", "warn", "not present"))
set_b64 = e.get("verification", {}).get("signedEntryTimestamp")
if set_b64 and rekor_pub is not None:
payload = json.dumps({"body": body_b64, "integratedTime": e["integratedTime"], "logID": e["logID"], "logIndex": e["logIndex"]}, separators=(",", ":"), sort_keys=True).encode()
try:
rekor_pub.verify(base64.b64decode(set_b64), payload, ec.ECDSA(hashes.SHA256())); st, sd_ = "pass", "signed entry timestamp verified with Rekor's key"
except InvalidSignature:
st, sd_ = "fail", "signed entry timestamp INVALID"
out.append(Check(S, f"{name} · SET", st, sd_))
spki = rekor_pub.public_bytes(serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo)
out.append(Check(S, f"{name} · log ID", "pass" if hashlib.sha256(spki).hexdigest() == e["logID"] else "fail", f"`{e['logID'][:16]}…` = sha256(Rekor public key)"))
elif set_b64:
out.append(Check(S, f"{name} · SET", "info", "enable network checks to verify with Rekor's public key"))
if online:
try:
live = json.loads(http_get(f"{REKOR_URL}/api/v1/log/entries/{uuid}"))
lv = live[uuid]
same = lv["body"] == body_b64 and lv["logIndex"] == e["logIndex"] and lv["integratedTime"] == e["integratedTime"]
out.append(Check(S, f"{name} · live lookup", "pass" if same else "fail", f"entry is in the public log · current tree size {lv['verification']['inclusionProof']['treeSize']:,}" if same else "live entry differs from the bundled copy"))
except urllib.error.HTTPError as ex:
out.append(Check(S, f"{name} · live lookup", "fail", f"Rekor returned HTTP {ex.code}"))
except Exception as ex:
out.append(Check(S, f"{name} · live lookup", "warn", f"lookup failed: {ex}"))
return out
# ---------------------------------------------------------------- certificate chains
def fetch_aia_issuers(bundle):
"""Download issuer certificates named in AIA extensions and add them to bundle['pool']."""
S = "Certificate chains"
out, seen_urls = [], set()
fetched = {}
for c in list(bundle["pool"]):
try:
aia = c.extensions.get_extension_for_class(x509.AuthorityInformationAccess).value
except x509.ExtensionNotFound:
continue
for a in aia:
if a.access_method != x509.oid.AuthorityInformationAccessOID.CA_ISSUERS:
continue
url = a.access_location.value
if url in seen_urls:
continue
seen_urls.add(url)
try:
for fc in load_certs_any(http_get(url)):
fetched.setdefault(fc.fingerprint(hashes.SHA256()), (fc, url))
except Exception as ex:
out.append(Check(S, "AIA fetch", "warn", f"<{url}> · {ex}"))
for fp, (fc, url) in fetched.items():
emb = bundle["embedded"].get(fp)
out.append(Check(S, "AIA issuer", "pass" if emb else "info", f"**{_cert_label(fc)}** fetched from <{url}>" + (f" · identical to the copy embedded in `{emb[1]}`" if emb else " · not in the bundle, added to the trust pool")))
if fp not in bundle["embedded"]:
bundle["pool"].append(fc)
bundle["fetched"] = fetched
return out
def check_chains(bundle):
S = "Certificate chains"
out = []
pool = list(bundle["pool"])
now = dt.datetime.now(dt.timezone.utc)
for name, leaf in bundle["certs"].items():
hops = build_chain(leaf, pool)
rooted = hops[-1][1] == "self-signed"
path = " → ".join(f"{_cert_label(c).split(',')[0]}" for c, _ in hops)
out.append(Check(S, f"{name}", "pass" if rooted else "warn", path + (f" · root `{_short_fp(hops[-1][0])}…`" if rooted else " · **issuer not found** — upload or fetch the CA certificate")))
for c, st in hops[1:]:
problems = []
try:
bc = c.extensions.get_extension_for_class(x509.BasicConstraints).value
if not bc.ca:
problems.append("basicConstraints CA=false")
except x509.ExtensionNotFound:
problems.append("no basicConstraints")
try:
ku = c.extensions.get_extension_for_class(x509.KeyUsage).value
if not ku.key_cert_sign:
problems.append("keyCertSign not set")
except x509.ExtensionNotFound:
pass
nb, na = certificate_validity(c)
if not (nb <= now <= na):
problems.append("outside validity period")
out.append(Check(S, f"{name} · {_cert_label(c).split(',')[0]}", "fail" if problems else "pass", ", ".join(problems) if problems else f"CA constraints OK · valid {nb:%Y-%m-%d} → {na:%Y-%m-%d}"))
try:
cdp = leaf.extensions.get_extension_for_class(x509.CRLDistributionPoints).value
urls = [str(n.value) for dp in cdp for n in (dp.full_name or [])]
if urls:
out.append(Check(S, f"{name} · revocation", "info", "CRL: " + ", ".join(f"<{u}>" for u in urls) + " (not checked)"))
except x509.ExtensionNotFound:
pass
if not bundle["certs"]:
out.append(Check(S, "No certificates", "skip", ""))
return out
# ---------------------------------------------------------------- revocation: CRL + OCSP
from cryptography.x509 import ocsp as x509_ocsp
def _uri_values(cert, ext_cls, method=None):
try:
ext = cert.extensions.get_extension_for_class(ext_cls).value
except x509.ExtensionNotFound:
return []
if ext_cls is x509.CRLDistributionPoints:
return [n.value for dp in ext for n in (dp.full_name or []) if isinstance(n, x509.UniformResourceIdentifier)]
return [a.access_location.value for a in ext if a.access_method == method and isinstance(a.access_location, x509.UniformResourceIdentifier)]
def _verify_raw_signature(pub, signature, data, hash_alg):
try:
if isinstance(pub, ec.EllipticCurvePublicKey):
pub.verify(signature, data, ec.ECDSA(hash_alg))
elif isinstance(pub, rsa.RSAPublicKey):
pub.verify(signature, data, padding.PKCS1v15(), hash_alg)
elif isinstance(pub, dsa.DSAPublicKey):
pub.verify(signature, data, hash_alg)
else:
pub.verify(signature, data)
return True
except (InvalidSignature, Exception):
return False
def check_crl(cert, issuer, cache=None, timeout=12):
"""Download the certificate's CRL(s), verify them against the issuer, and look the serial up.
Returns (status, detail) with status in pass|fail|warn|skip."""
urls = [u for u in _uri_values(cert, x509.CRLDistributionPoints) if u.lower().startswith("http")]
if not urls:
return "skip", "no HTTP CRL distribution point"
cache = cache if cache is not None else {}
now = dt.datetime.now(dt.timezone.utc)
problems = []
for url in urls:
try:
if url not in cache:
raw = http_get(url, timeout=timeout)
cache[url] = x509.load_pem_x509_crl(raw) if b"-----BEGIN" in raw else x509.load_der_x509_crl(raw)
crl = cache[url]
except Exception as e:
problems.append(f"<{url}> unreachable ({type(e).__name__})"); continue
if crl.issuer != issuer.subject or not crl.is_signature_valid(issuer.public_key()):
problems.append(f"<{url}> signature does not verify with the issuer"); continue
nxt = crl.next_update_utc
stale = nxt is not None and nxt < now
rev = crl.get_revoked_certificate_by_serial_number(cert.serial_number)
if rev is not None:
reason = ""
try:
reason = f" · reason {rev.extensions.get_extension_for_class(x509.CRLReason).value.reason.name}"
except x509.ExtensionNotFound:
pass
return "fail", f"**REVOKED** on {rev.revocation_date_utc:%Y-%m-%d %H:%M} UTC{reason} · CRL <{url}>"
detail = f"not listed in CRL <{url}> · {len(list(crl))} revoked · issued {crl.last_update_utc:%Y-%m-%d}" + (f" · next update {nxt:%Y-%m-%d}" if nxt else "")
if stale:
return "warn", detail + " · **CRL is stale** (next update has passed)"
return "pass", detail
return "warn", "; ".join(problems)
def check_ocsp(cert, issuer, timeout=12):
"""Query the certificate's OCSP responder(s) and validate the response signature.
Returns (status, detail) with status in pass|fail|warn|skip."""
urls = [u for u in _uri_values(cert, x509.AuthorityInformationAccess, x509.oid.AuthorityInformationAccessOID.OCSP) if u.lower().startswith("http")]
if not urls:
return "skip", "no OCSP responder in AIA"
req = x509_ocsp.OCSPRequestBuilder().add_certificate(cert, issuer, hashes.SHA1()).build().public_bytes(serialization.Encoding.DER)
problems = []
for url in urls:
try:
r = urllib.request.Request(url, data=req, headers={"Content-Type": "application/ocsp-request", "Accept": "application/ocsp-response", "User-Agent": "marimo-proof-verifier/1.0"})
with urllib.request.urlopen(r, timeout=timeout) as resp:
res = x509_ocsp.load_der_ocsp_response(resp.read())
except Exception as e:
problems.append(f"<{url}> {type(e).__name__}: {e}"); continue
if res.response_status != x509_ocsp.OCSPResponseStatus.SUCCESSFUL:
problems.append(f"<{url}> responder returned {res.response_status.name}"); continue
if res.serial_number != cert.serial_number:
problems.append(f"<{url}> response is for a different serial"); continue
# Who signed it? The issuer itself, or a delegated responder certified by the issuer with the OCSP-signing EKU.
signer, signer_note = None, ""
for c in list(res.certificates) + [issuer]:
if _verify_raw_signature(c.public_key(), res.signature, res.tbs_response_bytes, res.signature_hash_algorithm):
signer = c; break
if signer is None:
problems.append(f"<{url}> response signature does not verify"); continue
if signer is not issuer and signer.fingerprint(hashes.SHA256()) != issuer.fingerprint(hashes.SHA256()):
ok_eku = False
try:
ok_eku = x509.oid.ExtendedKeyUsageOID.OCSP_SIGNING in signer.extensions.get_extension_for_class(x509.ExtendedKeyUsage).value
except x509.ExtensionNotFound:
pass
if not issued_by(signer, issuer) or not ok_eku:
problems.append(f"<{url}> delegated responder **{signer.subject.rfc4514_string().split(',')[0]}** is not authorised by the issuer"); continue
signer_note = f" · responder {signer.subject.rfc4514_string().split(',')[0]}"
now = dt.datetime.now(dt.timezone.utc)
when = f"this update {res.this_update_utc:%Y-%m-%d %H:%M}" + (f" · next {res.next_update_utc:%Y-%m-%d %H:%M}" if res.next_update_utc else "")
stale = res.next_update_utc is not None and res.next_update_utc < now
if res.certificate_status == x509_ocsp.OCSPCertStatus.GOOD:
return ("warn" if stale else "pass"), f"GOOD per <{url}> · {when}{signer_note}" + (" · **response is stale**" if stale else "")
if res.certificate_status == x509_ocsp.OCSPCertStatus.REVOKED:
reason = f" · reason {res.revocation_reason.name}" if res.revocation_reason else ""
return "fail", f"**REVOKED** on {res.revocation_time_utc:%Y-%m-%d %H:%M} UTC{reason} per <{url}>{signer_note}"
return "warn", f"UNKNOWN to responder <{url}>{signer_note}"
return "warn", "; ".join(problems)
def check_revocation(bundle, online):
"""CRL + OCSP status of every certificate in the pool whose issuer we hold (roots excluded)."""
S = "Revocation (CRL / OCSP)"
if not online:
return [Check(S, "Revocation", "info", "enable network checks to query CRL distribution points and OCSP responders")]
out, seen, cache = [], set(), {}
pool = list(bundle["pool"])
named = {c.fingerprint(hashes.SHA256()): n for n, c in bundle["certs"].items()}
for c in pool:
fp = c.fingerprint(hashes.SHA256())
if fp in seen or (c.subject == c.issuer and issued_by(c, c)):
continue
seen.add(fp)
issuer = next((i for i in pool if i.subject == c.issuer and issued_by(c, i)), None)
label = named.get(fp) or _cert_label(c).split(",")[0]
if issuer is None:
out.append(Check(S, label, "skip", "issuer certificate not available, cannot build a revocation query")); continue
crl_s, crl_d = check_crl(c, issuer, cache)
ocsp_s, ocsp_d = check_ocsp(c, issuer)
out.append(Check(S, f"{label} · CRL", crl_s, crl_d))
out.append(Check(S, f"{label} · OCSP", ocsp_s, ocsp_d))
return out or [Check(S, "No certificates", "skip", "")]
# ---------------------------------------------------------------- orchestration + report
def run_bundle_checks(path, online=False):
bundle = load_bundle(path)
checks = []
aia_checks = []
if online:
try:
aia_checks = fetch_aia_issuers(bundle)
except Exception as e:
aia_checks = [Check("Certificate chains", "AIA fetch", "warn", f"{type(e).__name__}: {e}")]
steps = [(check_manifest, ()), (check_cross_refs, ()), (check_detached_signatures, ()), (check_cms, ()),
(check_timestamp, ()), (check_ots, (online,)), (check_rekor, (online,)), (check_chains, ()), (check_revocation, (online,))]
for fn, args in steps:
try:
res = fn(bundle, *args)
if fn is check_chains:
res = aia_checks + res
checks += res
except Exception as e:
checks.append(Check(fn.__name__, "crashed", "fail", f"{type(e).__name__}: {e}"))
counts = {k: sum(1 for c in checks if c.status == k) for k in STATUS_ICON}
serial = (bundle["manifest"] or {}).get("serial") or (bundle["provenance"] or {}).get("serial") or bundle["path"].name
return {"path": str(bundle["path"]), "serial": serial, "online": online, "ran_at": dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"),
"counts": counts, "checks": checks, "n_files": len(bundle["files"]), "certificates": certificate_records(bundle),
"provenance": {"title": ((bundle["provenance"] or {}).get("item") or {}).get("title"), "type": ((bundle["provenance"] or {}).get("item") or {}).get("type"),
"issuer": (bundle["provenance"] or {}).get("issuer"), "registered_at": (bundle["provenance"] or {}).get("registered_at"),
"verify_url": (bundle["provenance"] or {}).get("verify_url"), "serial": (bundle["provenance"] or {}).get("serial")},
"manifest": {"generated_at": (bundle["manifest"] or {}).get("generated_at"), "access_tier": (bundle["manifest"] or {}).get("access_tier")},
# portable identifier: folder name + SHA-256 over every "name:sha256" line, independent of where the bundle is stored
"bundle_id": {"name": bundle["path"].name, "n_files": len(bundle["files"]),
"content_sha256": hashlib.sha256("\n".join(f"{n}:{h}" for n, h in sorted(bundle["digests"].items())).encode()).hexdigest()}}
def report_markdown(results):
_l = [f"# Proof bundle verification — {results['serial']}", "",
f"- Folder: `{results['path']}`", f"- Files: {results['n_files']}", f"- Run at: {results['ran_at']}",
f"- Network checks: {'on' if results['online'] else 'off'}",
"- Result: " + " · ".join(f"{STATUS_ICON[k]} {v}" for k, v in results["counts"].items() if v), ""]
_sec = None
for c in results["checks"]:
if c.section != _sec:
_sec = c.section; _l += [f"## {_sec}", "", "| | Check | Detail |", "|---|---|---|"]
_l.append(f"| {STATUS_ICON[c.status]} | {c.name} | {c.detail.replace('|', '\\|')} |")
return "\n".join(_l) + "\n"
def report_json(results):
return json.dumps({**results, "checks": [asdict(c) for c in results["checks"]]}, indent=2, default=str)
return (
BUNDLES_DIR,
Check,
DATA_DIR,
LOCAL_PATHS_ALLOWED,
REPORTS_DIR,
STATUS_ICON,
check_crl,
check_ocsp,
extract_bundle_zip,
io,
is_allowed_bundle_dir,
re,
report_json,
report_markdown,
run_bundle_checks,
uploaded_bundles,
)
@app.cell
def _(DATA_DIR, REPORTS_DIR, dt, io, re):
from xml.sax.saxutils import escape as _xml_escape
from reportlab.lib import colors
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
from reportlab.lib.units import mm
from reportlab.platypus import KeepTogether, Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle
PDF_STATUS = {
"pass": ("PASS", colors.HexColor("#1a7f37")),
"fail": ("FAIL", colors.HexColor("#cf222e")),
"warn": ("WARN", colors.HexColor("#bf8700")),
"info": ("INFO", colors.HexColor("#57606a")),
"skip": ("SKIP", colors.HexColor("#8c959f")),
}
_PDF_CHARS = {"→": "->", "←": "<-", "✓": "ok", "✗": "x", "✅": "[pass]", "❌": "[fail]", "⚠️": "[warn]", "⚠": "[warn]", "ℹ️": "[info]", "ℹ": "[info]",
"⏭️": "[skip]", "⏭": "[skip]", "🔒": "[root]", "📜": "", "🔏": "", "·": "|", "—": "-", "–": "-", "‖": "||", "…": "...", " ": " "}
def pdf_text(s):
"""Sanitize a markdown-ish string for a reportlab Paragraph (Helvetica has no emoji/arrows)."""
s = str(s)
for k, v in _PDF_CHARS.items():
s = s.replace(k, v)
s = s.encode("latin-1", "replace").decode("latin-1")
s = _xml_escape(s)
s = re.sub(r"<(https?://[^&]+)>", r"\1", s) # <url>
s = re.sub(r"\*\*(.+?)\*\*", r"<b>\1</b>", s) # **bold**
s = re.sub(r"\*(.+?)\*", r"<i>\1</i>", s) # *italic*
s = re.sub(r"`([^`]*)`", r'<font face="Courier" size="7.5">\1</font>', s)
return s.replace("\\|", "|")
def _pdf_styles():
ss = getSampleStyleSheet()
return {
"title": ParagraphStyle("t", parent=ss["Title"], fontSize=18, spaceAfter=2 * mm, alignment=0),
"sub": ParagraphStyle("s", parent=ss["Normal"], fontSize=9, textColor=colors.HexColor("#57606a")),
"h": ParagraphStyle("h", parent=ss["Heading2"], fontSize=12, spaceBefore=5 * mm, spaceAfter=2 * mm),
"h3": ParagraphStyle("h3", parent=ss["Heading3"], fontSize=10, spaceBefore=3 * mm, spaceAfter=1 * mm),
"cell": ParagraphStyle("c", parent=ss["Normal"], fontSize=8, leading=10),
"status": ParagraphStyle("st", parent=ss["Normal"], fontSize=7.5, leading=9, textColor=colors.white, alignment=1, fontName="Helvetica-Bold"),
"verdict": ParagraphStyle("v", parent=ss["Normal"], fontSize=12, leading=15, textColor=colors.white, fontName="Helvetica-Bold"),
"small": ParagraphStyle("sm", parent=ss["Normal"], fontSize=7.5, leading=9, textColor=colors.HexColor("#57606a")),
}
def build_pdf_report(results, title="X.509 Proof Verification Report"):
"""Render a results dict (as produced by run_bundle_checks, or the manual verifier) to PDF bytes.
Expects keys: serial, path, ran_at, online, counts, checks (list[Check]); optional: certificates (list of dicts), subtitle."""
st = _pdf_styles()
buf = io.BytesIO()
doc = SimpleDocTemplate(buf, pagesize=A4, leftMargin=15 * mm, rightMargin=15 * mm, topMargin=15 * mm, bottomMargin=15 * mm,
title=f"{title} - {results['serial']}", author="Proof Bundle Verifier")
W = A4[0] - 30 * mm
c = results["counts"]
if c.get("fail"):
verdict, vcol = f"FAILED - {c['fail']} check(s) did not pass", PDF_STATUS["fail"][1]
elif c.get("warn"):
verdict, vcol = f"PASSED WITH {c['warn']} WARNING(S)", PDF_STATUS["warn"][1]
else:
verdict, vcol = "ALL CHECKS PASSED", PDF_STATUS["pass"][1]
story = [Paragraph(pdf_text(title), st["title"]),
Paragraph(pdf_text(results.get("subtitle") or f"Serial **{results['serial']}**"), st["sub"]),
Spacer(1, 3 * mm)]
vt = Table([[Paragraph(verdict, st["verdict"])]], colWidths=[W])
vt.setStyle(TableStyle([("BACKGROUND", (0, 0), (-1, -1), vcol), ("TOPPADDING", (0, 0), (-1, -1), 5), ("BOTTOMPADDING", (0, 0), (-1, -1), 5), ("LEFTPADDING", (0, 0), (-1, -1), 8)]))
story.append(vt)
meta = [["Source", results.get("path", "")], ["Generated", results.get("ran_at", "")],
["Network checks", "on" if results.get("online") else "off (Rekor signatures, Bitcoin anchoring and AIA fetches not confirmed)"],
["Summary", " ".join(f"{PDF_STATUS[k][0]} {v}" for k, v in c.items() if v)]]
mt = Table([[Paragraph(f"<b>{k}</b>", st["cell"]), Paragraph(pdf_text(v), st["cell"])] for k, v in meta], colWidths=[30 * mm, W - 30 * mm])
mt.setStyle(TableStyle([("VALIGN", (0, 0), (-1, -1), "TOP"), ("TOPPADDING", (0, 0), (-1, -1), 2), ("BOTTOMPADDING", (0, 0), (-1, -1), 2)]))
story += [Spacer(1, 2 * mm), mt]
sections = {}
for chk in results["checks"]:
sections.setdefault(chk.section, []).append(chk)
for name, items in sections.items():
rows = [[Paragraph("Status", st["cell"]), Paragraph("Check", st["cell"]), Paragraph("Detail", st["cell"])]]
style = [("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#eaeef2")), ("VALIGN", (0, 0), (-1, -1), "TOP"),
("GRID", (0, 0), (-1, -1), 0.25, colors.HexColor("#d0d7de")), ("TOPPADDING", (0, 0), (-1, -1), 2.5), ("BOTTOMPADDING", (0, 0), (-1, -1), 2.5)]
for i, chk in enumerate(items, 1):
label, col = PDF_STATUS.get(chk.status, PDF_STATUS["info"])
rows.append([Paragraph(label, st["status"]), Paragraph(pdf_text(chk.name), st["cell"]), Paragraph(pdf_text(chk.detail), st["cell"])])
style.append(("BACKGROUND", (0, i), (0, i), col))
t = Table(rows, colWidths=[16 * mm, 48 * mm, W - 64 * mm], repeatRows=1)
t.setStyle(TableStyle(style))
story += [Paragraph(pdf_text(name), st["h"]), t]
certs = results.get("certificates") or []
if certs:
story.append(Paragraph("Appendix - X.509 certificates", st["h"]))
for cinfo in certs:
fields = {k: v for k, v in cinfo.items() if k not in ("name", "source", "warnings")}
rows = [[Paragraph(f"<b>{pdf_text(k)}</b>", st["cell"]), Paragraph(pdf_text(v), st["cell"])] for k, v in fields.items()]
t = Table(rows, colWidths=[36 * mm, W - 36 * mm])
t.setStyle(TableStyle([("VALIGN", (0, 0), (-1, -1), "TOP"), ("GRID", (0, 0), (-1, -1), 0.25, colors.HexColor("#d0d7de")),
("BACKGROUND", (0, 0), (0, -1), colors.HexColor("#f6f8fa")), ("TOPPADDING", (0, 0), (-1, -1), 2), ("BOTTOMPADDING", (0, 0), (-1, -1), 2)]))
head = pdf_text(cinfo.get("name", "certificate")) + (f' <font color="#57606a">({pdf_text(cinfo["source"])})</font>' if cinfo.get("source") else "")
block = [Paragraph(head, st["h3"]), t]
if cinfo.get("warnings"):
block.append(Paragraph(pdf_text(" ".join(f"[warn] {w}" for w in cinfo["warnings"])), st["small"]))
story.append(KeepTogether(block))
def _footer(canvas, d):
canvas.saveState()
canvas.setFont("Helvetica", 7.5)
canvas.setFillColor(colors.HexColor("#57606a"))
canvas.drawString(15 * mm, 9 * mm, f"{title} - {results['serial']} - generated {results.get('ran_at', '')} by Proof Bundle Verifier")
canvas.drawRightString(A4[0] - 15 * mm, 9 * mm, f"page {d.page}")
canvas.restoreState()
doc.build(story, onFirstPage=_footer, onLaterPages=_footer)
return buf.getvalue()
# ---------------------------------------------------------------- customer edition (design: handoff index.html)
from reportlab.platypus import CondPageBreak, Flowable, PageBreak
from reportlab.pdfgen import canvas as _pdf_canvas
from reportlab.graphics.shapes import Circle, Drawing, PolyLine, String
C_INK, C_MUTED, C_LINE, C_GREEN, C_GREEN_BG, C_NOTE, C_AMBER, C_AMBER_BG, C_RED, C_RED_BG = (
colors.HexColor("#162c3a"), colors.HexColor("#52616c"), colors.HexColor("#d8e0e4"), colors.HexColor("#166044"),
colors.HexColor("#eef6f1"), colors.HexColor("#f2f5f8"), colors.HexColor("#775116"), colors.HexColor("#fbf4e6"),
colors.HexColor("#8a1c1c"), colors.HexColor("#fbeeee"))
_PDF_CHARS.update({"’": "'", "‘": "'", "“": '"', "”": '"', "·": "·"})
STATUS_WORD = {"pass": "PASS", "fail": "FAIL", "warn": "WARN", "info": "INFO", "skip": "SKIP"}
STATUS_COLOR = {"pass": C_GREEN, "fail": C_RED, "warn": C_AMBER, "info": C_MUTED, "skip": C_AMBER}
TAG_WORD = {"pass": "Passed", "fail": "Failed", "warn": "Needs attention", "skip": "Skipped", "info": "Recorded"}
GLOSSARY = [
("Proof bundle", "The collection of files used to check the item's digital documentation."),
("Manifest", "A file inventory that records expected file fingerprints and access levels."),
("Hash / fingerprint", "A value calculated from a file's contents. Matching values help detect changes to the file."),
("Digital signature", "A cryptographic check that connects signed data to a signing key and detects changes to that data."),
("Provenance", "The recorded information about an item's origin and history. A valid signature protects the record; it does not independently prove every claim in it."),
("Timestamp", "Evidence that a particular digital fingerprint existed by a recorded time."),
("Rekor", "A public transparency log that records signed data and provides evidence of inclusion."),
("Certificate chain", "The sequence of certificates linking a signing certificate to a root certificate that must be trusted."),
("CRL / OCSP", "Two ways to check whether a certificate has been revoked. CRL uses a published list; OCSP asks an online status service."),
("Out-of-band trust", "Trust established through a separate, reliable channel rather than inferred from the supplied bundle alone."),
]
class _Bookmark(Flowable):
"""Zero-height flowable that adds a PDF outline entry at its position."""
def __init__(self, title, key, level=0):
super().__init__(); self.title, self.key, self.level = title, key, level
def wrap(self, w, h):
return 0, 0
def draw(self):
self.canv.bookmarkPage(self.key)
self.canv.addOutlineEntry(self.title, self.key, level=self.level, closed=False)
class _SectionGap(Flowable):
"""Ruled gap between sections; collapses to a small spacer when it lands at the top of a page."""
def __init__(self, width, above=7 * mm, below=7 * mm):
super().__init__(); self.w, self.above, self.below, self._top = width, above, below, False
def wrap(self, aw, ah):
fr = getattr(self, "_frame", None)
self._top = fr is not None and ah >= fr._aH - 1
return self.w, (2 * mm if self._top else self.above + self.below + 1)
def draw(self):
if self._top:
return
self.canv.saveState(); self.canv.setStrokeColor(C_LINE); self.canv.setLineWidth(0.5)
self.canv.line(0, self.below, self.w, self.below); self.canv.restoreState()
def _cstyles():
base = dict(fontName="Helvetica", fontSize=10, leading=15, textColor=C_INK)
S = {
"brand": ParagraphStyle("brand", fontName="Times-Roman", fontSize=15, leading=18, textColor=C_INK),
"mastnote": ParagraphStyle("mastnote", fontName="Helvetica", fontSize=8.5, leading=11, textColor=C_MUTED, alignment=2),
"eyebrow": ParagraphStyle("eyebrow", fontName="Helvetica-Bold", fontSize=7.5, leading=10, textColor=C_MUTED, spaceAfter=5 * mm),
"display": ParagraphStyle("display", fontName="Times-Roman", fontSize=36, leading=40, textColor=C_INK, spaceAfter=3 * mm),
"subtitle": ParagraphStyle("subtitle", fontName="Helvetica", fontSize=11.5, leading=16.5, textColor=C_MUTED, spaceAfter=7 * mm),
"h2": ParagraphStyle("h2", fontName="Times-Roman", fontSize=21, leading=25, textColor=C_INK),
"h3": ParagraphStyle("h3", fontName="Helvetica-Bold", fontSize=11, leading=14, textColor=C_INK, spaceBefore=3 * mm, spaceAfter=1.5 * mm, keepWithNext=1),
"h4": ParagraphStyle("h4", fontName="Helvetica-Bold", fontSize=10.5, leading=13, textColor=C_INK, spaceBefore=5 * mm, spaceAfter=2 * mm),
"tcap": ParagraphStyle("tcap", fontName="Helvetica-Bold", fontSize=9, leading=12, textColor=C_INK, spaceBefore=3 * mm, spaceAfter=1.5 * mm),
"body": ParagraphStyle("body", **base, spaceAfter=3 * mm),
"lead": ParagraphStyle("lead", fontName="Helvetica", fontSize=11, leading=17, textColor=C_INK, spaceAfter=4 * mm),
"muted": ParagraphStyle("muted", fontName="Helvetica", fontSize=9.5, leading=14, textColor=C_MUTED, spaceAfter=2 * mm),
"caption": ParagraphStyle("caption", fontName="Helvetica", fontSize=8.5, leading=12, textColor=C_MUTED, spaceAfter=2 * mm),
"label": ParagraphStyle("label", fontName="Helvetica", fontSize=8.5, leading=11, textColor=C_MUTED),
"value": ParagraphStyle("value", fontName="Helvetica-Bold", fontSize=10, leading=13, textColor=C_INK),
"num": ParagraphStyle("num", fontName="Courier", fontSize=8.5, leading=12, textColor=C_MUTED),
"count": ParagraphStyle("count", fontName="Times-Roman", fontSize=26, leading=30, textColor=C_INK),
"cell": ParagraphStyle("cell", fontName="Helvetica", fontSize=8.5, leading=11.5, textColor=C_INK),
"cellmuted": ParagraphStyle("cellmuted", fontName="Helvetica", fontSize=8.5, leading=11.5, textColor=C_MUTED),
"th": ParagraphStyle("th", fontName="Helvetica-Bold", fontSize=8.5, leading=11, textColor=C_INK),
"gloss_t": ParagraphStyle("gloss_t", fontName="Helvetica-Bold", fontSize=9.5, leading=13, textColor=C_INK),
"gloss_d": ParagraphStyle("gloss_d", fontName="Helvetica", fontSize=9.5, leading=13, textColor=C_MUTED),
}
return S
def _status_span(status, extra=""):
col = STATUS_COLOR.get(status, C_MUTED).hexval().replace("0x", "#")
return f'<font name="Helvetica-Bold" size="7.5" color="{col}">{STATUS_WORD.get(status, status.upper())}</font>' + (f" {extra}" if extra else "")
def _worst(statuses):
"""Worst status for a group of checks, ignoring info/skip unless nothing else exists."""
for s in ("fail", "warn", "pass"):
if s in statuses:
return s
if "skip" in statuses:
return "skip"
return "info" if statuses else None
def _status_icon(status, size=11):
"""Small vector icon: circle with a tick (pass), an exclamation mark (warn/fail) or a dash (skip/info)."""
col = STATUS_COLOR.get(status, C_MUTED)
d = Drawing(size, size)
d.add(Circle(size / 2, size / 2, size / 2 - 0.6, strokeColor=col, strokeWidth=0.8, fillColor=None))
if status == "pass":
d.add(PolyLine([size * 0.28, size * 0.52, size * 0.44, size * 0.34, size * 0.74, size * 0.68], strokeColor=col, strokeWidth=1.3, strokeLineCap=1, strokeLineJoin=1))
elif status in ("warn", "fail"):
d.add(String(size / 2, size * 0.24, "!", fontName="Helvetica-Bold", fontSize=size * 0.72, fillColor=col, textAnchor="middle"))
else:
d.add(PolyLine([size * 0.3, size / 2, size * 0.7, size / 2], strokeColor=col, strokeWidth=1.2, strokeLineCap=1))
return d
def _tag(status):
col = STATUS_COLOR[status].hexval().replace("0x", "#")
t = Table([[_status_icon(status, 10), Paragraph(f'<font name="Helvetica-Bold" size="7.5" color="{col}">{TAG_WORD[status].upper()}</font>', ParagraphStyle("tag", leading=11))]],
colWidths=[13, None], hAlign="RIGHT")
t.setStyle(TableStyle([("VALIGN", (0, 0), (-1, -1), "MIDDLE"), ("LEFTPADDING", (0, 0), (-1, -1), 0), ("RIGHTPADDING", (0, 0), (-1, -1), 2), ("TOPPADDING", (0, 0), (-1, -1), 0), ("BOTTOMPADDING", (0, 0), (-1, -1), 0)]))
return t
def _fmt_ts(iso):
try:
d = dt.datetime.fromisoformat(iso.replace("Z", "+00:00")).astimezone(dt.timezone.utc)
return f"{d.day} {d:%B %Y}, {d:%H:%M:%S} UTC"
except Exception:
return iso
def _join(parts):
parts = [p for p in parts if p]
if len(parts) <= 1:
return "".join(parts)
if len(parts) == 2:
return f"{parts[0]} and {parts[1]}"
return ", ".join(parts[:-1]) + f", and {parts[-1]}"
# Evidence appendix grouping, following the headings used in the customer design.
EVIDENCE_GROUPS = [
("Report details & file inventory", ["Manifest"]),
("Item record & digital signatures", ["Provenance record", "Detached signatures", "CMS / PKCS#7 signatures"]),
("Independent timestamp evidence", ["RFC 3161 timestamp", "OpenTimestamps"]),
("Public transparency log · Rekor", ["Rekor transparency log"]),
("Certificate chains & revocation status", ["Certificate chains", "Revocation (CRL / OCSP)"]),
]
# Phrases for the opening sentence: (sections, phrase when every check passed, phrase when something failed or needs attention)
OPENING_TOPICS = [
(("Manifest", "Provenance record"), "matching files and item identifiers", "a file or identifier that does not match its record"),
(("Detached signatures", "CMS / PKCS#7 signatures"), "valid digital signatures", "a digital signature that did not verify"),
(("RFC 3161 timestamp", "OpenTimestamps", "Rekor transparency log"), "supporting timestamp and public-log evidence", "timestamp or public-log evidence that could not be confirmed"),
(("Certificate chains",), "complete certificate chains", "a certificate chain that is incomplete or has a problem"),
(("Revocation (CRL / OCSP)",), "no reported revocation", "a certificate whose revocation status needs attention"),
]
def _opening(sections, counts):
"""Compose the plain-language opening from the per-topic outcomes (no fixed wording for any bundle)."""
good, bad = [], []
for names, ok_phrase, bad_phrase in OPENING_TOPICS:
statuses = {c.status for n in names for c in sections.get(n, [])}
st = _worst(statuses)
if st == "pass":
good.append(ok_phrase)
elif st in ("fail", "warn"):
bad.append(bad_phrase)
# refine the timestamp phrase when only one kind of evidence was examined
ts = {c.status for n in ("RFC 3161 timestamp", "OpenTimestamps") for c in sections.get(n, [])}
rk = {c.status for c in sections.get("Rekor transparency log", [])}
if "supporting timestamp and public-log evidence" in good:
if not rk:
good[good.index("supporting timestamp and public-log evidence")] = "supporting timestamp evidence"
elif not ts:
good[good.index("supporting timestamp and public-log evidence")] = "supporting public-log evidence"
s = []
if good:
s.append(f"This verification found {_join(good)}.")
if bad:
s.append(("It also found " if good else "This verification found ") + _join(bad) + ".")
if counts.get("fail"):
s.append("In plain language: the checked documentation is not fully consistent with the signed digital record.")
elif counts.get("warn"):
s.append("In plain language: the checked documentation is consistent with the signed digital record, with points to review below.")
elif good:
s.append("In plain language: the checked documentation is consistent with the signed digital record.")
else:
s.append("No check produced a result, so nothing can be said about the record.")
return " ".join(s)
def _plain_items(sections, results):
"""Plain-language 'what this means' items, derived from the actual checks per topic."""
def group(*names):
return [c for n in names for c in sections.get(n, [])]
items = []
# 1. files + identifiers
g = group("Manifest", "Provenance record")
if g:
files = [c for c in sections.get("Manifest", []) if c.status in ("pass", "fail") and c.name not in ("Serial", "Unlisted files")]
ok = sum(1 for c in files if c.status == "pass")
unavailable = [c.name for c in sections.get("Manifest", []) if c.status == "info" and "unavailable" in c.detail]
serial_ok = any(c.name == "Serial agrees across files" and c.status == "pass" for c in g)
txt = f"{ok} of {len(files)} listed files match the SHA-256 fingerprints recorded in the manifest."
if serial_ok:
txt += " The item serial number agrees across the checked records."
if unavailable:
txt += f" The issuer declared {len(unavailable)} file{'s' if len(unavailable) != 1 else ''} unavailable; that is recorded as information, not a failure."
bad = [c.name for c in files if c.status == "fail"]
if bad:
txt += " Mismatched or missing: " + ", ".join(bad) + "."
items.append(("The files match their recorded fingerprints" if not bad else "Some files do not match their recorded fingerprints", txt, _worst({c.status for c in g})))
# 2. signatures
g = group("Detached signatures", "CMS / PKCS#7 signatures")
if g:
det = [c for c in sections.get("Detached signatures", []) if c.status in ("pass", "fail")]
cms_ = [c for c in sections.get("CMS / PKCS#7 signatures", []) if c.status in ("pass", "fail") and "chain" not in c.name]
parts = []
if det:
parts.append(f"{sum(1 for c in det if c.status == 'pass')} of {len(det)} detached signature{'s' if len(det) != 1 else ''} verified against a certificate in the bundle")
if cms_:
parts.append(f"{sum(1 for c in cms_ if c.status == 'pass')} of {len(cms_)} signed CMS record{'s' if len(cms_) != 1 else ''} passed the content and signature checks")
st = _worst({c.status for c in g})
items.append(("The digital record's signatures check out" if st == "pass" else "A signature did not verify", "; ".join(parts) + ".", st))
# 3. timestamps
g = group("RFC 3161 timestamp", "OpenTimestamps")
if g:
times = [re.search(r"\*\*(.+?)\*\*", c.detail).group(1) for c in sections.get("RFC 3161 timestamp", []) if c.name.endswith("· time") and re.search(r"\*\*(.+?)\*\*", c.detail)]
tsa = [re.search(r"TSA \*\*CN=([^,*]+)", c.detail).group(1) for c in sections.get("RFC 3161 timestamp", []) if "TSA signature" in c.name and re.search(r"TSA \*\*CN=([^,*]+)", c.detail)]
ots = [c for c in sections.get("OpenTimestamps", []) if "Bitcoin" in c.name]
txt = ""
if times:
txt += f"A timestamp{' from ' + tsa[0] if tsa else ''} records the certificate's fingerprint as existing at {times[0]}."
if ots:
o = ots[0]
txt += (" OpenTimestamps evidence was confirmed against the Bitcoin blockchain." if o.status == "pass" else
" OpenTimestamps evidence is recorded but was not confirmed against the Bitcoin blockchain in this run." if o.status == "info" else
" The OpenTimestamps evidence needs attention; see the technical evidence.")
items.append(("There is evidence of when the certificate existed", txt.strip() or "Timestamp evidence was examined; see the technical evidence.", _worst({c.status for c in g})))
# 4. rekor
g = sections.get("Rekor transparency log", [])
if g:
entries = sum(1 for c in g if c.name.endswith("· entry"))
live = [c for c in g if "live lookup" in c.name]
offline_bits = [c for c in g if c.status == "info" and "enable network" in c.detail]
txt = f"{entries} log entr{'ies' if entries != 1 else 'y'} for this item's signatures {'were' if entries != 1 else 'was'} checked against the bundled files, inclusion proofs and checkpoints."
if live and all(c.status == "pass" for c in live):
txt += " The entries were also found in the live public log."
if offline_bits:
txt += " Rekor's own signatures were not confirmed because network checks were off."
items.append(("The signatures appear in a public log", txt, _worst({c.status for c in g})))
# 5. chains
g = sections.get("Certificate chains", [])
if g:
txt = "The item's certificates chain to a root certificate with valid issuer signatures and CA constraints." if _worst({c.status for c in g}) == "pass" else "A certificate chain is incomplete or has a problem; see the technical evidence."
txt += " A valid chain does not by itself establish trust in its root; see the important details below."
items.append(("The certificates form complete chains", txt, _worst({c.status for c in g})))
# 6. revocation
g = sections.get("Revocation (CRL / OCSP)", [])
if g:
performed = [c for c in g if c.status in ("pass", "fail", "warn")]
skipped = [c for c in g if c.status == "skip"]
if not performed:
items.append(("Revocation status was not checked", "Revocation checks run only when network checks are enabled; they were not performed in this run.", "skip"))
else:
txt = f"{sum(1 for c in performed if c.status == 'pass')} of {len(performed)} revocation lookups found no revocation at the time of the report."
if skipped:
txt += f" {len(skipped)} lookup{'s' if len(skipped) != 1 else ''} could not run: " + "; ".join(f"{c.name} ({c.detail})" for c in skipped) + "."
st = _worst({c.status for c in performed})
items.append(("The checked certificates had no reported revocation" if st == "pass" else "A certificate's revocation status needs attention", txt, st))
return items
def build_customer_pdf_report(results, title="Verification Report"):
"""Customer-friendly edition of a results dict (same input as build_pdf_report). Returns PDF bytes."""
S = _cstyles()
buf = io.BytesIO()
MARGIN = 18 * mm
doc = SimpleDocTemplate(buf, pagesize=A4, leftMargin=MARGIN, rightMargin=MARGIN, topMargin=MARGIN, bottomMargin=20 * mm,
title=f"{title} - {results['serial']}", author=results.get("provenance", {}).get("issuer") or "Proof Bundle Verifier",
subject=f"Verification report for {results['serial']}")
W = A4[0] - 2 * MARGIN
c = results["counts"]
prov = results.get("provenance") or {}
bundle_id = results.get("bundle_id") or {}
item_title = prov.get("title") or results["serial"]
issuer = prov.get("issuer") or "—"
item_type = prov.get("type") or "—"
verify_url = prov.get("verify_url")
sections = {}
for chk in results["checks"]:
sections.setdefault(chk.section, []).append(chk)
performed = c.get("pass", 0) + c.get("fail", 0) + c.get("warn", 0)
if c.get("fail"):
verdict, vtext, vcol, vbg, vborder = "Some checks failed", f"{c['fail']} check{'s' if c['fail'] != 1 else ''} did not pass. Treat this record as unverified until the failures in the technical evidence are understood.", C_RED, C_RED_BG, colors.HexColor("#e6c3c3")
elif c.get("warn"):
verdict, vtext, vcol, vbg, vborder = "Passed, with points to review", f"No check failed, but {c['warn']} item{'s need' if c['warn'] != 1 else ' needs'} attention. " + ("They are" if c['warn'] != 1 else "It is") + " listed under important details.", C_AMBER, C_AMBER_BG, colors.HexColor("#e3d2b0")
elif performed:
verdict, vtext, vcol, vbg, vborder = "All performed checks passed", "The verification found no failed checks in the item's digital proof bundle.", C_GREEN, C_GREEN_BG, colors.HexColor("#c8dfd1")
else:
verdict, vtext, vcol, vbg, vborder = "No checks were performed", "Nothing could be verified from the supplied files.", C_MUTED, C_NOTE, C_LINE
def rule():
t = Table([[""]], colWidths=[W], rowHeights=[1]); t.setStyle(TableStyle([("LINEBELOW", (0, 0), (-1, -1), 0.5, C_LINE)])); return t
def section_head(num, text, key, first=False):
head = Paragraph(f'<font name="Courier" size="8.5" color="#52616c">{num}</font> {pdf_text(text)}', S["h2"])
pre = [] if first else [CondPageBreak(45 * mm), _SectionGap(W)]
return pre + [_Bookmark(f"{num} {text}", key), head, Spacer(1, 5 * mm)]
def zero_pad(t):
t.setStyle(TableStyle([("LEFTPADDING", (0, 0), (-1, -1), 0), ("RIGHTPADDING", (0, 0), (-1, -1), 0), ("TOPPADDING", (0, 0), (-1, -1), 0), ("BOTTOMPADDING", (0, 0), (-1, -1), 0)]))
return t
story = []
# ---- masthead: issuer brand + section note, ruled (mirrors the design's masthead)
mast = Table([[Paragraph(pdf_text(issuer if issuer != "—" else "Verification report"), S["brand"]), Paragraph("Item records / Verification", S["mastnote"])]], colWidths=[W * 0.6, W * 0.4])
mast.setStyle(TableStyle([("VALIGN", (0, 0), (-1, -1), "MIDDLE"), ("LEFTPADDING", (0, 0), (-1, -1), 0), ("RIGHTPADDING", (0, 0), (-1, -1), 0), ("TOPPADDING", (0, 0), (-1, -1), 0), ("BOTTOMPADDING", (0, 0), (-1, -1), 8), ("LINEBELOW", (0, 0), (-1, -1), 0.5, C_LINE)]))
story += [_Bookmark(item_title, "top"), mast, Spacer(1, 10 * mm)]
# ---- document head
story += [Paragraph("YOUR ITEM'S DIGITAL RECORD", S["eyebrow"]), Paragraph(pdf_text(item_title), S["display"]),
Paragraph("Verification, explained.<br/>A clear guide to the checks behind your item's digital documentation.", S["subtitle"])]
ident = [[Paragraph("Item serial number", S["label"]), Paragraph("Issued by", S["label"])],
[Paragraph(f'<font name="Courier">{pdf_text(results["serial"])}</font>', S["value"]), Paragraph(pdf_text(issuer), S["value"])],
[Paragraph("Item type", S["label"]), Paragraph("Report generated", S["label"])],
[Paragraph(pdf_text(item_type.replace("_", " ").capitalize()) + (f' <font color="#52616c" size="8.5">({pdf_text(item_type)})</font>' if "_" in item_type else ""), S["value"]), Paragraph(_fmt_ts(results["ran_at"]), S["value"])]]
it = Table(ident, colWidths=[W * 0.53, W * 0.47])
it.setStyle(TableStyle([("LEFTPADDING", (0, 0), (-1, -1), 0), ("TOPPADDING", (0, 0), (-1, -1), 1), ("BOTTOMPADDING", (0, 0), (-1, -1), 1), ("BOTTOMPADDING", (0, 1), (-1, 1), 10)]))
story += [it, Spacer(1, 8 * mm), rule(), Spacer(1, 5 * mm)]
toc = ["The result at a glance", "What this means for you", "A few important details", "Keep the record with your item", "The evidence, in detail", "Without the jargon"]
contents = " · ".join(f'<font name="Courier">0{i}</font> {t}' for i, t in enumerate(toc, 1))
story += [Paragraph("IN THIS REPORT", ParagraphStyle("eb2", parent=S["eyebrow"], spaceAfter=2 * mm)), Paragraph(contents, S["caption"])]
# ---- 01 result
story += section_head("01", "The result at a glance", "s1")
vcolhex = vcol.hexval().replace("0x", "#")
vstatus = "pass" if vcol is C_GREEN else ("fail" if vcol is C_RED else ("warn" if vcol is C_AMBER else "skip"))
banner = Table([[_status_icon(vstatus, 22), [Paragraph(f'<font name="Helvetica-Bold" size="15" color="{vcolhex}">{verdict}</font>', ParagraphStyle("bh", leading=19, spaceAfter=3)),
Paragraph(pdf_text(vtext), ParagraphStyle("bp", fontName="Helvetica", fontSize=10, leading=14.5, textColor=C_INK))]]], colWidths=[32, W - 32])
banner.setStyle(TableStyle([("BACKGROUND", (0, 0), (-1, -1), vbg), ("BOX", (0, 0), (-1, -1), 0.75, vborder), ("VALIGN", (0, 0), (-1, -1), "TOP"), ("TOPPADDING", (0, 0), (-1, -1), 12), ("BOTTOMPADDING", (0, 0), (-1, -1), 12), ("LEFTPADDING", (0, 0), (-1, -1), 14), ("RIGHTPADDING", (0, 0), (-1, -1), 14), ("LEFTPADDING", (1, 0), (1, 0), 12), ("TOPPADDING", (0, 0), (0, 0), 14)]))
story += [banner, Spacer(1, 5 * mm), Paragraph(pdf_text(_opening(sections, c)), S["lead"])]
cnt = [(c.get("pass", 0), "Passed checks", "Passed check", C_GREEN), (c.get("fail", 0), "Failed checks", "Failed check", C_RED), (c.get("warn", 0), "Need attention", "Needs attention", C_AMBER),
(c.get("info", 0), "Informational entries", "Informational entry", C_INK), (c.get("skip", 0), "Skipped checks", "Skipped check", C_INK)]
cnt = [(n, (s if n == 1 else l), col) for n, l, s, col in cnt if n] or [(0, "Checks performed", C_INK)]
cw = W / len(cnt)
ct = Table([[Paragraph(f'<font color="{col.hexval().replace("0x", "#")}">{n}</font>', S["count"]) for n, _, col in cnt], [Paragraph(l, S["label"]) for _, l, _ in cnt]], colWidths=[cw] * len(cnt))
cst = [("LINEABOVE", (0, 0), (-1, 0), 0.5, C_LINE), ("LINEBELOW", (0, -1), (-1, -1), 0.5, C_LINE), ("TOPPADDING", (0, 0), (-1, 0), 9), ("BOTTOMPADDING", (0, -1), (-1, -1), 9), ("LEFTPADDING", (0, 0), (0, -1), 0)]
for i in range(1, len(cnt)):
cst.append(("LINEBEFORE", (i, 0), (i, -1), 0.5, C_LINE))
ct.setStyle(TableStyle(cst))
net = "Network checks were enabled." if results.get("online") else "Network checks were disabled, so online evidence (public-log signatures, Bitcoin anchoring, issuer downloads, revocation) was not confirmed."
story += [ct, Spacer(1, 3 * mm), Paragraph(pdf_text(f"Results come from a verification run performed live when this report was generated. {net} Informational and skipped entries are explained below."), S["caption"])]
# ---- 02 what this means
items = _plain_items(sections, results)
if items:
story += section_head("02", "What this means for you", "s2")
for i, (head, text, st) in enumerate(items, 1):
row = Table([[Paragraph(f"0{i}", S["num"]), [Paragraph(pdf_text(head), ParagraphStyle("h3i", parent=S["h3"], spaceBefore=0)), Paragraph(pdf_text(text), S["muted"])], _tag(st)]], colWidths=[12 * mm, W - 12 * mm - 34 * mm, 34 * mm])
row.setStyle(TableStyle([("VALIGN", (0, 0), (-1, -1), "TOP"), ("LEFTPADDING", (0, 0), (-1, -1), 0), ("TOPPADDING", (0, 0), (-1, -1), 7), ("BOTTOMPADDING", (0, 0), (-1, -1), 7), ("TOPPADDING", (0, 0), (0, 0), 8), ("TOPPADDING", (2, 0), (2, 0), 9), ("LINEBELOW", (0, 0), (-1, -1), 0.5, C_LINE)]))
if i == len(items):
row.setStyle(TableStyle([("LINEBELOW", (0, 0), (-1, -1), 0, colors.white)]))
story.append(KeepTogether(row))
# ---- 03 important details
story += section_head("03", "A few important details", "s3")
details = [("These are digital checks, not a physical inspection", "They assess the supplied files, signatures and related records. On their own they do not prove the physical item's authenticity, condition, ownership or value.")]
unavailable = [ch.name for ch in sections.get("Manifest", []) if ch.status == "info" and "unavailable" in ch.detail]
if unavailable:
details.append((f"{len(unavailable)} file{'s were' if len(unavailable) != 1 else ' was'} not supplied", f"The issuer declared {', '.join(unavailable)} unavailable. This is recorded as information, not as a failed integrity check."))
skipped = [ch for ch in results["checks"] if ch.status == "skip" and ch.detail]
if skipped:
details.append((f"{len(skipped)} check{'s' if len(skipped) != 1 else ''} could not run", " ".join(f"{ch.name}: {ch.detail}." for ch in skipped)))
warns = [ch for ch in results["checks"] if ch.status == "warn"]
if warns:
details.append((f"{len(warns)} item{'s need' if len(warns) != 1 else ' needs'} attention", " ".join(f"{ch.name}: {ch.detail}." for ch in warns)))
fails = [ch for ch in results["checks"] if ch.status == "fail"]
if fails:
details.append((f"{len(fails)} check{'s' if len(fails) != 1 else ''} failed", " ".join(f"{ch.name}: {ch.detail}." for ch in fails)))
roots = [ci for ci in results.get("certificates", []) if any("self-signed" in w for w in ci.get("warnings", []))]
oob = [ch for ch in results["checks"] if "trust this root out of band" in ch.detail]
if roots or oob:
rtxt = ""
if roots:
rtxt += "The root certificate" + ("s " if len(roots) != 1 else " ") + ", ".join(ci["name"].split(",")[0].replace("CN=", "") for ci in roots) + (" are" if len(roots) != 1 else " is") + " self-signed, so trust in " + ("them" if len(roots) != 1 else "that root") + " must be established separately."
if oob:
rtxt += " The timestamp chain also ends at a root that should be trusted out of band."
rtxt += " A valid chain does not, by itself, establish trust in its root."
details.append(("Certificate trust still needs an independent basis", rtxt.strip()))
for j, (head, text) in enumerate(details):
story.append(KeepTogether([Paragraph(pdf_text(head), ParagraphStyle("h3d", parent=S["h3"], spaceBefore=0 if j == 0 else 4 * mm)), Paragraph(pdf_text(text), S["muted"])]))
note = Table([[[Paragraph("Think of this as a dated snapshot", ParagraphStyle("h3n", parent=S["h3"], spaceBefore=0)), Paragraph(pdf_text(f"These results describe the verification run of {_fmt_ts(results['ran_at'])}. Certificate and online status information can change; this document does not update itself."), S["muted"])]]], colWidths=[W])
note.setStyle(TableStyle([("BACKGROUND", (0, 0), (-1, -1), C_NOTE), ("TOPPADDING", (0, 0), (-1, -1), 12), ("BOTTOMPADDING", (0, 0), (-1, -1), 8), ("LEFTPADDING", (0, 0), (-1, -1), 14), ("RIGHTPADDING", (0, 0), (-1, -1), 14)]))
story += [Spacer(1, 5 * mm), note]
# ---- 04 next steps
_s4 = section_head("04", "Keep the record with your item", "s4")
steps = [f"<b>Match the serial number.</b> Look for <font name=\"Courier\">{pdf_text(results['serial'])}</font> in the item's documentation.",
"<b>Keep the original evidence.</b> Retain the certificate, proof bundle and this verification report together."]
if verify_url:
steps.append("<b>Visit the verification page when needed.</b> Use the issuer's link below to reach the item's verification page.")
_s4_body = []
for i, s in enumerate(steps, 1):
_s4_body.append(Table([[Paragraph(str(i) + ".", S["body"]), Paragraph(s, S["body"])]], colWidths=[7 * mm, W - 7 * mm], style=TableStyle([("VALIGN", (0, 0), (-1, -1), "TOP"), ("LEFTPADDING", (0, 0), (-1, -1), 0), ("TOPPADDING", (0, 0), (-1, -1), 0), ("BOTTOMPADDING", (0, 0), (-1, -1), 2)])))
if verify_url:
u = pdf_text(verify_url)
link = Table([[Paragraph(f'<a href="{u}"><font name="Helvetica-Bold" color="white">Open item verification</font></a>', ParagraphStyle("lk", fontSize=10, leading=13))]], colWidths=[58 * mm], hAlign="LEFT")
link.setStyle(TableStyle([("BACKGROUND", (0, 0), (-1, -1), C_INK), ("TOPPADDING", (0, 0), (-1, -1), 8), ("BOTTOMPADDING", (0, 0), (-1, -1), 8), ("LEFTPADDING", (0, 0), (-1, -1), 16), ("RIGHTPADDING", (0, 0), (-1, -1), 16)]))
_s4_body += [Spacer(1, 3 * mm), link, Spacer(1, 2 * mm), Paragraph(f'<a href="{u}" color="#162c3a"><font name="Courier" size="8.5">{u}</font></a>', S["caption"])]
story += _s4[:3] + [KeepTogether(_s4[3:] + _s4_body)]
# ---- 05 technical evidence: always starts on a new page (as the design's print stylesheet does)
story += [PageBreak()] + section_head("05", "The evidence, in detail", "s5", first=True)
story.append(Paragraph("For anyone who wants the technical record. Every recorded finding from this run is included below; abbreviated hashes are shown exactly as the verifier recorded them.", S["muted"]))
bundle_label = bundle_id.get("name") or results["serial"]
if bundle_id.get("n_files"):
bundle_label += f" · {bundle_id['n_files']} files"
meta = [("Bundle", bundle_label),
("Bundle fingerprint", f"<font name=\"Courier\" size=\"8\">{bundle_id['content_sha256']}</font>" if bundle_id.get("content_sha256") else "—"),
("Generated", results.get("ran_at", "")), ("Verifier", "Proof Bundle Verifier (X.509 proof verification)"),
("Network checks", "on" if results.get("online") else "off"), ("Summary", " / ".join(f"{STATUS_WORD[k]} {v}" for k, v in c.items() if v))]
mt = Table([[Paragraph(k, S["cellmuted"]), Paragraph(v if k == "Bundle fingerprint" else pdf_text(v), S["cell"])] for k, v in meta], colWidths=[34 * mm, W - 34 * mm])
mt.setStyle(TableStyle([("VALIGN", (0, 0), (-1, -1), "TOP"), ("LEFTPADDING", (0, 0), (-1, -1), 0), ("TOPPADDING", (0, 0), (-1, -1), 1.5), ("BOTTOMPADDING", (0, 0), (-1, -1), 1.5)]))
story += [Spacer(1, 2 * mm), mt, Spacer(1, 1 * mm), Paragraph("The bundle fingerprint is the SHA-256 of every file name and file hash in the bundle, so the same bundle produces the same value wherever it is stored.", S["caption"])]
def check_table(items_):
rows = [[Paragraph("Status", S["th"]), Paragraph("Check", S["th"]), Paragraph("Recorded finding", S["th"])]]
for ch in items_:
rows.append([Paragraph(_status_span(ch.status), S["cell"]), Paragraph(pdf_text(ch.name), S["cell"]), Paragraph(pdf_text(ch.detail), S["cell"])])
t = Table(rows, colWidths=[14 * mm, 46 * mm, W - 60 * mm], repeatRows=1)
t.setStyle(TableStyle([("BACKGROUND", (0, 0), (-1, 0), C_NOTE), ("VALIGN", (0, 0), (-1, -1), "TOP"), ("LINEBELOW", (0, 0), (-1, -1), 0.4, C_LINE),
("LEFTPADDING", (0, 0), (-1, -1), 4), ("RIGHTPADDING", (0, 0), (-1, -1), 4), ("TOPPADDING", (0, 0), (-1, -1), 3), ("BOTTOMPADDING", (0, 0), (-1, -1), 3)]))
return t
placed = set()
groups = [(g, [n for n in names if n in sections]) for g, names in EVIDENCE_GROUPS]
groups = [(g, names) for g, names in groups if names]
rest = [n for n in sections if n not in {n for _, names in groups for n in names}]
if rest:
groups.append(("Other findings", rest))
for gi, (gtitle, names) in enumerate(groups):
story += [CondPageBreak(45 * mm), Spacer(1, 3 * mm), rule(), _Bookmark(gtitle, f"grp-{gi}", level=1), Paragraph(pdf_text(gtitle), S["h4"])]
for n in names:
story += [CondPageBreak(30 * mm), Paragraph(pdf_text(n), S["tcap"]), check_table(sections[n])]
certs = results.get("certificates") or []
if certs:
story += [CondPageBreak(60 * mm), Spacer(1, 3 * mm), rule(), _Bookmark("X.509 certificate reference & fingerprints", "certs", level=1), Paragraph("X.509 certificate reference & fingerprints", S["h4"]),
Paragraph("Dates are UTC. Fingerprints are shown in full so they can be compared with other copies of each certificate.", S["cellmuted"])]
for ci in certs:
fields = {k: v for k, v in ci.items() if k not in ("name", "source", "warnings")}
rows = [[Paragraph(pdf_text(k), S["cellmuted"]), Paragraph(pdf_text(v).replace("<font face=", "<font name="), S["cell"])] for k, v in fields.items()]
t = Table(rows, colWidths=[34 * mm, W - 34 * mm])
t.setStyle(TableStyle([("VALIGN", (0, 0), (-1, -1), "TOP"), ("LINEBELOW", (0, 0), (-1, -1), 0.3, C_LINE), ("LEFTPADDING", (0, 0), (-1, -1), 0), ("TOPPADDING", (0, 0), (-1, -1), 2), ("BOTTOMPADDING", (0, 0), (-1, -1), 2)]))
head = [CondPageBreak(45 * mm), Paragraph(pdf_text(ci.get("name", "certificate")) + (f' <font color="#52616c" size="8.5">· {pdf_text(ci["source"])}</font>' if ci.get("source") else ""), S["tcap"])]
story += head + [t]
if ci.get("warnings"):
story.append(Paragraph(pdf_text("Recorded warnings: " + " ".join(ci["warnings"])), S["cellmuted"]))
story += [Spacer(1, 5 * mm), rule(), Spacer(1, 3 * mm),
Paragraph(pdf_text(f"**Source.** This customer edition and the technical report were produced by the same verification run on {_fmt_ts(results['ran_at'])}. The findings above are the verifier's own records for that run; nothing has been reworded in this appendix."), S["caption"])]
# ---- 06 glossary
story += section_head("06", "Without the jargon", "s6")
gt = Table([[Paragraph(pdf_text(k), S["gloss_t"]), Paragraph(pdf_text(v), S["gloss_d"])] for k, v in GLOSSARY], colWidths=[38 * mm, W - 38 * mm])
gt.setStyle(TableStyle([("VALIGN", (0, 0), (-1, -1), "TOP"), ("LEFTPADDING", (0, 0), (-1, -1), 0), ("TOPPADDING", (0, 0), (-1, -1), 3), ("BOTTOMPADDING", (0, 0), (-1, -1), 3)]))
story.append(gt)
footer_left = (f"{item_title} · {results['serial']}" if item_title != results["serial"] else results["serial"]).encode("latin-1", "replace").decode("latin-1")
class _NumberedCanvas(_pdf_canvas.Canvas):
"""Two-pass canvas so the footer can say 'Page N of M'."""
def __init__(self, *a, **k):
super().__init__(*a, **k); self._saved = []
def showPage(self):
self._saved.append(dict(self.__dict__)); self._startPage()
def save(self):
total = len(self._saved)
self.showOutline()
for st in self._saved:
self.__dict__.update(st)
self.saveState()
self.setStrokeColor(C_LINE); self.setLineWidth(0.5); self.line(MARGIN, 14 * mm, A4[0] - MARGIN, 14 * mm)
self.setFont("Helvetica", 7.5); self.setFillColor(C_MUTED)
self.drawString(MARGIN, 9.5 * mm, footer_left)
self.drawRightString(A4[0] - MARGIN, 9.5 * mm, f"Customer edition · Page {self._pageNumber} of {total}")
self.restoreState()
super().showPage()
super().save()
doc.build(story, canvasmaker=_NumberedCanvas)
return buf.getvalue()
def save_report(data, filename):
"""Persist report bytes under REPORTS_DIR (e.g. /data/reports on the Space). Returns the path or None."""
if not DATA_DIR.is_dir():
return None
try:
REPORTS_DIR.mkdir(parents=True, exist_ok=True)
p = REPORTS_DIR / filename
p.write_bytes(data)
return p
except OSError:
return None
return build_customer_pdf_report, build_pdf_report, save_report
@app.cell
def _(mo):
import base64
import binascii
import datetime as dt
import hashlib
import pathlib
from cryptography.hazmat.primitives import serialization
from cryptography import x509
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import (
rsa,
ec,
dsa,
ed25519,
ed448,
padding,
utils as asym_utils,
)
def upload_bytes(value):
"""Return (name, bytes) for a mo.ui.file value, or (None, None) if nothing was uploaded."""
if not value:
return None, None
_f = value[0] if isinstance(value, (list, tuple)) else value
return getattr(_f, "name", None), getattr(_f, "contents", _f)
def load_certificate(data):
"""Load a PEM or DER X.509 certificate from bytes/str."""
if isinstance(data, str):
data = data.encode()
if b"-----BEGIN CERTIFICATE-----" in data:
return x509.load_pem_x509_certificate(data.strip())
return x509.load_der_x509_certificate(data)
def decode_sha256(text):
"""Parse a hex SHA-256 digest (colons/spaces allowed). Returns 32 bytes."""
_h = (text or "").strip().replace(":", "").replace(" ", "").lower()
if len(_h) != 64:
raise ValueError(f"A SHA-256 hash is 64 hex characters; got {len(_h)}.")
return bytes.fromhex(_h)
def decode_signature(data):
"""Accept a signature as raw bytes, hex (optionally colon-separated) or base64.
Returns (signature_bytes, encoding_label)."""
if isinstance(data, bytes):
try:
_text = data.decode("ascii")
except UnicodeDecodeError:
return data, "raw bytes"
if not _text.isprintable() and not _text.strip().isprintable():
return data, "raw bytes"
data = _text
_s = "".join(data.split())
_hex = _s.replace(":", "")
if _hex and len(_hex) % 2 == 0 and all(c in "0123456789abcdefABCDEF" for c in _hex):
return bytes.fromhex(_hex), "hex"
try:
return base64.b64decode(_s, validate=True), "base64"
except (binascii.Error, ValueError):
raise ValueError("Signature must be a raw .sig file, hex (optionally colon-separated), or base64.")
def verification_attempts(pub, digest):
"""Candidate (label, verify_fn) pairs for a public key and a 32-byte SHA-256 digest."""
_pre = asym_utils.Prehashed(hashes.SHA256())
_hex_msg = digest.hex().encode()
attempts = []
if isinstance(pub, rsa.RSAPublicKey):
attempts.append(("RSA PKCS#1 v1.5, prehashed SHA-256", lambda s: pub.verify(s, digest, padding.PKCS1v15(), _pre)))
attempts.append(("RSA PSS, prehashed SHA-256", lambda s: pub.verify(s, digest, padding.PSS(padding.MGF1(hashes.SHA256()), padding.PSS.AUTO), _pre)))
for _h in (hashes.SHA256(), hashes.SHA1()):
for _lbl, _msg in (("raw digest bytes", digest), ("hex digest string", _hex_msg)):
attempts.append((f"RSA PKCS#1 v1.5 / {_h.name} over {_lbl}", lambda s, h=_h, m=_msg: pub.verify(s, m, padding.PKCS1v15(), h)))
elif isinstance(pub, ec.EllipticCurvePublicKey):
attempts.append(("ECDSA, prehashed SHA-256", lambda s: pub.verify(s, digest, ec.ECDSA(_pre))))
for _h in (hashes.SHA256(), hashes.SHA1()):
for _lbl, _msg in (("raw digest bytes", digest), ("hex digest string", _hex_msg)):
attempts.append((f"ECDSA / {_h.name} over {_lbl}", lambda s, h=_h, m=_msg: pub.verify(s, m, ec.ECDSA(h))))
elif isinstance(pub, dsa.DSAPublicKey):
attempts.append(("DSA, prehashed SHA-256", lambda s: pub.verify(s, digest, _pre)))
attempts.append(("DSA / sha256 over raw digest bytes", lambda s: pub.verify(s, digest, hashes.SHA256())))
elif isinstance(pub, (ed25519.Ed25519PublicKey, ed448.Ed448PublicKey)):
_alg = "Ed25519" if isinstance(pub, ed25519.Ed25519PublicKey) else "Ed448"
attempts.append((f"{_alg} over raw digest bytes", lambda s: pub.verify(s, digest)))
attempts.append((f"{_alg} over hex digest string", lambda s: pub.verify(s, _hex_msg)))
return attempts
def certificate_validity(cert):
try:
return cert.not_valid_before_utc, cert.not_valid_after_utc
except AttributeError:
return (
cert.not_valid_before.replace(tzinfo=dt.timezone.utc),
cert.not_valid_after.replace(tzinfo=dt.timezone.utc),
)
def _ext_or_none(cert, cls):
try:
return cert.extensions.get_extension_for_class(cls).value
except x509.ExtensionNotFound:
return None
def certificate_summary(cert):
"""Return (fields: dict, warnings: list[str]) describing a certificate for display."""
pub = cert.public_key()
nb, na = certificate_validity(cert)
now = dt.datetime.now(dt.timezone.utc)
warnings = []
if isinstance(pub, ec.EllipticCurvePublicKey):
key_desc = f"ECDSA {pub.curve.name} ({pub.key_size}-bit)"
elif isinstance(pub, rsa.RSAPublicKey):
key_desc = f"RSA ({pub.key_size}-bit)"
elif isinstance(pub, dsa.DSAPublicKey):
key_desc = f"DSA ({pub.key_size}-bit)"
else:
key_desc = type(pub).__name__.replace("_", "").replace("PublicKey", "")
san = _ext_or_none(cert, x509.SubjectAlternativeName)
san_text = ", ".join(str(v) for v in san.get_values_for_type(x509.UniformResourceIdentifier) + san.get_values_for_type(x509.DNSName)) if san else "—"
ku = _ext_or_none(cert, x509.KeyUsage)
if ku is not None:
ku_flags = [n for n in ("digital_signature", "content_commitment", "key_encipherment", "key_cert_sign", "crl_sign") if getattr(ku, n, False)]
ku_text = ", ".join(f.replace("_", " ") for f in ku_flags) or "none"
if not ku.digital_signature:
warnings.append("Key usage does **not** include *digital signature*; this certificate is not meant for signing data.")
else:
ku_text = "—"
eku = _ext_or_none(cert, x509.ExtendedKeyUsage)
eku_text = ", ".join(getattr(o, "_name", None) or o.dotted_string for o in eku) if eku else "—"
if now < nb:
warnings.append(f"Certificate is **not yet valid** (starts {nb:%Y-%m-%d %H:%M} UTC).")
elif now > na:
warnings.append(f"Certificate **expired** on {na:%Y-%m-%d %H:%M} UTC.")
if cert.subject == cert.issuer:
warnings.append("Certificate is **self-signed**; trust must be established out of band.")
fields = {
"Subject": cert.subject.rfc4514_string(),
"Issuer": cert.issuer.rfc4514_string(),
"Serial": format(cert.serial_number, "x"),
"Valid": f"{nb:%Y-%m-%d} → {na:%Y-%m-%d} UTC" + (" ✅" if nb <= now <= na else " ⚠️"),
"Public key": key_desc,
"Signature algorithm": cert.signature_algorithm_oid._name,
"Subject alt names": san_text,
"Key usage": ku_text,
"Extended key usage": eku_text,
"SHA-256 fingerprint": cert.fingerprint(hashes.SHA256()).hex(":").upper(),
"SHA-1 fingerprint": cert.fingerprint(hashes.SHA1()).hex(":").upper(),
}
return fields, warnings
def fields_table(fields):
"""Render a dict as a two-column markdown table."""
_rows = "\n".join(f"| **{k}** | `{v}` |" for k, v in fields.items())
return mo.md(f"| Field | Value |\n|---|---|\n{_rows}")
def validate_sha256_signature(cert, digest, sig):
"""Try every plausible scheme for this key type. Returns (matched_label_or_None, tried_labels)."""
pub = cert.public_key()
tried = []
for label, fn in verification_attempts(pub, digest):
try:
fn(sig)
return label, tried
except InvalidSignature:
tried.append(label)
except Exception as e:
tried.append(f"{label} (error: {e})")
return None, tried
def issued_by(child, issuer):
"""True if `issuer`'s public key verifies `child`'s certificate signature."""
pub = issuer.public_key()
try:
if isinstance(pub, ec.EllipticCurvePublicKey):
pub.verify(child.signature, child.tbs_certificate_bytes, ec.ECDSA(child.signature_hash_algorithm))
elif isinstance(pub, rsa.RSAPublicKey):
pub.verify(child.signature, child.tbs_certificate_bytes, padding.PKCS1v15(), child.signature_hash_algorithm)
elif isinstance(pub, dsa.DSAPublicKey):
pub.verify(child.signature, child.tbs_certificate_bytes, child.signature_hash_algorithm)
else:
pub.verify(child.signature, child.tbs_certificate_bytes)
return True
except (InvalidSignature, Exception):
return False
def load_certs_any(data):
"""Load every certificate from PEM (one or many), DER, or a CMS/PKCS#7 blob. Returns a list."""
if isinstance(data, str):
data = data.encode()
if b"-----BEGIN CERTIFICATE-----" in data:
return list(x509.load_pem_x509_certificates(data))
try:
return [x509.load_der_x509_certificate(data)]
except Exception:
pass
from asn1crypto import cms as _cms
_ci = _cms.ContentInfo.load(data)
if _ci["content_type"].native == "signed_data" and _ci["content"]["certificates"]:
return [x509.load_der_x509_certificate(c.chosen.dump()) for c in _ci["content"]["certificates"]]
raise ValueError("No certificates found (expected PEM, DER or PKCS#7).")
def build_chain(leaf, pool):
"""Walk issuer links from `leaf` through `pool` (list of certs). Returns list of
(cert, status) hops: 'ok' when the next issuer's signature verifies, 'self-signed' at a root,
'missing' when no issuer in the pool signs it."""
hops, cur, seen = [], leaf, set()
while True:
seen.add(cur.fingerprint(hashes.SHA256()))
if cur.subject == cur.issuer and issued_by(cur, cur):
hops.append((cur, "self-signed"))
return hops
nxt = next((c for c in pool if c.subject == cur.issuer and c.fingerprint(hashes.SHA256()) not in seen and issued_by(cur, c)), None)
if nxt is None:
hops.append((cur, "missing"))
return hops
hops.append((cur, "ok"))
cur = nxt
def copy_button(text, label="Copy"):
"""A small button that copies `text` to the clipboard."""
import html as _html
import json as _json
_js = _html.escape(_json.dumps(text), quote=True)
return mo.Html(
f'<button onclick="navigator.clipboard.writeText({_js}).then(()=>{{this.textContent=\'✓ Copied\';setTimeout(()=>this.textContent=\'📋 {label}\',1200)}})" '
'style="font-size:12px;padding:2px 8px;border:1px solid var(--gray-6,#ccc);border-radius:6px;background:var(--gray-2,#f4f4f5);cursor:pointer">'
f"📋 {label}</button>"
)
return (
InvalidSignature,
base64,
build_chain,
certificate_summary,
certificate_validity,
copy_button,
decode_sha256,
decode_signature,
dsa,
dt,
ec,
fields_table,
hashes,
hashlib,
issued_by,
load_certificate,
load_certs_any,
padding,
pathlib,
rsa,
serialization,
upload_bytes,
validate_sha256_signature,
x509,
)
if __name__ == "__main__":
app.run()
import marimo
__generated_with = "0.25.2"
app = marimo.App(
width="medium",
)
def _():
import marimo as mo
return (mo,)
def _(mo):
mo.md("""
# 🔏 Proof Bundle Verifier
Everything runs locally in this notebook. Nothing is uploaded anywhere; network calls happen only when you switch on **network checks**.
**Two ways to use it**
- **Verify a whole bundle** — upload it as a `.zip` and every artifact is checked and cross-referenced: manifest hashes, detached signatures, CMS/PKCS#7, RFC 3161 timestamp, OpenTimestamps, Rekor transparency log, certificate chains, and CRL / OCSP revocation.
- **Manual check** (further down) — drop one certificate, one file and one signature to verify a single detached signature.
""")
return
def _(
BUNDLES_DIR,
DATA_DIR,
LOCAL_PATHS_ALLOWED,
REPORTS_DIR,
mo,
pathlib,
uploaded_bundles,
):
_uploaded = uploaded_bundles()
bundle_zip = mo.ui.file(filetypes=[".zip"], multiple=False, kind="area", label="Upload a proof bundle as .zip")
bundle_choice = mo.ui.dropdown(options={k: str(v) for k, v in _uploaded.items()}, label="…or pick a bundle uploaded earlier") if _uploaded else None
online_switch = mo.ui.switch(label="Network checks — Rekor live lookup & key, Bitcoin block headers, CA issuer download, CRL / OCSP revocation")
if LOCAL_PATHS_ALLOWED:
bundle_picker = mo.ui.file_browser(
initial_path=DATA_DIR if DATA_DIR.is_dir() else pathlib.Path.home(),
selection_mode="directory",
multiple=False,
restrict_navigation=False,
label="Pick a proof-bundle folder on this machine",
)
bundle_path_text = mo.ui.text(placeholder="…or paste a folder path, e.g. ~/Downloads/SW-2026-428978-proof-bundle", full_width=True)
_local_controls = [mo.md("*Local folder access is on (editor mode). It is hidden in the deployed app unless `VERIFIER_ALLOW_LOCAL_PATHS=1`.*"), bundle_picker, bundle_path_text]
else:
bundle_picker = None
bundle_path_text = None
_local_controls = []
_storage_note = (
mo.md(f"Uploaded bundles are kept in `{BUNDLES_DIR}` and PDF reports in `{REPORTS_DIR}`.")
if DATA_DIR.is_dir()
else mo.md(f"*No persistent storage at `{DATA_DIR}`; uploads are extracted to a temporary folder and reports are download-only.*")
)
mo.vstack(
[mo.md("## 1 · Verify a whole bundle"), bundle_zip]
+ ([bundle_choice] if bundle_choice is not None else [])
+ _local_controls
+ [online_switch, _storage_note]
)
return (
bundle_choice,
bundle_path_text,
bundle_picker,
bundle_zip,
online_switch,
)
def _(
bundle_choice,
bundle_path_text,
bundle_picker,
bundle_zip,
extract_bundle_zip,
is_allowed_bundle_dir,
mo,
online_switch,
pathlib,
run_bundle_checks,
upload_bytes,
):
bundle_dir = None
bundle_source = None
bundle_error = None
_zip_name, _zip_bytes = upload_bytes(bundle_zip.value)
if _zip_bytes is not None:
try:
bundle_dir = extract_bundle_zip(_zip_name, _zip_bytes)
bundle_source = f"uploaded `{_zip_name}` → `{bundle_dir}`"