#!/usr/bin/env python3 """ verify_audit.py - verify a Tellus audit integrity export, offline. python3 verify_audit.py [] [--anchor ] [--lang sv|en] This is the readable twin of tellus-audit-verify, the .NET verifier. It exists for anyone who would rather read the code that decides than run a binary: it is one file, it uses only the Python standard library, and it never touches the network. Both verifiers must print exactly the same thing for the same file -- Tellus's CI runs them side by side and fails the build if they ever disagree. What it proves, in order (numbered as in the Tellus verification manual): 0 Format the file is a complete, well-formed export of one chain 2 Linkage every event's prevHash is the previous event's eventHash, starting from 32 zero bytes (the genesis of the chain) 3 Recomputation every eventHash is SHA-256(prevHash || canonical(rawEvent)) 4 Anchor binding the chain holds, at the anchored position, exactly the head and the number of events that the anchor signed Check 1, the anchor's signature, is not here on purpose: it is done with openssl and the key published on the Tellus trust portal (verify-anchor.sh), so the one step that establishes trust runs on no code from Tellus at all. The hard part is check 3. The hash covers a canonical form of each event's JSON, and that form must be reproduced byte for byte. Python's json.dumps does not do it -- it escapes differently -- so the canonicalisation below is written out by hand, and its rules are the specification. See canonicalize(). Exit status: 0 verified, 1 not verified, 3 consistent but not anchored, 2 could not run. """ import base64 import hashlib import io import json import os import re import shutil import subprocess import sys import tempfile import time import zipfile from datetime import datetime KIND = "tellus-audit-integrity-export" FORMAT_VERSION = 1 EVENTS_ENTRY = "events.ndjson" ANCHOR_ENTRY = "anchor.json" START_ANCHOR_ENTRY = "start-anchor.json" MANIFEST_ENTRY = "manifest.json" GENESIS = bytes(32) MAX_DEPTH = 64 # deeper JSON is refused, as by the .NET parser INT64_MIN, INT64_MAX = -(2 ** 63), 2 ** 63 - 1 VERIFIED, NOT_VERIFIED, COULD_NOT_RUN, CONSISTENT_BUT_UNANCHORED = 0, 1, 2, 3 # Check ids, in the order they run. FORMAT, SIGNATURE, LINKAGE, RECOMPUTATION, ANCHOR_BINDING, CONTINUITY = 0, 1, 2, 3, 4, 5 CHECKS = (FORMAT, SIGNATURE, LINKAGE, RECOMPUTATION, ANCHOR_BINDING, CONTINUITY) CHECK_NAMES = {FORMAT: "Format", SIGNATURE: "Signature", LINKAGE: "Linkage", RECOMPUTATION: "Recomputation", ANCHOR_BINDING: "Anchor binding", CONTINUITY: "Continuity"} PASSED, FAILED, NOT_RUN = "PASS", "FAIL", " -- " USAGE = """usage: tellus-audit-verify [] [--anchor ] [--lang sv|en] Verifies a Tellus audit integrity export offline: the checkpoint was signed by Tellus, every event links to the one before it, every hash recomputes from the event's content, and the chain carries exactly the head the checkpoint signed. the verification key published on the Tellus trust portal. Without it the signature check is skipped and the rest still runs. --anchor bind to this anchor instead of the one inside the export, e.g. one you saved from an earlier export --lang answer in this language (sv or en). Without it, the language the export was asked for in, then the system's, then English. --report also save the answer as a file, to keep with the export --seen the memory of checkpoints verified before (check 5). Defaults to a file beside this program; keep it, it is what lets a later export be held against what you already saw. --forget do not add this export's checkpoint to that memory You can also check the signature without this program, with openssl and verify-anchor.sh -- the guide on the trust portal shows how. Exit status: 0 verified, 1 not verified, 3 consistent but not anchored, 2 could not run. """ class FormatProblem(Exception): """The file is not a well-formed export -- before any question of whether its contents are genuine.""" class InvalidJson(Exception): pass # --------------------------------------------------------------------------- # Reading JSON the way the .NET verifier reads it. # # Numbers are kept as their exact source text (the hash covers the token, not # the value), NaN/Infinity are refused, nesting beyond 64 levels is refused, and # a string holding a lone surrogate is refused when it is read (it has no UTF-8 # form). Objects keep every key/value pair in order, duplicates included. # --------------------------------------------------------------------------- class Number(str): """A JSON number, as its exact source token.""" class Pairs(list): """A JSON object: its (key, value) pairs in document order.""" def get(self, name, default=None): # The last definition wins, as with .NET's JsonElement.TryGetProperty. for key, value in reversed(self): if key == name: return value return default def has(self, name): return any(key == name for key, _ in self) _MISSING = object() def parse_json(text): """Parse like .NET's JsonDocument: containers nested deeper than 64 levels are refused at parse time. A lone surrogate is NOT refused here -- .NET only refuses it when the string is read, so string_field() and canonicalize() do.""" def refuse_constant(token): raise InvalidJson(token) try: value = json.loads(text, object_pairs_hook=Pairs, parse_int=Number, parse_float=Number, parse_constant=refuse_constant) except (ValueError, RecursionError): raise InvalidJson("not JSON") # Iterative, so a hostile document cannot exhaust Python's recursion limit # before the depth rule gets to refuse it. stack = [(value, 1)] while stack: node, depth = stack.pop() if isinstance(node, (Pairs, list)): if depth > MAX_DEPTH: raise InvalidJson("too deep") for item in (node if not isinstance(node, Pairs) else [v for _, v in node]): stack.append((item, depth + 1)) return value def has_lone_surrogate(text): return any(0xD800 <= ord(ch) <= 0xDFFF for ch in text) # --------------------------------------------------------------------------- # The canonical form -- the specification of what check 3 hashes. # # * objects: keys sorted by UTF-16 code unit (ordinal), duplicates kept in # their original order; arrays keep their order; no whitespace # * numbers: exactly the token as written # * true / false / null as themselves # * strings: printable ASCII 0x20-0x7E is written as itself, EXCEPT # " & ' + < > ` which become \\u0022 \\u0026 \\u0027 ... # backslash becomes \\\\, and \\b \\t \\n \\f \\r use their short forms; # every other character -- other control characters, DEL, and ALL # non-ASCII -- becomes \\uXXXX with upper-case hex, and a character beyond # U+FFFF becomes its two UTF-16 surrogates, each as \\uXXXX. # # This is what .NET's Utf8JsonWriter emits by default; the CI cross-validation # checks this function against it for every one of the 65,536 BMP characters. # --------------------------------------------------------------------------- _SHORT_ESCAPES = {"\b": "\\b", "\t": "\\t", "\n": "\\n", "\f": "\\f", "\r": "\\r", "\\": "\\\\"} _ESCAPED_ASCII = set('"&\'+<>`') def _canonical_string(text): if has_lone_surrogate(text): raise InvalidJson("lone surrogate") out = ['"'] for ch in text: code = ord(ch) if ch in _SHORT_ESCAPES: out.append(_SHORT_ESCAPES[ch]) elif 0x20 <= code <= 0x7E and ch not in _ESCAPED_ASCII: out.append(ch) elif code > 0xFFFF: code -= 0x10000 out.append("\\u%04X\\u%04X" % (0xD800 + (code >> 10), 0xDC00 + (code & 0x3FF))) else: out.append("\\u%04X" % code) out.append('"') return "".join(out) def _utf16_order(key): return key.encode("utf-16-be") def _canonical(value, out): if isinstance(value, Pairs): if any(has_lone_surrogate(key) for key, _ in value): raise InvalidJson("lone surrogate") out.append("{") for index, (key, item) in enumerate(sorted(value, key=lambda pair: _utf16_order(pair[0]))): if index: out.append(",") out.append(_canonical_string(key)) out.append(":") _canonical(item, out) out.append("}") elif isinstance(value, list): out.append("[") for index, item in enumerate(value): if index: out.append(",") _canonical(item, out) out.append("]") elif isinstance(value, Number): out.append(str.__str__(value)) elif isinstance(value, str): out.append(_canonical_string(value)) elif value is True: out.append("true") elif value is False: out.append("false") else: out.append("null") def canonicalize(json_text): """The canonical form of a JSON document, as text.""" out = [] _canonical(parse_json(json_text), out) return "".join(out) def event_hash(prev_hash, raw_event): return hashlib.sha256(prev_hash + canonicalize(raw_event).encode("utf-8")).digest() # --------------------------------------------------------------------------- # Field rules -- mirrored one for one from TellusAudit.Integrity/StrictJson.cs. # Each returns None for "missing or not of that shape". # --------------------------------------------------------------------------- _INTEGER = re.compile(r"-?(0|[1-9][0-9]*)") _GUID = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}") _HASH = re.compile(r"[0-9a-fA-F]{64}") _BASE64 = re.compile(r"[A-Za-z0-9+/]*={0,2}") def string_field(obj, name): value = obj.get(name, _MISSING) if not isinstance(value, str) or isinstance(value, Number) or has_lone_surrogate(value): return None return value def integer_field(obj, name): value = obj.get(name, _MISSING) if not isinstance(value, Number) or len(value) > 20 or not _INTEGER.fullmatch(value): return None number = int(value) return number if INT64_MIN <= number <= INT64_MAX else None def guid_field(obj, name): text = string_field(obj, name) return text.lower() if text is not None and _GUID.fullmatch(text) else None def hash_field(obj, name): text = string_field(obj, name) return bytes.fromhex(text) if text is not None and _HASH.fullmatch(text) else None def base64_field(obj, name): import base64 text = string_field(obj, name) if text is None or len(text) % 4 != 0 or not _BASE64.fullmatch(text): return None return base64.b64decode(text) def decode_utf8(data): try: return data.decode("utf-8") except UnicodeDecodeError: return None # --------------------------------------------------------------------------- # Check 1 -- the checkpoint's signature (ECDSA P-256 over SHA-256). # # Done with openssl, the same way verify-anchor.sh does it and the same way the # compiled verifier does it internally. Python's standard library has no ECDSA, # and this file deliberately depends on nothing else -- so the one piece of # cryptography is handed to the tool every system already has. # --------------------------------------------------------------------------- def verify_anchor_signature(anchor_json, key_pem): """None when the signature holds; otherwise a plain sentence saying what is wrong.""" try: root = parse_json(anchor_json) except InvalidJson: return "the anchor is not valid JSON" if not isinstance(root, Pairs): return "the anchor is not a JSON object" signed = base64_field(root, "signedPayload") if signed is None: return "the anchor carries no signedPayload, so there is nothing to check the signature over" signature_hex = string_field(root, "signature") parsed = _parse_vault_signature(signature_hex) if signature_hex is not None else None if parsed is None: return "the anchor is not signed in the published form (vault:vN:...), so no published key can check it" key_version, der_signature = parsed if shutil.which("openssl") is None: return "openssl is not installed, so the signature could not be checked" # The file may hold every published key version, so every key in it is a candidate for # whichever version signed this anchor. Trying several is not a weakening: a signature either # verifies under a key or it does not, and a customer holding the portal's whole key list must # not get a false alarm on an export signed before the last rotation. A file with no PEM block # at all is still tried, so the reader is told what is wrong with their key file. candidates = pem_blocks(key_pem) or [key_pem] first_problem = None for pem in candidates: problem = _verify_under(pem, signed, der_signature, key_version) if problem is None: return None if first_problem is None: first_problem = problem return first_problem def _verify_under(key_pem, signed, der_signature, key_version): with tempfile.TemporaryDirectory() as workdir: payload_path = "%s/payload" % workdir signature_path = "%s/signature.der" % workdir key_path = "%s/key.pem" % workdir for path, data in ((payload_path, signed), (signature_path, der_signature), (key_path, key_pem.encode("utf-8"))): with open(path, "wb") as handle: handle.write(data) probe = subprocess.run(["openssl", "pkey", "-pubin", "-in", key_path, "-noout"], capture_output=True) if probe.returncode != 0: return "the key file is not a public key in PEM form" result = subprocess.run( ["openssl", "dgst", "-sha256", "-verify", key_path, "-signature", signature_path, payload_path], capture_output=True) return None if result.returncode == 0 else \ "the key for v%d does not verify this anchor's signature" % key_version def pem_blocks(text): """The PEM blocks in a key file, in the order they appear. A customer is given one file holding every published key version, and openssl reads only the first block, so they are split here instead of asking the reader to cut the file up.""" blocks = [] start = 0 while True: begin = text.find("-----BEGIN ", start) if begin < 0: return blocks end = text.find("-----END ", begin) if end < 0: return blocks close = text.find("-----", end + len("-----END ")) if close < 0: return blocks close += len("-----") blocks.append(text[begin:close]) start = close def _parse_vault_signature(signature_hex): """The signature is hex of the text "vault:vN:". The prefix names the key version, which is what lets a rotated key keep verifying old anchors.""" if not re.fullmatch(r"(?:[0-9a-fA-F]{2})*", signature_hex): return None text = decode_utf8(bytes.fromhex(signature_hex)) if text is None: return None parts = text.split(":", 2) if len(parts) != 3 or parts[0] != "vault" or not re.fullmatch(r"v\d+", parts[1]): return None try: return int(parts[1][1:]), base64.b64decode(parts[2], validate=True) except (ValueError, TypeError): return None # --------------------------------------------------------------------------- # The anchor document -- mirrored from AuditAnchorPublication.Read. # --------------------------------------------------------------------------- _SIGNED_TIMESTAMP = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{7}Z") _DIGITS = re.compile(r"0|[1-9][0-9]*") _UPPER_HEX = re.compile(r"(?:[0-9A-F]{2})*") class Anchor: def __init__(self, anchor_id, chain_key, head_hash, head_seq, event_count, anchored_at, from_signed): self.anchor_id = anchor_id self.chain_key = chain_key self.head_hash = head_hash self.head_seq = head_seq self.event_count = event_count self.anchored_at = anchored_at self.from_signed = from_signed def payload_from_signed_bytes(data): """The five values out of the bytes the anchor's signature covers, or None unless they are exactly a Tellus anchor payload: five fields, each ended by 0x1F, that would be written back as these same bytes.""" text = decode_utf8(data) if text is None: return None fields = text.split("\x1f") if len(fields) != 6 or fields[5] != "": return None chain_key, head_hex, seq, count, anchored_at = fields[:5] if not _UPPER_HEX.fullmatch(head_hex): return None if not _DIGITS.fullmatch(seq) or not _DIGITS.fullmatch(count): return None if int(seq) > INT64_MAX or int(count) > INT64_MAX: return None if not _SIGNED_TIMESTAMP.fullmatch(anchored_at): return None try: datetime.strptime(anchored_at[:19], "%Y-%m-%dT%H:%M:%S") except ValueError: return None return chain_key, bytes.fromhex(head_hex), int(seq), int(count), anchored_at def read_anchor(text): try: root = parse_json(text) except InvalidJson: raise FormatProblem("the anchor is not valid JSON") if not isinstance(root, Pairs): raise FormatProblem("the anchor is not a JSON object") def field_error(name): return FormatProblem("the anchor has a missing or invalid %s" % name) anchor_id = None if root.has("anchorId") and root.get("anchorId") is not None: anchor_id = integer_field(root, "anchorId") if anchor_id is None: raise field_error("anchorId") if root.has("signedPayload") and root.get("signedPayload") is not None: signed = base64_field(root, "signedPayload") if signed is None: raise field_error("signedPayload") payload = payload_from_signed_bytes(signed) if payload is None: raise FormatProblem("the anchor's signedPayload is not a Tellus anchor payload") return Anchor(anchor_id, *payload, from_signed=True) anchored_at = string_field(root, "anchoredAt") if anchored_at is None: raise field_error("anchoredAt") chain_key = string_field(root, "chainKey") if chain_key is None: raise field_error("chainKey") head = hash_field(root, "headEventHash") if head is None: raise field_error("headEventHash") seq = integer_field(root, "headIngestSeq") if seq is None: raise field_error("headIngestSeq") count = integer_field(root, "eventCount") if count is None: raise field_error("eventCount") return Anchor(anchor_id, chain_key, head, seq, count, anchored_at, from_signed=False) # --------------------------------------------------------------------------- # The export -- mirrored from AuditIntegrityExportReader. # --------------------------------------------------------------------------- class Manifest: def __init__(self, root): def field_error(name): return FormatProblem("the manifest has a missing or invalid %s" % name) if string_field(root, "kind") != KIND: raise FormatProblem("the file is not a Tellus audit integrity export") version = integer_field(root, "formatVersion") if version is None: raise field_error("formatVersion") if version != FORMAT_VERSION: raise FormatProblem( "format version %d is not supported by this verifier (it reads version %d); " "use a newer verifier" % (version, FORMAT_VERSION)) self.chain_key = string_field(root, "chainKey") if self.chain_key is None: raise field_error("chainKey") self.event_count = integer_field(root, "eventCount") if self.event_count is None: raise field_error("eventCount") self.first_seq = integer_field(root, "firstIngestSeq") self.last_seq = integer_field(root, "lastIngestSeq") self.from_seq = integer_field(root, "fromIngestSeq") self.start_anchor_id = integer_field(root, "startAnchorId") self.head_hash = string_field(root, "headEventHash") self.events_sha256 = string_field(root, "eventsSha256") if self.events_sha256 is None: raise field_error("eventsSha256") # Presentation only, and never a reason to reject a file: an unknown language reads as # none, and the verifier falls back exactly as it does without the field. self.reader_language = normalize_language(string_field(root, "readerLanguage")) class Export: def __init__(self, path): try: self.zip = zipfile.ZipFile(path) except zipfile.BadZipFile: raise FormatProblem("the file is not a zip archive") names = set(self.zip.namelist()) if MANIFEST_ENTRY not in names: raise FormatProblem("the export has no %s" % MANIFEST_ENTRY) manifest_bytes = self._read(MANIFEST_ENTRY) if EVENTS_ENTRY not in names: raise FormatProblem("the export has no %s" % EVENTS_ENTRY) manifest_text = decode_utf8(manifest_bytes) if manifest_text is None: raise FormatProblem("%s is not valid UTF-8" % MANIFEST_ENTRY) try: root = parse_json(manifest_text) except InvalidJson: raise FormatProblem("the manifest is not valid JSON") if not isinstance(root, Pairs): raise FormatProblem("the manifest is not a JSON object") self.manifest = Manifest(root) self.anchor_json = None if ANCHOR_ENTRY in names: self.anchor_json = decode_utf8(self._read(ANCHOR_ENTRY)) if self.anchor_json is None: raise FormatProblem("%s is not valid UTF-8" % ANCHOR_ENTRY) self.start_anchor_json = None if START_ANCHOR_ENTRY in names: self.start_anchor_json = decode_utf8(self._read(START_ANCHOR_ENTRY)) if self.start_anchor_json is None: raise FormatProblem("%s is not valid UTF-8" % START_ANCHOR_ENTRY) def _read(self, name): try: return self.zip.read(name) except (zipfile.BadZipFile, EOFError, OSError, ValueError): raise FormatProblem("the zip archive is damaged") def events(self): """(line number, event) in file order. Lines end at LF, a CR before it is dropped, a last line without LF still counts; each line is strict UTF-8 on its own.""" try: with self.zip.open(EVENTS_ENTRY) as stream: pending = b"" number = 0 while True: chunk = stream.read(64 * 1024) if not chunk: break pending += chunk *lines, pending = pending.split(b"\n") for line in lines: number += 1 yield number, self._parse_event(line, number) if pending: number += 1 yield number, self._parse_event(pending, number) except (zipfile.BadZipFile, EOFError, OSError, ValueError): raise FormatProblem("the zip archive is damaged") @staticmethod def _parse_event(line, number): if line.endswith(b"\r"): line = line[:-1] def line_error(problem): return FormatProblem("%s line %d %s" % (EVENTS_ENTRY, number, problem)) text = decode_utf8(line) if text is None: raise line_error("is not valid UTF-8") try: root = parse_json(text) except InvalidJson: raise line_error("is not valid JSON") if not isinstance(root, Pairs): raise line_error("is not a JSON object") event = {} for name, rule in (("chainKey", string_field), ("ingestSeq", integer_field), ("eventId", guid_field), ("prevHash", hash_field), ("eventHash", hash_field), ("rawEvent", string_field)): value = rule(root, name) if value is None: raise line_error("has a missing or invalid %s" % name) event[name] = value return event def events_sha256(self): digest = hashlib.sha256() try: with self.zip.open(EVENTS_ENTRY) as stream: for chunk in iter(lambda: stream.read(64 * 1024), b""): digest.update(chunk) except (zipfile.BadZipFile, EOFError, OSError, ValueError): raise FormatProblem("the zip archive is damaged") return digest.hexdigest() # --------------------------------------------------------------------------- # The checks -- mirrored from AuditIntegrityExportVerifier. # --------------------------------------------------------------------------- class Report: def __init__(self, manifest): self.manifest = manifest self.results = {} self.events_verified = 0 self.anchor = None self.start_anchor = None self.anchor_supplied = False self.signature_checked = False self.events_after_anchor = 0 self.verdict = None def fail(self, check, detail, event=None): self.results[check] = (FAILED, detail, event) if event is not None: if self.signature_checked: self.results.setdefault(SIGNATURE, (PASSED, None, None)) # The other per-event checks held for every event before this one: say so. for walk_check in (FORMAT, LINKAGE, RECOMPUTATION): self.results.setdefault(walk_check, ( NOT_RUN, "held for the first %d events; not checked past ingestSeq %d" % (self.events_verified, event["ingestSeq"]), None)) self.verdict = NOT_VERIFIED return self def walk_completed(self): if self.signature_checked: self.results.setdefault(SIGNATURE, (PASSED, None, None)) for check in (FORMAT, LINKAGE, RECOMPUTATION): self.results.setdefault(check, (PASSED, None, None)) def checks(self): for check in CHECKS: if check == SIGNATURE and not self.signature_checked and check not in self.results: yield check, (NOT_RUN, "no key given: run again with the key published on the Tellus trust portal, " "or check the signature yourself with verify-anchor.sh", None) continue yield check, self.results.get( check, (NOT_RUN, "not completed: verification stops at the first failure", None)) def check_status(self, wanted): for check, (status, _, _) in self.checks(): if check == wanted: return status return NOT_RUN def failed_check(self): return next(check for check, (status, _, _) in self.checks() if status == FAILED) def verify(path, anchor_json=None, key_pem=None, remembered=None): try: export = Export(path) except FormatProblem as problem: return Report(None).fail(FORMAT, str(problem)) manifest = export.manifest report = Report(manifest) anchor = None supplied = anchor_json is not None if anchor_json is None: anchor_json = export.anchor_json if anchor_json is not None: try: anchor = read_anchor(anchor_json) except FormatProblem as problem: return report.fail(FORMAT, str(problem)) report.anchor = anchor report.anchor_supplied = supplied if anchor.chain_key != manifest.chain_key: return report.fail(FORMAT, "the anchor is for chain %s, the export is chain %s" % (anchor.chain_key, manifest.chain_key)) # A segment continues from a signed checkpoint instead of from genesis. Its head hash is what # the first event must link to, and the events it counted are what the end anchor's count is # measured against -- so a segment proves exactly as much as a whole chain does, for the # stretch it covers, as long as BOTH checkpoints' signatures hold (check 1). start_anchor = None if export.start_anchor_json is not None: try: start_anchor = read_anchor(export.start_anchor_json) except FormatProblem as problem: return report.fail(FORMAT, "start checkpoint: %s" % problem) report.start_anchor = start_anchor if start_anchor.chain_key != manifest.chain_key: return report.fail(FORMAT, "the start checkpoint is for chain %s, the export is chain %s" % (start_anchor.chain_key, manifest.chain_key)) if manifest.from_seq is None or manifest.from_seq != start_anchor.head_seq: return report.fail(FORMAT, "the manifest says the export continues from ingestSeq %s, " "the start checkpoint ends at %d" % (_dotnet_nullable(manifest.from_seq), start_anchor.head_seq)) if anchor is not None and anchor.head_seq <= start_anchor.head_seq: return report.fail(FORMAT, "the anchor ends at ingestSeq %d, at or before the start checkpoint " "at %d - it covers none of these events" % (anchor.head_seq, start_anchor.head_seq)) elif manifest.from_seq is not None: return report.fail(FORMAT, "the manifest says the export continues from ingestSeq %d, but it carries no " "%s to continue from" % (manifest.from_seq, START_ANCHOR_ENTRY)) # Check 1 before the walk: if the checkpoints are not genuine, everything the walk # proves is about a history nobody vouched for. report.signature_checked = key_pem is not None if key_pem is not None: for json_text, what in ((anchor_json, "the anchor"), (export.start_anchor_json, "the start checkpoint")): if json_text is None: continue problem = verify_anchor_signature(json_text, key_pem) if problem is not None: return report.fail(SIGNATURE, "%s: %s" % (what, problem)) expected_prev = start_anchor.head_hash if start_anchor else GENESIS events_before_segment = start_anchor.event_count if start_anchor else 0 walked = 0 previous_seq = start_anchor.head_seq if start_anchor else None walked_to_anchor = None # Only the checkpoints this file could possibly answer: same chain, past where it starts. # One pass over the events settles all of them. to_compare = sorted( (c for c in (remembered or []) if c["chainKey"] == manifest.chain_key and c["headIngestSeq"] > (start_anchor.head_seq if start_anchor else 0)), key=lambda c: c["headIngestSeq"]) remembered_seqs = set(c["headIngestSeq"] for c in to_compare) seen_at_remembered_seq = {} try: for _, event in export.events(): seq = event["ingestSeq"] if event["chainKey"] != manifest.chain_key: return report.fail(FORMAT, "ingestSeq %d belongs to chain %s, not %s" % (seq, event["chainKey"], manifest.chain_key), event) if previous_seq is not None and seq <= previous_seq: return report.fail(FORMAT, "ingestSeq %d follows %d - events are not in chain order" % (seq, previous_seq), event) if event["prevHash"] != expected_prev: if walked != 0: detail = ("prevHash is not the previous event's eventHash: " "an event was removed, inserted or reordered here") elif start_anchor is None: detail = ("the first event does not start from genesis (32 zero bytes), " "so the start of the chain is missing") else: detail = ("the first event does not continue from the start checkpoint: " "its prevHash is not the head the checkpoint signed") return report.fail(LINKAGE, detail, event) try: recomputed = event_hash(event["prevHash"], event["rawEvent"]) except InvalidJson: return report.fail(RECOMPUTATION, "rawEvent is not valid JSON, so its hash cannot be recomputed", event) if recomputed != event["eventHash"]: return report.fail(RECOMPUTATION, "eventHash does not recompute from rawEvent: " "the event's content was changed", event) expected_prev = event["eventHash"] previous_seq = seq walked += 1 report.events_verified = walked if seq in remembered_seqs: seen_at_remembered_seq[seq] = (event["eventHash"].hex().upper(), walked) if anchor is not None and seq == anchor.head_seq: if event["eventHash"] != anchor.head_hash: return report.fail(ANCHOR_BINDING, "the event at the anchored ingestSeq is not the event " "the anchor signed: history was rewritten", event) walked_to_anchor = walked except FormatProblem as problem: return report.fail(FORMAT, str(problem)) report.walk_completed() if manifest.event_count != walked: return report.fail(FORMAT, "the manifest lists %d events, the file holds %d: the file is incomplete" % (manifest.event_count, walked)) if manifest.last_seq != previous_seq: return report.fail(FORMAT, "the manifest ends at ingestSeq %s, the file at %s" % (_dotnet_nullable(manifest.last_seq), _dotnet_nullable(previous_seq))) if walked > 0 and not _same_hex(manifest.head_hash, expected_prev.hex()): return report.fail(FORMAT, "the manifest's headEventHash is not the last event's hash") try: actual_sha = export.events_sha256() except FormatProblem as problem: return report.fail(FORMAT, str(problem)) if not _same_hex(manifest.events_sha256, actual_sha): return report.fail(FORMAT, "%s does not match the manifest's SHA-256: the file changed after export" % EVENTS_ENTRY) # Check 5 is evaluated here but applied after check 4 has had its say, so the checks are # reported in their own order: a file that fails its own anchor binding should be told that # first, not held against an older checkpoint it was never going to match. continuity_status, continuity_detail = _continuity( to_compare, seen_at_remembered_seq, events_before_segment, previous_seq) if anchor is None: # Check 4's own outcome is recorded before check 5 can fail, so a continuity failure never # leaves the line above it reading "not completed" about a check that finished. report.results[ANCHOR_BINDING] = ( NOT_RUN, "no anchor: the chain had not been anchored when it was exported" if start_anchor is None else "no anchor covers this stretch yet: it was recorded after the checkpoint it continues from", None) if continuity_status == FAILED: return report.fail(CONTINUITY, continuity_detail) report.results[CONTINUITY] = (continuity_status, continuity_detail, None) report.events_after_anchor = walked report.verdict = CONSISTENT_BUT_UNANCHORED return report if walked_to_anchor is None: return report.fail(ANCHOR_BINDING, "the anchored ingestSeq %d is not in the export: " "the chain was truncated or rewritten" % anchor.head_seq) if walked_to_anchor + events_before_segment != anchor.event_count: return report.fail(ANCHOR_BINDING, "the anchor signed %d events up to ingestSeq %d, the export accounts for %d" % (anchor.event_count, anchor.head_seq, walked_to_anchor + events_before_segment)) # An export can satisfy its own anchor perfectly and still contradict a checkpoint the customer # saw months ago. That is the case a single file cannot speak to, and the only one a # rewrite-and-re-anchor would survive. # Check 4 passed either way, and says so. report.results[ANCHOR_BINDING] = (PASSED, None, None) if continuity_status == FAILED: return report.fail(CONTINUITY, continuity_detail) report.results[CONTINUITY] = (continuity_status, continuity_detail, None) report.events_after_anchor = walked - walked_to_anchor report.verdict = VERIFIED return report def _same_hex(claimed, actual): # ASCII-only case folding, like .NET's OrdinalIgnoreCase on hex. Python's # str.upper() would fold some non-ASCII letters into ASCII ("\ufb00" -> "FF"). return claimed is not None and claimed.isascii() and claimed.upper() == actual.upper() def _continuity(to_compare, seen, events_before_segment, last_seq): """Check 5 against every checkpoint this program verified before. A checkpoint newer than this export says nothing about it -- the customer may simply be re-checking an older file -- so it is passed over rather than counted against the export.""" matched = 0 oldest = None for checkpoint in to_compare: if checkpoint["headIngestSeq"] > (last_seq or 0): continue found = seen.get(checkpoint["headIngestSeq"]) if found is None: return FAILED, ("a checkpoint Tellus signed on %s, which this program verified before, ends at " "ingestSeq %d, but that event is not in this export: the chain was truncated or rewritten" % (checkpoint["anchoredAt"], checkpoint["headIngestSeq"])) found_hash, found_walked = found if found_hash.upper() != checkpoint["headEventHash"].upper(): return FAILED, ("the event at ingestSeq %d is not the one in the checkpoint Tellus signed on %s, " "which this program verified before: this export and that checkpoint cannot both be this chain" % (checkpoint["headIngestSeq"], checkpoint["anchoredAt"])) if found_walked + events_before_segment != checkpoint["eventCount"]: return FAILED, ("the checkpoint Tellus signed on %s, which this program verified before, counted %d events up to ingestSeq %d, " "this export accounts for %d" % (checkpoint["anchoredAt"], checkpoint["eventCount"], checkpoint["headIngestSeq"], found_walked + events_before_segment)) matched += 1 if oldest is None: oldest = checkpoint if matched > 0: return PASSED, ("agrees with %d checkpoint(s) this program verified before, the oldest taken %s" % (matched, oldest["anchoredAt"])) if not to_compare: return NOT_RUN, "nothing to compare with yet: this is the first export of this chain this program has verified" return NOT_RUN, "nothing to compare with in this export: every checkpoint this program remembers is newer than it" def _dotnet_nullable(value): # .NET interpolates a null long? as the empty string. return "" if value is None else str(value) # --------------------------------------------------------------------------- # Output -- mirrored from TellusAudit.Verifier/ReportRenderer.cs. # --------------------------------------------------------------------------- def ascii_only(text): """Printable ASCII and newlines pass through; everything else becomes \\uXXXX per UTF-16 code unit, so the output is the same bytes on every platform.""" out = [] for ch in text: code = ord(ch) if ch == "\n" or 0x20 <= code <= 0x7E: out.append(ch) elif code > 0xFFFF: code -= 0x10000 out.append("\\u%04X\\u%04X" % (0xD800 + (code >> 10), 0xDC00 + (code & 0x3FF))) else: out.append("\\u%04X" % code) return "".join(out) READER_LANGUAGES = ("sv", "en") DEFAULT_LANGUAGE = "en" def normalize_language(language): """A supported language, compared on the primary subtag, or None.""" if not language or not language.strip(): return None primary = language.strip().lower().replace("_", "-").split("-")[0] return primary if primary in READER_LANGUAGES else None def operating_system_language(): """The language the environment states. Windows sets none of these, so a Windows reader gets the language the export carries -- which is the better answer anyway, being the one they chose in Tellus rather than the one their IT department installed.""" for name in ("LC_ALL", "LC_MESSAGES", "LANG", "LANGUAGE"): value = os.environ.get(name) if value and value.strip() and value not in ("C", "POSIX"): return re.split(r"[.:]", value)[0] return None def resolve_language(requested, manifest): """What the reader asked for, else what the export says its requester reads, else the operating system's language, else English.""" return (normalize_language(requested) or normalize_language(manifest.reader_language if manifest else None) or normalize_language(operating_system_language()) or DEFAULT_LANGUAGE) def headline(report, language): """The lines a customer actually reads, in their own language, before the technical report. Real UTF-8, unlike the report, which escapes everything outside printable ASCII because it repeats values out of a file an attacker may have written. These are our own literals. The one value taken from the file -- the checkpoint's timestamp -- is escaped before it goes in. Every line here also exists in TellusAudit.Verifier/VerdictHeadline.cs, word for word, and the cross-validation runs the whole gallery in both languages.""" sv = language == "sv" lines = [] if report.verdict == VERIFIED: lines.append("Loggen är oförändrad." if sv else "The log is unaltered.") lines.append( ("%d händelser, kontrollerade mot en kontrollpunkt som Tellus signerade %s." % (report.events_verified, _moment(report))) if sv else ("%d events, checked against a checkpoint Tellus signed %s." % (report.events_verified, _moment(report)))) checked = report.check_status(SIGNATURE) == PASSED if sv: lines.append("Kontrollpunktens signatur stämmer mot nyckeln ni angav." if checked else "Signaturen är INTE kontrollerad — kör igen med nyckeln från trust-portalen.") else: lines.append("The checkpoint's signature matches the key you supplied." if checked else "The signature was NOT checked — run again with the key from the trust portal.") elif report.verdict == CONSISTENT_BUT_UNANCHORED: lines.append("Loggen hänger ihop, men ingen signerad kontrollpunkt täcker den ännu." if sv else "The log holds together, but no signed checkpoint covers it yet.") lines.append( ("%d händelser hänger ihop. Det är normalt strax efter att händelser registrerats — " "exportera igen senare." % report.events_verified) if sv else ("%d events hold together. This is normal shortly after events are recorded — " "export again later." % report.events_verified)) else: failed = report.failed_check() lines.append("Något stämmer inte." if sv else "Something does not add up.") lines.append(("Kontrollen %s underkändes." % _failed_check_name(failed, sv)) if sv else ("The %s check failed." % _failed_check_name(failed, sv))) # Re-downloading is the right first move for every failure except check 5, where the file # is fine and it is the history that disagrees with itself. if failed == CONTINUITY: lines.append("Fråga Tellus om en återställning från säkerhetskopia har deklarerats för er logg." if sv else "Ask Tellus whether a restore from backup has been declared for your log.") else: lines.append("Ladda ner exporten igen först — en avbruten nedladdning ger samma svar." if sv else "Download the export again first — an interrupted download gives the same answer.") lines.append("Kvarstår det: spara filerna precis som ni fick dem och kontakta Tellus." if sv else "If it persists, keep the files exactly as you received them and contact Tellus.") lines.append("") lines.append("Den tekniska rapporten nedan är på engelska." if sv else "The technical report follows.") return "\n".join(lines) + "\n\n" _FAILED_CHECK_NAMES_SV = { FORMAT: "av filens form (Format)", SIGNATURE: "av signaturen (Signature)", LINKAGE: "av länkningen (Linkage)", RECOMPUTATION: "av omräkningen (Recomputation)", ANCHOR_BINDING: "av ankarbindningen (Anchor binding)", CONTINUITY: "mot vad ni redan sett (Continuity)", } _FAILED_CHECK_NAMES_EN = { FORMAT: "format", SIGNATURE: "signature", LINKAGE: "linkage", RECOMPUTATION: "recomputation", ANCHOR_BINDING: "anchor binding", CONTINUITY: "continuity", } def _failed_check_name(check, sv): return (_FAILED_CHECK_NAMES_SV if sv else _FAILED_CHECK_NAMES_EN)[check] def _moment(report): """The checkpoint's own timestamp, cut to the minute rather than formatted, so the two verifiers cannot disagree about it and no time zone of the reader's machine enters into it.""" text = report.anchor.anchored_at if report.anchor is not None else None if text is None or len(text) < 16 or text[10] != "T": return "?" return ascii_only("%s %s UTC" % (text[:10], text[11:16])) MEMORY_FILE = "tellus-checkpoints-seen.json" def memory_default_path(): """Beside the program, which is where the download puts it and where it stays from one export to the next. The export's own folder would be wrong: a new export usually lands in a new folder, and the memory would start empty every time.""" return os.path.join(os.path.dirname(os.path.abspath(__file__)) or ".", MEMORY_FILE) def memory_load(path): """What the program has seen. A missing, unreadable or malformed file is an empty memory and nothing more -- it must never be a reason to refuse to verify an export.""" try: with open(path, "rb") as handle: text = decode_utf8(handle.read()) except OSError: return [] if text is None: return [] try: root = parse_json(text) except InvalidJson: return [] if not isinstance(root, Pairs): return [] entries = root.get("checkpoints") if not isinstance(entries, list): return [] checkpoints = [] for entry in entries: if not isinstance(entry, Pairs): continue chain_key = string_field(entry, "chainKey") head_hash = string_field(entry, "headEventHash") anchored_at = string_field(entry, "anchoredAt") head_seq = integer_field(entry, "headIngestSeq") event_count = integer_field(entry, "eventCount") if None in (chain_key, head_hash, anchored_at, head_seq, event_count): continue checkpoints.append({ "chainKey": chain_key, "anchorId": integer_field(entry, "anchorId"), "headIngestSeq": head_seq, "headEventHash": head_hash, "eventCount": event_count, "anchoredAt": anchored_at, }) return checkpoints def memory_remember(path, report): """Add this export's checkpoints to the memory, unless it did not verify or its signature was not checked -- a memory that could be seeded from an unverified file would vouch for whatever was put in it, which is worse than no memory at all. An entry already held is kept as it is: the older record has been sitting there longest, and overwriting it with a newer claim about the same position would quietly dissolve the very evidence check 5 rests on. Best-effort; a read-only folder is worth a line and nothing more.""" if report.verdict != VERIFIED or report.check_status(SIGNATURE) != PASSED: return False known = {} for checkpoint in memory_load(path): known[(checkpoint["chainKey"], checkpoint["headIngestSeq"])] = checkpoint added = False for anchor in (report.start_anchor, report.anchor): if anchor is None: continue key = (anchor.chain_key, anchor.head_seq) if key in known: continue known[key] = { "chainKey": anchor.chain_key, "anchorId": anchor.anchor_id, "headIngestSeq": anchor.head_seq, "headEventHash": anchor.head_hash.hex().upper(), "eventCount": anchor.event_count, "anchoredAt": anchor.anchored_at, } added = True if not added: return False lines = ['{', ' "about": "Checkpoints this program has verified. Keep this file: it is what lets ' 'a later export be held against what you already saw.",', ' "checkpoints": ['] ordered = sorted(known.values(), key=lambda c: (c["chainKey"], c["headIngestSeq"])) for index, checkpoint in enumerate(ordered): fields = [' "chainKey": %s' % json.dumps(checkpoint["chainKey"])] if checkpoint["anchorId"] is not None: fields.append(' "anchorId": %d' % checkpoint["anchorId"]) fields.append(' "headIngestSeq": %d' % checkpoint["headIngestSeq"]) fields.append(' "headEventHash": %s' % json.dumps(checkpoint["headEventHash"])) fields.append(' "eventCount": %d' % checkpoint["eventCount"]) fields.append(' "anchoredAt": %s' % json.dumps(checkpoint["anchoredAt"])) lines.append(" {") lines.append(",\n".join(fields)) lines.append(" }" + ("," if index + 1 < len(ordered) else "")) lines.append(" ]") lines.append("}") try: with open(path, "w", encoding="utf-8") as handle: handle.write("\n".join(lines) + "\n") except OSError: return False return True def save_report(path, text, language): """The answer as a file to keep. What an auditor needs is not a terminal window that closes, but something to put with the export in the file. Best-effort: a read-only folder is a reason to say so, never a reason to withhold a verification that already succeeded.""" sv = language == "sv" when = time.strftime("%Y-%m-%d %H:%M") footer = (("\nKontrollen kördes %s med verify_audit.py.\n" "Filen är en kopia av vad programmet skrev ut. Spara den tillsammans med exporten.\n" % when) if sv else ("\nThis check was run %s with verify_audit.py.\n" "The file is a copy of what the program printed. Keep it with the export.\n" % when)) try: with open(path, "w", encoding="utf-8") as handle: handle.write(text + footer) except OSError: return (("\nKunde inte spara svaret i %s - kontrollen gäller ändå.\n" % ascii_only(path)) if sv else ("\nCould not save the answer in %s - the check still stands.\n" % ascii_only(path))) return (("\nSvaret är sparat i %s\n" % ascii_only(path)) if sv else ("\nThe answer is saved in %s\n" % ascii_only(path))) def render(path, report): lines = [] manifest = report.manifest lines.append("Tellus audit integrity verification") lines.append("") lines.append(" File %s" % path) lines.append(" Chain %s" % (manifest.chain_key if manifest else "unknown")) if manifest is None: events = "unknown" elif manifest.event_count == 0: events = "0 listed in the manifest" else: events = "%d listed in the manifest, ingestSeq %s to %s" % ( manifest.event_count, "?" if manifest.first_seq is None else manifest.first_seq, "?" if manifest.last_seq is None else manifest.last_seq) lines.append(" Events %s" % events) start = report.start_anchor if start is not None: lines.append(" From #%s: continues after ingestSeq %d, %d events before this segment, taken %s" % ( "?" if start.anchor_id is None else start.anchor_id, start.head_seq, start.event_count, start.anchored_at)) anchor = report.anchor if anchor is None: anchor_line = "unknown" if manifest is None else "none" else: anchor_line = "#%s: %d events up to ingestSeq %d, taken %s, %s" % ( "?" if anchor.anchor_id is None else anchor.anchor_id, anchor.event_count, anchor.head_seq, anchor.anchored_at, "supplied with --anchor" if report.anchor_supplied else "bundled with the export") if not anchor.from_signed: anchor_line += "\n (read from its readable fields: this anchor predates signedPayload)" lines.append(" Anchor %s" % anchor_line) lines.append("") for check, (status, detail, event) in report.checks(): lines.append(" [%s] %d %s" % (status, check, CHECK_NAMES[check])) if event is not None: lines.append(" at ingestSeq %d, event %s" % (event["ingestSeq"], event["eventId"])) # Printed whatever the status: a check that passed and has something to say -- check 5 # naming what it matched -- is exactly the line a reader wants to see. if detail is not None: lines.append(" %s" % detail) lines.append("") lines.extend(_verdict(report)) return ascii_only("\n".join(lines) + "\n") def _verdict(report): if report.verdict == VERIFIED: lines = [ "VERIFIED - the events in this export are the history behind the anchor.", "", ] if report.start_anchor is None: lines.append(" %d events link to one another from the start of the chain," % report.events_verified) else: lines.append(" %d events link to one another from the signed checkpoint this" % report.events_verified) lines.append(" segment continues from,") lines += [ " every hash recomputes from its event's content, and the chain carries", " exactly the head the anchor signed.", ] if report.events_after_anchor > 0: lines += [ "", " %d events came after the anchor. They link and recompute," % report.events_after_anchor, " but no signed checkpoint covers them yet; a later export will.", ] signature_passed = any(status == PASSED for check, (status, _, _) in report.checks() if check == SIGNATURE) if signature_passed: return lines + ["", *( [" The checkpoint's own signature was checked against the key you supplied,"] if report.start_anchor is None else [" Both checkpoints' signatures were checked against the key you supplied,"]), " so this rests on nothing but that key. Make sure it is the one the", " Tellus trust portal publishes."] return lines + ["", *( [" All of this rests on the anchor being genuine. Check its signature by"] if report.start_anchor is None else [" All of this rests on both checkpoints being genuine. Check their signatures"]), " running this again with the key published on the Tellus trust portal,", " or with openssl and verify-anchor.sh."] if report.verdict == CONSISTENT_BUT_UNANCHORED: lines = [ "CONSISTENT, NOT ANCHORED - the chain holds together, but nothing signed vouches for it.", "", " %d events link and recompute, so no event was altered, removed" % report.events_verified, ] if report.start_anchor is None: return lines + [ " or inserted in place. Without an anchor this cannot rule out the whole", " chain having been rewritten. Verify a later export once the chain is", " anchored, or bind this one to an anchor you already hold with --anchor.", ] return lines + [ " or inserted in place, and they continue from a signed checkpoint. Until a", " later checkpoint covers them, nothing signed vouches for these events", " themselves. Export again after the next checkpoint.", ] failed = next(check for check, (status, _, _) in report.checks() if status == FAILED) return ["NOT VERIFIED - %s failed." % CHECK_NAMES[failed].lower(), ""] + _GUIDANCE[failed] _GUIDANCE = { SIGNATURE: [ " The checkpoint was not signed by the key you supplied. In order of likelihood:", " the key is the wrong version (the anchor names the version it was signed with),", " the file changed on the way to you, or the checkpoint is not authentic. Fetch the", " key from the Tellus trust portal again and compare its fingerprint before", " concluding anything.", ], FORMAT: [ " The file is not a complete, well-formed export. Download it again before", " drawing any conclusion: an interrupted or re-encoded download fails here.", ], LINKAGE: [ " An event is missing, extra or out of place at the position shown. A restored", " backup can cause this too, so ask Tellus whether a restore was declared for", " this chain. Keep the file exactly as you received it.", ], RECOMPUTATION: [ " The event at the position shown no longer matches its hash: its content is", " not what was recorded. Keep the file exactly as you received it and contact", " Tellus.", ], ANCHOR_BINDING: [ " The history in this export is not the history the anchor signed. First make", " sure the anchor belongs to this chain and is genuine (verify-anchor.sh). If it", " is, keep both files exactly as you received them and contact Tellus.", ], CONTINUITY: [ " This export disagrees with a checkpoint this program verified earlier. The", " likeliest explanation is a restore from backup, which breaks the chain and which", " Tellus records as a signed declaration - ask whether one was declared for this", " chain. If none was, the log you have been shown before and the log you were just", " given cannot both be the real one. Keep every file exactly as you received it,", " including the memory file, and contact Tellus.", ], } # --------------------------------------------------------------------------- def main(argv): out = sys.stdout.buffer err = sys.stderr.buffer export_path = anchor_path = key_path = language = report_path = memory_path = None remember = True index = 0 while index < len(argv): arg = argv[index] if arg in ("-h", "--help"): out.write(USAGE.encode("ascii")) return VERIFIED if arg == "--canonicalize" and index + 1 < len(argv): # For the cross-validation only: print the canonical form of a JSON file. with open(argv[index + 1], "rb") as source: out.write(canonicalize(source.read().decode("utf-8")).encode("utf-8")) return VERIFIED if arg == "--anchor" and index + 1 < len(argv) and anchor_path is None: anchor_path = argv[index + 1] index += 2 continue if arg == "--key" and index + 1 < len(argv) and key_path is None: key_path = argv[index + 1] index += 2 continue if arg == "--lang" and index + 1 < len(argv) and language is None: language = argv[index + 1] index += 2 continue if arg == "--report" and index + 1 < len(argv) and report_path is None: report_path = argv[index + 1] index += 2 continue if arg == "--seen" and index + 1 < len(argv) and memory_path is None: memory_path = argv[index + 1] index += 2 continue if arg == "--forget": remember = False index += 1 continue if not arg.startswith("-") and export_path is None: export_path = arg index += 1 continue # The second loose argument is the key, so the simplest possible command line is # "verify_audit.py " -- no flags to remember. if not arg.startswith("-") and key_path is None: key_path = arg index += 1 continue err.write(USAGE.encode("ascii")) return COULD_NOT_RUN if export_path is None: err.write(USAGE.encode("ascii")) return COULD_NOT_RUN anchor_json = None if anchor_path is not None: try: with open(anchor_path, "rb") as source: anchor_json = decode_utf8(source.read()) except OSError: anchor_json = None if anchor_json is None: err.write(("error: cannot read %s as UTF-8 text\n" % ascii_only(anchor_path)).encode("ascii")) return COULD_NOT_RUN key_pem = None if key_path is not None: try: with open(key_path, "rb") as source: key_pem = decode_utf8(source.read()) except OSError: key_pem = None if key_pem is None: err.write(("error: cannot read %s as UTF-8 text\n" % ascii_only(key_path)).encode("ascii")) return COULD_NOT_RUN try: with open(export_path, "rb") as source: data = source.read() except OSError: err.write(("error: cannot open %s\n" % ascii_only(export_path)).encode("ascii")) return COULD_NOT_RUN # What this program has verified before, so check 5 can hold this export against it. Reading # it is unconditional; it is the writing that --forget turns off. if memory_path is None: memory_path = memory_default_path() remembered = memory_load(memory_path) report = verify(io.BytesIO(data), anchor_json, key_pem, remembered) answer_in = resolve_language(language, report.manifest) text = headline(report, answer_in) + render(export_path, report) out.write(text.encode("utf-8")) if remember and memory_remember(memory_path, report): out.write((("\nKontrollpunkten är nu ihågkommen i %s, och nästa export hålls mot den.\n" % ascii_only(memory_path)) if answer_in == "sv" else ("\nThe checkpoint is now remembered in %s, and the next export will be held against it.\n" % ascii_only(memory_path))).encode("utf-8")) if report_path is not None: out.write(save_report(report_path, text, answer_in).encode("utf-8")) return report.verdict if __name__ == "__main__": sys.exit(main(sys.argv[1:]))