# Copyright (c) 2026 Tiago Pinto # SPDX-License-Identifier: MIT # # Independent producer-side implementation of # draft-fassbender-scitt-time-anchor-03, written from the published # Internet-Draft alone. See README.md and CONSTRUCTION.md. """upgrade.py: UPGRADE-PENDING-PROOFS (profile Appendix D.2). Converts calendar commitments into Bitcoin proofs, and promotes a proof to "anchored" only after performing, itself, the two checks a verifier performs in Section 3.1: comparison of the replayed digest against the Merkle root of the header at the attested height (step 2e), and confirmation depth at least MIN_CONFIRMATIONS (step 2f). That is precisely the gate Appendix C records as not evaluated by any deployment described in the profile. The header source is the implementer's own node (node_rpc), which satisfies the independence requirement directly (D.2: "A producing implementation that operates its own Bitcoin node satisfies that requirement directly"). Storage: state.json. INSERT/SELECT/UPDATE are non-normative placeholders (Section 1.4: "an implementation MAY use any storage mechanism"). Inline comments below are in Portuguese, the implementer's working language. """ import json import os import time from datetime import datetime, timezone import ots_wire import calendar_client from node_rpc import BitcoinRPC, header_merkle_root, header_ntime MIN_CONFIRMATIONS = 6 # §3.1 passo 9; Assunção A2; §6.11 UPGRADE_BATCH = 100 # operacional (D.2: "bear on latency ... and nothing else") FAIL_AFTER_SECONDS = 14 * 86400 # operacional; D.2 passo 2b def _now_iso(): return datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') def load_state(path): if os.path.exists(path): with open(path) as f: return json.load(f) return {'origin_attestations': {}, 'core_ots_proofs': {}} def save_state(state, path): tmp = path + '.tmp' with open(tmp, 'w') as f: json.dump(state, f, indent=2) os.replace(tmp, path) # escrita em cópia, troca atómica def _ots_upgrade(proof: ots_wire.Proof) -> bytes | None: """OTS-UPGRADE (§1.4): contacta os calendários das atestações pendentes e enxerta o caminho Bitcoin quando disponível. Semântica do NULL — decisão de leitura (candidato a defeito n.º 1): o §1.4 define NULL como "where no Bitcoin Merkle path is yet available"; o comentário do D.2 passo 2b sugere "the calendar returned nothing new". As duas leituras divergem para uma prova que JÁ carrega atestação Bitcoin mas ficou pendente no gate 2f: sob a leitura do 2b, OTS-UPGRADE devolve NULL em todos os ciclos e o 2b acaba por marcá-la "failed" — contradizendo o §2.4.1, que define "failed" como prova "whose calendar commitment was never answered". Sob a leitura estrita do §1.4, o ramo do 2c que persiste compromissos só-de-calendário fica inalcançável. Implementa-se aqui a única leitura que torna todos os ramos do D.2 alcançáveis e consistentes com o §2.4.1: devolver bytes se a prova (após fundir respostas) carrega caminho Bitcoin OU se algo foi fundido; NULL só quando nada mudou e não há caminho Bitcoin.""" changed = False for node, prefix in list(_walk_nodes(proof.root)): pendings = [a for a in node.attestations if a.kind == 'pending'] if not pendings: continue commitment = proof.replay(prefix, proof.msg) if commitment is None: continue for att in pendings: resp = calendar_client.get_timestamp(att.uri, commitment) if resp is None: continue node.attestations.extend(resp.attestations) node.children.extend(resp.children) changed = True has_bitcoin = any(a.kind == 'bitcoin' for _, a in proof.root.walk()) return proof.serialize() if (changed or has_bitcoin) else None def _walk_nodes(root: ots_wire.Node): """Itera (node, caminho_de_ops_até_ele).""" stack = [(root, [])] while stack: node, prefix = stack.pop() yield node, prefix for op, child in node.children: stack.append((child, prefix + [op])) def upgrade_pending_proofs(state, rpc: BitcoinRPC, log=print): """Algoritmo UPGRADE-PENDING-PROOFS() — D.2, passos 1 e 2a-2g.""" # Passo 1 — pendentes em ordem de criação, com o hash da origem # (necessário em 2e e não guardado em core_ots_proofs) pending = sorted( (p for p in state['core_ots_proofs'].values() if p['status'] == 'pending'), key=lambda p: p['created_at'])[:UPGRADE_BATCH] for rec in pending: origin = state['origin_attestations'][rec['origin_id']] artifact_hash = bytes.fromhex(origin['hash'].removeprefix('sha256:')) proof_bytes = bytes.fromhex(rec['ots_proof']) # 2a — CONTACT CALENDAR parsed0 = ots_wire.parse(proof_bytes) upgraded = _ots_upgrade(parsed0) # 2b — nada de novo: retry, ou "failed" se a submissão se # presume perdida (FAIL_AFTER). O registo de origem sobrevive; # nenhuma alegação temporal foi feita. if upgraded is None: created = datetime.strptime(rec['created_at'], '%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=timezone.utc) age = (datetime.now(timezone.utc) - created).total_seconds() if age > FAIL_AFTER_SECONDS: rec['status'] = 'failed' log(f"{rec['origin_id']}: failed (sem resposta há demasiado tempo)") continue # 2c — PARSE try: parsed = ots_wire.parse(upgraded) except ots_wire.WireError as e: log(f"{rec['origin_id']}: prova melhorada não parseia: {e}") continue btc = parsed.bitcoin_branches() if not btc: # ainda só calendário; persistir compromissos adicionais, # estado não muda (cf. §3.1 passo 4) rec['ots_proof'] = upgraded.hex() log(f"{rec['origin_id']}: ainda pendente (sem atestação Bitcoin)") continue # 2d — FETCH EACH BRANCH'S BLOCK HEADER BY HEIGHT # altura e digest do MESMO ramo; cada atestação regista a sua # altura e nenhum outro dado do bloco branches = [] for ops, att in btc: header = rpc.fetch_block_header_by_height(att.height) branches.append({'ops': ops, 'H': att.height, 'header': header}) if all(b['header'] is None for b in branches): rec['ots_proof'] = upgraded.hex() log(f"{rec['origin_id']}: ledger indisponível; fica pendente") continue # 2e — COMPARE DIGEST TO HEADER MERKLE ROOT, PER BRANCH verified = [] unevaluated = False for b in branches: digest = parsed.replay(b['ops'], artifact_hash) if digest is None or b['header'] is None: unevaluated = True continue if digest == header_merkle_root(b['header']): verified.append(b) if not verified: rec['ots_proof'] = upgraded.hex() if unevaluated: log(f"{rec['origin_id']}: ramos por avaliar; fica pendente") else: log(f"{rec['origin_id']}: ERRO — verificação do upgrade falhou") continue best = min(verified, key=lambda b: b['H']) # altura mais baixa: bound mais apertado # 2f — CHECK CONFIRMATION DEPTH (o gate; §3.1 passo 9, §6.11) k = rpc.confirmations_on(best['H']) if k is None or k < MIN_CONFIRMATIONS: rec['ots_proof'] = upgraded.hex() log(f"{rec['origin_id']}: k={k} < {MIN_CONFIRMATIONS}; não promovido") continue # 2g — PERSIST UPGRADED PROOF rec.update({ 'ots_proof': upgraded.hex(), 'status': 'anchored', 'bitcoin_block_height': best['H'], # binding normativo (§2.6.1) 'block_time': header_ntime(best['header']), # F8, informativo (RWCP) 'anchored_at': _now_iso(), # relógio local; não é o tempo da âncora }) log(f"{rec['origin_id']}: anchored @ altura {best['H']} (k={k})") if __name__ == '__main__': import sys state_path = sys.argv[1] if len(sys.argv) > 1 else 'state.json' state = load_state(state_path) upgrade_pending_proofs(state, BitcoinRPC()) save_state(state, state_path)