"""Samostatny forenzny overovac audit retazca (dokaz NEZAVISLY od beziacej appky). Re-derivuje per-inspection hash-chain od GENESIS a overuje (DESIGN-proof-core.md §4): 0) prev_hash linkovanie (GENESIS pre prvy zaznam), 1) payload integritu: SHA256(canonical(payload_json)) == payload_sha256, 2) record integritu: SHA256(canonical(record_core)) == record_hash, 3) autorstvo OFFLINE: KMS GetPublicKey (raz) + ECDSA verify cez `cryptography` — ZAMERNE NEvolame kms.Verify; pointa je nezavislost od KMS aj od appky, 4) existenciu v case: RFC 3161 TSA token na hlave (vendored _tsa.verify_timestamp). NEZAVISLOST: tento modul NEimportuje `app.*`. Kanonikalizaciu, record_core tvar, ISO formatovanie, ASN.1/CMS overenie si nesie sam (verifier/_asn1.py, verifier/_tsa.py). Dokaz tak nestoji na rovnakom kode, ktory data vyrobil (DESIGN-proof-core.md §1: sporova strana musi vediet overit aj bez pristupu k aplikacii). KRITICKE: kanonikalizacia + record_core tvar + ISO MUSIA byt BAJT-ZA-BAJT zhodne s api/app/audit.py + api/app/canonical.py, inak by neporuseny retazec vysiel ako TAMPERED. Cross-check zabezpecuju parity testy v api/tests/test_audit_chain.py (test_canonical_parity_with_verifier, test_verifier_iso_parity). Vstupy (CLI): - --export FILE : JSON export audit_log (zoznam riadkov alebo {"rows": [...]}) - --dsn DSN + --inspection-id IID : nacitaj riadky priamo z RDS (psycopg). - --kms-key-id / --region : KMS GetPublicKey na offline overenie podpisov. - --no-tsa : preskoc TSA overenie (dev exporty bez zapecatenia / cert chainu). Riadok (dict) ma kluce zhodne s audit_log stlpcami: seq, inspection_id, event_type, actor_id, actor_role, server_ts, payload_json, payload_sha256, prev_hash, record_hash, kms_key_id, signature (DER bytes/base64/hex), tsa_token (DER bytes/base64/hex). """ from __future__ import annotations import os import sys # STARY PYTHON KONCI CHYBOU PRIKAZU, KOD 2 (3. 10. 2026, OVEROVAC-1, zaverecne review). # Predvoleny `python3` na macOS je 3.9 a padal na `from datetime import UTC` vynimkou # s kodom 1, teda ako TAMPERED, hoci sa nic neoverilo. Kontrola musi stat pred importmi. if sys.version_info < (3, 12): # noqa: UP036 - zamerne: hlaska pre cloveka namiesto vynimky sys.stdout.write( f"ERROR (python): this tool needs Python 3.12 or newer, this is " f"{sys.version.split()[0]}\n" ) raise SystemExit(2) # PRIECINOK SKRIPTU AZ NA KONIEC sys.path (3. 10. 2026, OVEROVAC-1 nalez N17). # Python hlada moduly najprv vedla skriptu. Kto vybalil z PDF vsetky prilohy, mal vedla # nastroja podstrceny `structlog.py`, ktory zatienil kniznicu a podvrhol overenie: PDF # s pokazenym dokazom vyslo VERIFIED s kodom 0. Presun to pre kniznice, ktore prostredie # ma, zastavi; vlastne moduly (_asn1, _tsa, _anchor, _worm) sa najdu aj na konci. # NEUZATVARA TO VSETKO (review 3. 10. 2026, overene behom): modul, ktory prostredie NEMA # (napr. `rich.py`, ktory si structlog skusi nacitat), sa na konci sys.path najde vedla # nastroja aj tak, `__future__` sa importuje este pred tymto riadkom, a priloha s menom # `verify_chain.py` prepise nastroj samotny. Skutocna ochrana je navod: z PDF sa vybaluje # len dokaz (`pdfdetach -savefile walkcheck-proof.json`). Uplne uzavretie je v reporte # OVEROVAC-1 ako N17b. Cesty sa porovnavaju cez realpath: na macOS je /tmp odkaz na # /private/tmp a `sys.path[0]` je uz rozbaleny, `abspath(__file__)` nie. _PRIECINOK_NASTROJA = os.path.realpath(os.path.dirname(os.path.abspath(__file__))) if sys.path and os.path.realpath(sys.path[0] or os.curdir) == _PRIECINOK_NASTROJA: sys.path.append(sys.path.pop(0)) import argparse # noqa: E402 - az po uprave sys.path vyssie import base64 import binascii import hashlib import json import uuid from dataclasses import dataclass from datetime import UTC, datetime, timedelta from pathlib import Path from typing import TYPE_CHECKING, Any import structlog from pydantic import BaseModel, ConfigDict, Field, model_validator if TYPE_CHECKING: from collections.abc import Sequence log = structlog.get_logger("walkcheck.verifier") GENESIS = "0" * 64 SEAL_EVENT = "chain_sealed" # audit_log stlpce v presnom poradi (musi sediet so SELECTom v _load_rows_from_rds). _AUDIT_COLUMNS: tuple[str, ...] = ( "seq", "inspection_id", "event_type", "actor_id", "actor_role", "server_ts", "payload_json", "payload_sha256", "prev_hash", "record_hash", "kms_key_id", "signature", "tsa_token", ) # --------------------------------------------------------------------------- # # Logging (self-contained — verifier nezavisi od app.logging) # --------------------------------------------------------------------------- # def configure_logging(*, json_logs: bool = False) -> None: """Idempotentne nastav structlog pre verifikator (stderr; verdikt ide na stdout). JSON renderer pre strojove spracovanie (CI/forenzny pipeline), inak farebny konzolovy renderer. Verdikt VERIFIED/TAMPERED sa pise na stdout (nie cez log), aby bol parsovatelny aj ked logy idu na stderr. """ renderer: structlog.typing.Processor = ( structlog.processors.JSONRenderer() if json_logs else structlog.dev.ConsoleRenderer(colors=sys.stderr.isatty()) ) structlog.configure( processors=[ structlog.contextvars.merge_contextvars, structlog.processors.add_log_level, structlog.processors.TimeStamper(fmt="iso", utc=True), renderer, ], logger_factory=structlog.PrintLoggerFactory(file=sys.stderr), cache_logger_on_first_use=True, ) def vypis(riadok: str) -> None: """JEDINA cesta na stdout: jeden riadok, netlacitelne znaky escapovane (3. 10. 2026). Vo vystupe stoja aj retazce z exportu (poloha, ID, text vynimky pri casovej peciatke, ktora nie je pod hashom). Novy riadok, CR alebo ESC z nich by na obrazovku podvrhol vlastny riadok „VERIFIED ..." alebo prepisal skutocny (review OVEROVAC-1). Escapuje sa vsetko, co `str.isprintable()` nepusti. """ sys.stdout.write( "".join(z if z.isprintable() else z.encode("unicode_escape").decode("ascii") for z in riadok) + "\n" ) # --------------------------------------------------------------------------- # # Kanonikalizacia (MUSI sa zhodovat s api/app/canonical.py — bajt za bajt) # --------------------------------------------------------------------------- # def canonical_bytes(obj: Any) -> bytes: # noqa: ANN401 — JSON-serializovatelny vstup """Kanonicke UTF-8 bajty JSON (sort_keys, compact, ensure_ascii=False).""" return json.dumps( obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False ).encode("utf-8") def sha256_hex(data: bytes) -> str: """SHA256 -> hex string (64 znakov).""" return hashlib.sha256(data).hexdigest() def _iso(ts: Any) -> str: # noqa: ANN401 — datetime alebo ISO string z exportu """RFC3339 v UTC s 'Z' (zhodne s app.audit._iso).""" dt: datetime if isinstance(ts, str): dt = datetime.fromisoformat(ts) # py3.11+ zvlada 'Z' elif isinstance(ts, datetime): dt = ts else: msg = f"server_ts must be a datetime or ISO string, got {type(ts)!r}" raise TypeError(msg) if dt.tzinfo is None: dt = dt.replace(tzinfo=UTC) return dt.astimezone(UTC).isoformat().replace("+00:00", "Z") def _build_core(row: dict[str, Any]) -> dict[str, Any]: """Re-zostav record_core PRESNE ako api/app/audit.py:_build_core. Tvar (sorted keys pri kanonikalizacii): actor, event_type, inspection_id, payload_sha256, prev_hash, seq, ts. """ iid = row["inspection_id"] return { "seq": int(row["seq"]), "inspection_id": str(iid) if iid is not None else None, "event_type": row["event_type"], "actor": {"id": row.get("actor_id"), "role": row.get("actor_role")}, "ts": _iso(row["server_ts"]), "payload_sha256": row["payload_sha256"], "prev_hash": row["prev_hash"], } # --------------------------------------------------------------------------- # # Normalizacia hodnot z roznych exportov (psycopg / JSON / CSV) # --------------------------------------------------------------------------- # def _as_bytes(value: Any) -> bytes | None: # noqa: ANN401 """Normalizuj signature/token na bytes (memoryview / base64 str / hex str).""" if value is None: return None if isinstance(value, memoryview): return value.tobytes() if isinstance(value, (bytes, bytearray)): return bytes(value) if isinstance(value, str): try: return base64.b64decode(value, validate=True) except (ValueError, binascii.Error): return bytes.fromhex(value) msg = f"unsupported type for bytes: {type(value)!r}" raise TypeError(msg) def _as_payload(value: Any) -> Any: # noqa: ANN401 """payload_json moze prist ako dict (psycopg jsonb) alebo str (CSV/text export).""" return json.loads(value) if isinstance(value, str) else value # --------------------------------------------------------------------------- # # Konfiguracia (Pydantic v2) — argumenty verifikacie # --------------------------------------------------------------------------- # class VerifierConfig(BaseModel): """Vstupna konfiguracia behu verifikatora (z CLI / env).""" model_config = ConfigDict(frozen=True, extra="forbid") export: Path | None = Field(default=None, description="JSON export audit_log") dsn: str | None = Field(default=None, description="psycopg DSN for RDS") inspection_id: str | None = Field(default=None, description="inspection UUID (with --dsn)") kms_key_id: str | None = Field(default=None, description="KMS key id for GetPublicKey") public_key: Path | None = Field(default=None, description="DER/PEM public key (offline)") region: str = Field(default="eu-central-1", description="AWS region for KMS GetPublicKey") tsa_trust: Path | None = Field(default=None, description="PEM bundle of pinned TSA certs") tsa_clock_skew_seconds: int = Field(default=300, description="clock tolerance for genTime") verify_tsa: bool = Field(default=True, description="verify the RFC 3161 TSA token on the head") anchor_export: Path | None = Field( default=None, description="JSON export of epoch_anchor (inclusion)" ) require_anchor: bool = Field( default=False, description="require inclusion in the global anchor" ) # FINAL vrstva — externa WORM+TSA atestacia epoch rootu. `attestation_path` = # adresar/subor s WORM objektmi (anchor-attestations/.json), alebo `worm_bucket` # = S3 bucket. require_attestation cross-checkne zivu kotvu proti NAJNOVSEJ atestacii # (chyti tail-delete vlastneho anchor riadku) a anchored hlavu bez atestacie hlasi PENDING. attestation_path: Path | None = Field( default=None, description="directory/file of WORM attestations of the epoch root" ) worm_bucket: str | None = Field( default=None, description="S3 WORM bucket with attestations (alt. to --attestation-path)" ) require_attestation: bool = Field( default=False, description="cross-check the live anchor against the WORM attestation" ) # FIX #3 — FAIL-CLOSED: standardne verifikator VYZADUJE doveryhodny kluc. # `--allow-unsigned` je EXPLICITNY dev escape hatch (vrati UNVERIFIED_DEV, nie VERIFIED). allow_unsigned: bool = Field( default=False, description="dev: do not require signatures (prints UNVERIFIED_DEV, not VERIFIED)" ) json_logs: bool = Field(default=False, description="structlog JSON renderer") @model_validator(mode="after") def _check_source(self) -> VerifierConfig: if self.export is None and self.dsn is None: msg = "give --export FILE or --dsn DSN" raise ValueError(msg) if self.export is not None and self.dsn is not None: msg = "give exactly ONE source: --export or --dsn" raise ValueError(msg) if self.dsn is not None and not self.inspection_id: msg = "--dsn requires --inspection-id" raise ValueError(msg) # FIX #3 — FAIL-CLOSED: bez --allow-unsigned MUSI byt zdroj doveryhodneho # kluca (KMS key id alebo lokalny public key PEM/DER). Inak by sa podpisy # ticho preskocili (povodna diera). Vyzaduj explicitne rozhodnutie. if not self.allow_unsigned and not (self.kms_key_id or self.public_key): msg = ( "FAIL-CLOSED: give --kms-key-id OR --public-key (a trusted key " "to verify signatures with), or --allow-unsigned explicitly for dev " "(prints UNVERIFIED_DEV, not VERIFIED)" ) raise ValueError(msg) # VRSTVA BEZ DAT JE CHYBA PRIKAZU, KOD 2 (3. 10. 2026, OVEROVAC-1 nalez N3). # Stranka radila pridat --require-anchor, PDF kotvu nenesie a nad prazdnou kotvou # vysiel zdravy zaznam TAMPERED (global_anchor:anchor_empty); atestacia bez kotvy # tiez (_worm: anchor_truncated_vs_attestation) a --require-attestation bez dat # davalo PENDING. Nastroj, ktory nemal co overit, nesmie dat verdikt (Brain). ma_kotvu = self.anchor_export is not None or self.dsn is not None ma_atestaciu = self.attestation_path is not None or bool(self.worm_bucket) if self.require_anchor and not ma_kotvu: msg = ( "--require-anchor needs the global anchor: give --anchor-export FILE " "(or --dsn). The PDF does not carry it" ) raise ValueError(msg) if self.require_attestation and not (ma_kotvu and ma_atestaciu): msg = ( "--require-attestation needs the global anchor (--anchor-export FILE or " "--dsn) and the WORM attestation (--attestation-path or --worm-bucket). " "The PDF carries neither" ) raise ValueError(msg) return self @property def require_signatures(self) -> bool: """Trust-gate: podpisy sa VYNUCUJU vzdy, okrem explicitneho --allow-unsigned.""" return not self.allow_unsigned # --------------------------------------------------------------------------- # # Vysledok # --------------------------------------------------------------------------- # @dataclass(frozen=True) class VerifyResult: """Vysledok overenia retazca jednej inspekcie. `dev_unverified` (FIX #1/#3): bezalo sa s --allow-unsigned (bez trust-gate), takze ok=True znamena len internu hash-konzistenciu, NIE kryptograficky dokaz. main() taky vysledok mapuje na UNVERIFIED_DEV, NIKDY na VERIFIED. `pending` (FINAL vrstva): retazec je zapecateny/ukotveny a interne v poriadku, ALE ziadna WORM atestacia este nekryje jeho anchor seq (atestacne okno). Taky dokaz NIE je finalizovany (un-attested anchor riadok je stale zmazatelny) -> main() ho mapuje na PENDING, NIKDY na VERIFIED. `seal_hash` (3. 10. 2026, OVEROVAC-1 nalez N5): record_hash pecate, ktoru overenie naozaj overilo. Verdikt nemeni; main() ho vypise, aby ho clovek porovnal s PDF. """ ok: bool broken_seq: int | None = None reason: str | None = None sealed: bool = False dev_unverified: bool = False pending: bool = False seal_hash: str | None = None #: Verdikty, ktore verifier pise na stdout. Poradie = poradie prednosti vo #: `verdict()` a je 1:1 s `app.audit.verify_status` (parita v #: api/tests/test_verify_status.py). STATUS_TAMPERED = "TAMPERED" STATUS_PENDING = "PENDING" STATUS_UNSEALED = "UNSEALED" STATUS_UNVERIFIED_DEV = "UNVERIFIED_DEV" STATUS_VERIFIED = "VERIFIED" def verdict(result: VerifyResult) -> str: """Mapovanie VerifyResult -> verdikt pre cloveka. Rovnake ako v API. Prednost: TAMPERED > PENDING > UNSEALED > UNVERIFIED_DEV > VERIFIED. UNSEALED (22.08.2026): do tohto dna verifier nad exportom bez pecate pisal "VERIFIED (sealed=False)", kym PDF toho isteho zaznamu pisalo UNSEALED a navod v PDF tvrdil, ze sa verdikty musia zhodovat. Konzistentny retazec bez chain_sealed nie je dokaz, je to rozpracovany zaznam; verifier to musi povedat rovnakym slovom ako API a PDF, inak sa klientovi predklada "overene" na nieco, co nikto nezapecatil. """ if not result.ok: return STATUS_TAMPERED if result.pending: return STATUS_PENDING if not result.sealed: return STATUS_UNSEALED if result.dev_unverified: return STATUS_UNVERIFIED_DEV return STATUS_VERIFIED # --------------------------------------------------------------------------- # # OFFLINE KMS podpis verify (cryptography; NIE kms.Verify) # --------------------------------------------------------------------------- # def _offline_kms_verify( record_hash: str, signature: bytes | None, public_key_der: bytes | None ) -> bool: """Offline ECDSA verify cez cryptography (public key z KMS GetPublicKey, DER). KMS podpisuje s MessageType=DIGEST nad RAW 32B z record_hashu => message je `bytes.fromhex(record_hash)` a algoritmus ECDSA_SHA_256. ZAMERNE nevolame kms.Verify — overujeme lokalne verejnym klucom, takze dokaz nestoji na KMS. """ if not signature or not public_key_der: return False from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import ec, utils from cryptography.hazmat.primitives.serialization import load_der_public_key key = load_der_public_key(public_key_der) if not isinstance(key, ec.EllipticCurvePublicKey): return False # KMS s MessageType=DIGEST podpisuje 32 B TAK, AKO SU — nehashuje ich znova. # `ec.ECDSA(hashes.SHA256())` by digest zahashovalo druhykrat a kazdy REALNY # podpis z KMS by vysiel ako neplatny (22.08.2026: prod export f3a02eec # hlasil TAMPERED at seq=283, kym zive API VERIFIED). Prehashed = "spravu # uz niekto zahashoval, over len podpis nad nou". try: key.verify( signature, bytes.fromhex(record_hash), ec.ECDSA(utils.Prehashed(hashes.SHA256())), ) except InvalidSignature: return False return True # --------------------------------------------------------------------------- # # Jadro overenia (poradie krokov 1:1 s app.audit.verify) # --------------------------------------------------------------------------- # _FINAL_INSPECTION_STATUSES = frozenset({"finished", "scored", "reported"}) # Udalosti anonymizacie (23.08.2026) — parita s app/audit.py. INGEST_EVENT = "video_ingested" ANONYMIZED_EVENT = "video_anonymized" SCORED_EVENT = "inspection_scored" # Zalozenie zaznamu; od 23.08.2026 nesie aj `ulozisko_rezim` (0039). CREATED_EVENT = "inspection_created" REZIM_PILOT = "pilot" REZIM_OSTRY = "ostry" def ulozisko_rezim_z_retaze( *, created: dict[str, Any] | None, ingested: dict[str, Any] | None, ) -> str: """V akom rezime uloziska zaznam vznikol — podla RETAZE, nie podla stlpca. DOSLOVNA KOPIA `app.audit.ulozisko_rezim_z_retaze` (verifier nesmie zavisiet od app/; parita je kryta testom v api/tests). Existuje preto, aby si tretia strana vedela overit vetu o ulozisku bez pristupu k nasej DB: stlpec `inspections.ulozisko_rezim` vie prepisat ktokolvek s DB write, payload pod hashom a KMS podpisom nie. PRAVIDLO: staci JEDNA udalost, ktora hovori `pilot`, a zaznam sa za nezvratne ulozeny NEVYHLASI. Udalosti su dve ('inspection_created' hovori, s cim zaznam vznikol, 'video_ingested' kam pristali bajty) a keby si protirecili, tvrdit nezvratnost by znamenalo vybrat si tu prijemnejsiu polovicu rozporu. Nezvratnost sa tvrdi len vtedy, ked ju nespochybnuje nic. Ziadna z udalosti pole nema = zaznam spred 0039. Tie su naozaj zamknute (vedro malo default retenciu COMPLIANCE), takze `ostry` je pravda. VERDIKT TO NEMENI: retazec pilotneho zaznamu je rovnako platny ako ostreho. """ hodnoty = [ (p or {}).get("ulozisko_rezim") for p in (created, ingested) if p is not None ] if any(h == REZIM_PILOT for h in hodnoty): return REZIM_PILOT return REZIM_OSTRY def ulozisko_rezim_z_riadkov(rows: Sequence[dict[str, Any]]) -> str: """Rezim uloziska z nacitanych riadkov retazca (prve vyskyty udalosti).""" payloady: dict[str, dict[str, Any] | None] = {CREATED_EVENT: None, INGEST_EVENT: None} for row in rows: et = str(row.get("event_type")) if et in payloady and payloady[et] is None: payload = _as_payload(row.get("payload_json")) if isinstance(payload, dict): payloady[et] = payload return ulozisko_rezim_z_retaze(created=payloady[CREATED_EVENT], ingested=payloady[INGEST_EVENT]) # --------------------------------------------------------------------------- # # Poloha zariadenia v dokaze (ADR-POLOHA-ZARIADENIA-V-DOKAZE-2026-09-16, dodatok §10) # --------------------------------------------------------------------------- # # Blok nizsie je DOSLOVNA KOPIA bloku v api/app/poloha.py (verifier nesmie # zavisiet od app/). Zhodu textu aj spravania strazi api/tests/test_poloha_parita.py. # Overovac z polohy len VYPISUJE fakty; verdikt nemeni nikdy (ADR §5) a pri # TAMPERED riadok nevypise vobec, lebo porusenej retazi sa neveri ani v tomto. # --- ZACIATOK DOSLOVNEJ KOPIE (poloha zariadenia, verzia 1) --- # Tento blok je doslovne aj vo verifier/verify_chain.py. Kopia sa nemeni po # kusoch: zmena prahu alebo okna je nova VERZIA, nie oprava (ADR §5). #: Verzia tvaru a prahov v retazi. Zaznam nesie cislo verzie, podla ktorej sa #: klasifikuje; neznama verzia dava triedy `nezname`, nikdy pad. VERZIA = 1 #: Do tohto prahu (vratane) je bod `poloha`, nad nim `priblizna` (ADR §5). PRAH_POLOHA_M = 100 #: Nad tymto prahom sa suradnice nezapisu a bod sa pocita ako neprisiel. PRAH_ZAPIS_M = 2000 #: Okno natacania: `recorded_started_at - OKNO_S` az `recorded_finished_at + OKNO_S`. OKNO_S = 2 #: Fix viac nez tolko sekund v buducnosti voci casu servera sa odmietne. BUDUCNOST_S = 300 #: Dlzka soli zavazku v bajtoch. Bez soli sa bunka troch desatinnych miest #: prepocita zo sto kandidatov (ADR §4). DLZKA_SOLI = 32 KEDY = ("zaciatok", "koniec") PLATFORMY = ("ios", "android", "nezname") PRESNOSTI_POVOLENE = ("presna", "priblizna", "nezname") #: Stav polohy zaznamu pre appku, PDF, stranku aj overovac (plan §3). STAVY = ("pred_zavedenim", "nezaznamenavala_sa", "bez_bodu", "zaznamenana") TRIEDY = ("poloha", "priblizna", "nezname") V_OKNE = ("pocas", "mimo", "nezname") #: Zdroj vety pri `nezaznamenavala_sa`; vycvik ma prednost pred kamerou. ZDROJE = ("vycvik", "kamera") #: Stav otvorenia (presnej hodnoty) pre clena organizacie (dodatok R3). OTVORENIA = ("zhodne", "chyba", "nesedi") def zavazok(kedy: str, lat_e4: int, lon_e4: int, sol_hex: str) -> str: """Zavazok presnej polohy: sha256 kanonickeho JSON `{kedy, lat_e4, lon_e4, sol}`. Kanonicky tvar je ten isty ako v `app/canonical.py` (sort_keys, bez medzier, UTF-8). Sol je hex 32 bajtov. Kto ma otvorenie, prepocita zavazok a porovna ho s tym, co stoji v zapecatenej retazi. """ obsah = {"kedy": kedy, "lat_e4": lat_e4, "lon_e4": lon_e4, "sol": sol_hex} return hashlib.sha256( json.dumps(obsah, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8"), ).hexdigest() def _cele_cislo(hodnota: Any) -> int | None: # noqa: ANN401 - hodnota z retaze """Cele cislo, ale nie bool (`True == 1` v Pythone).""" if isinstance(hodnota, bool) or not isinstance(hodnota, int): return None return hodnota def _cas(hodnota: Any) -> datetime | None: # noqa: ANN401 - hodnota z retaze """ISO cas s casovou zonou -> datetime v UTC. Bez zony alebo nezmysel -> None.""" if not isinstance(hodnota, str) or len(hodnota) > 64: # noqa: PLR2004 return None try: cas = datetime.fromisoformat(hodnota) except ValueError: return None if cas.tzinfo is None or cas.utcoffset() is None: return None return cas.astimezone(UTC) def _okno(ingested: dict[str, Any] | None) -> tuple[datetime, datetime] | None: """Okno natacania z PRVEJ udalosti `video_ingested` (hodiny telefonu). Doplnene casy admina (`inspection_recorded_times_filled`) okno nemenia: volajuci sem podava len payload `video_ingested`. Chybajuci alebo obrateny cas dava None, teda `v_okne: nezname`. """ if not isinstance(ingested, dict): return None od = _cas(ingested.get("recorded_started_at")) do = _cas(ingested.get("recorded_finished_at")) if od is None or do is None or do < od: return None return od - timedelta(seconds=OKNO_S), do + timedelta(seconds=OKNO_S) def _trieda(presnost_m: int | None, *, znama: bool) -> str: if not znama or presnost_m is None or presnost_m <= 0: return "nezname" if presnost_m <= PRAH_POLOHA_M: return "poloha" if presnost_m <= PRAH_ZAPIS_M: return "priblizna" return "nezname" def _v_okne(fix_at: Any, okno: tuple[datetime, datetime] | None) -> str: # noqa: ANN401 fix = _cas(fix_at) if okno is None or fix is None: return "nezname" return "pocas" if okno[0] <= fix <= okno[1] else "mimo" def _bod_z_retaze(bod: dict[str, Any], *, znama: bool, okno: tuple[datetime, datetime] | None) -> dict[str, Any]: """Fakty bodu z retaze a jeho triedy. Chybajuce alebo cudzie pole je None.""" presnost_m = _cele_cislo(bod.get("presnost_m")) zavazok_bodu = bod.get("zavazok") return { "kedy": bod.get("kedy") if isinstance(bod.get("kedy"), str) else None, "lat_e3": _cele_cislo(bod.get("lat_e3")), "lon_e3": _cele_cislo(bod.get("lon_e3")), "presnost_m": presnost_m, "fix_at": bod.get("fix_at") if isinstance(bod.get("fix_at"), str) else None, "mocked": bod.get("mocked") if isinstance(bod.get("mocked"), bool) else None, "zavazok": zavazok_bodu if isinstance(zavazok_bodu, str) else None, "trieda": _trieda(presnost_m, znama=znama), "v_okne": _v_okne(bod.get("fix_at"), okno) if znama else "nezname", } def poloha_z_retaze( *, created: dict[str, Any] | None, ingested: dict[str, Any] | None, ) -> dict[str, Any]: """Poloha zaznamu podla RETAZE: fakty a triedy podla verzie, bez pecate. `created` je payload prvej `inspection_created`, `ingested` payload prvej `video_ingested`. Nikdy nepadne: kluc chyba (`pred_zavedenim`), `null` alebo iny tvar (`nezaznamenavala_sa`), prazdne body (`bez_bodu`), body (`zaznamenana`). Neznama verzia dava triedy aj okno `nezname`. Presnu hodnotu (e4) tato funkcia NEPOZNA a poznat nema: otvorenie lezi mimo retaze a cita ho len server pre clena organizacie (dodatok R3). """ vysledok: dict[str, Any] = { "stav": "pred_zavedenim", "verzia": None, "platforma": None, "presnost_povolena": None, "body": [], } if not isinstance(created, dict) or "poloha_zariadenia" not in created: return vysledok hodnota = created["poloha_zariadenia"] if not isinstance(hodnota, dict): vysledok["stav"] = "nezaznamenavala_sa" return vysledok verzia = _cele_cislo(hodnota.get("verzia")) znama = verzia == VERZIA okno = _okno(ingested) if znama else None surove_body = hodnota.get("body") body = [ _bod_z_retaze(bod, znama=znama, okno=okno) for bod in (surove_body if isinstance(surove_body, list) else []) if isinstance(bod, dict) ] platforma = hodnota.get("platforma") presnost_povolena = hodnota.get("presnost_povolena") vysledok.update({ "stav": "zaznamenana" if body else "bez_bodu", "verzia": verzia, "platforma": platforma if isinstance(platforma, str) else None, "presnost_povolena": (presnost_povolena if isinstance(presnost_povolena, str) else None), "body": body, }) return vysledok def vyber_zobrazeny_bod(body: list[dict[str, Any]]) -> str | None: """Ktory bod ukazat, JEDNA funkcia pre appku, PDF aj stranku (plan §3). Poradie: zaciatok v okne, koniec v okne, zaciatok, koniec. Bez bodov None. """ for kedy, len_v_okne in (("zaciatok", True), ("koniec", True), ("zaciatok", False), ("koniec", False)): for bod in body: if bod.get("kedy") == kedy and (not len_v_okne or bod.get("v_okne") == "pocas"): return kedy return None # --- KONIEC DOSLOVNEJ KOPIE --- #: Predpona riadku polohy na stdout (plan §3, kontrakt overovaca). _POLOHA_PREDPONA = "LOCATION (reported by device): " _POLOHA_KEDY_EN = {"zaciatok": "start", "koniec": "end"} _POLOHA_OKNO_EN = { "pocas": "within recording window", "mimo": "outside recording window", "nezname": "recording window unknown", } def poloha_z_riadkov(rows: Sequence[dict[str, Any]]) -> dict[str, Any]: """Poloha z nacitanych riadkov retazca: prva `inspection_created` a prva `video_ingested`. Udalost admina `inspection_recorded_times_filled` sa necita vobec: okno je z hodin telefonu v case prijatia videa, nie z neskorsieho doplnenia. """ payloady: dict[str, dict[str, Any] | None] = {CREATED_EVENT: None, INGEST_EVENT: None} for row in rows: et = str(row.get("event_type")) if et in payloady and payloady[et] is None: payload = _as_payload(row.get("payload_json")) if isinstance(payload, dict): payloady[et] = payload return poloha_z_retaze(created=payloady[CREATED_EVENT], ingested=payloady[INGEST_EVENT]) def _stupne_z_e3(hodnota: int | None, kladne: str, zaporne: str) -> str | None: """49151 -> `49.151 N`. Z celeho cisla, bez floatu, aby sa nic nezaokruhlilo znova.""" if hodnota is None: return None strana = kladne if hodnota >= 0 else zaporne cele, tisiciny = divmod(abs(hodnota), 1000) return f"{cele}.{tisiciny:03d} {strana}" def riadky_polohy(rows: Sequence[dict[str, Any]]) -> list[str]: """Riadky LOCATION pre stdout. SURADNICE LEN SEM, NIKDY DO `log.*`. Logy overovaca idu na stderr a do CI; suradnice su osobny udaj a ADR §6 ich v logoch zakazuje. Stdout je vystup pre cloveka, ktory dokaz overuje. """ p = poloha_z_riadkov(rows) if p["stav"] == "pred_zavedenim": return [_POLOHA_PREDPONA + "none in chain (record predates device location)"] if p["stav"] == "nezaznamenavala_sa": return [_POLOHA_PREDPONA + "not recorded for this record"] if p["stav"] == "bez_bodu": return [_POLOHA_PREDPONA + "attempted, no fix recorded"] riadky: list[str] = [] for bod in p["body"]: lat = _stupne_z_e3(bod["lat_e3"], "N", "S") lon = _stupne_z_e3(bod["lon_e3"], "E", "W") miesto = f"{lat} {lon}" if lat and lon else "coordinates unknown" presnost = (f"\u00b1{bod['presnost_m']} m" if bod["presnost_m"] is not None else "accuracy unknown") mocked = ("true" if bod["mocked"] is True else "false" if bod["mocked"] is False else "unknown") riadok = ( f"{_POLOHA_KEDY_EN.get(bod['kedy'] or '', bod['kedy'] or 'point')} {miesto} " f"{presnost} at {bod['fix_at'] or 'unknown time'}, " f"{_POLOHA_OKNO_EN[bod['v_okne']]}, mocked={mocked}, " f"commitment={bod['zavazok'] or 'none'}" ) if p["verzia"] != VERZIA: riadok += f", version={p['verzia']} unknown to this verifier" riadky.append(_POLOHA_PREDPONA + riadok) return riadky def check_anonymization( # noqa: PLR0911 — kazdy return je jeden dovod TAMPERED *, ingested: dict[str, Any] | None, anonymized: dict[str, Any] | None, anonymized_seq: int | None, scored: dict[str, Any] | None, seal_seq: int | None, ) -> str | None: """Pravidla anonymizacie nad payloadmi (23.08.2026). None = v poriadku. DOSLOVNA KOPIA `app.audit.check_anonymization` (verifier nesmie zavisiet od app/; parita je kryta testom v api/tests). Pravidla: - 'video_ingested' s `intake: true` a pecat BEZ 'video_anonymized' -> `anonymization_missing` (surove video pod pecatou), - 'video_anonymized' musi byt PRED pecatou -> `anonymization_after_seal`, - `source_sha256` v anonymizacii == hash v 'video_ingested' (`source_sha256`, inak `video_sha256`) -> `anonymization_source_mismatch`, - `rendered_sha256` == `video_sha256` v 'inspection_scored' -> `anonymization_render_mismatch`. Zaznam bez `intake: true` a bez 'video_anonymized' je stary zaznam: ziadne pravidlo sa nan nevztahuje, ostava VERIFIED (vznikol pred zavedenim). """ via_intake = bool(ingested) and ingested.get("intake") is True if anonymized is None: if via_intake and seal_seq is not None: return "anonymization_missing" return None if seal_seq is not None and (anonymized_seq is None or anonymized_seq > seal_seq): return "anonymization_after_seal" if ingested is not None: zdroj = ingested.get("source_sha256") or ingested.get("video_sha256") if not zdroj or anonymized.get("source_sha256") != zdroj: return "anonymization_source_mismatch" rendered = anonymized.get("rendered_sha256") if not rendered: return "anonymization_render_mismatch" if scored is not None and scored.get("video_sha256") != rendered: return "anonymization_render_mismatch" return None def verify( # noqa: C901, PLR0911, PLR0912, PLR0913, PLR0915 — guard clauses + parita s app.audit.verify rows: Sequence[dict[str, Any]], *, public_key_der: bytes | None = None, verify_tsa: bool = True, require_signatures: bool = True, tsa_trust_pem: str = "", tsa_clock_skew_seconds: int = 300, inspection_status: str | None = None, anchor_rows: Sequence[dict[str, Any]] | None = None, require_anchor: bool = False, attestation: Any = None, # noqa: ANN401 — _worm.Attestation | None (vendored) require_attestation: bool = False, ) -> VerifyResult: """Over per-inspection retazec (rows ZORADENE podla seq ASC, JEDNA inspekcia). `public_key_der` = doveryhodny verejny kluc (KMS GetPublicKey DER / PEM->DER). `require_signatures` (FIX #1/#3): ak True, KAZDY zaznam MUSI mat platny podpis proti tomuto klucu — chybajuci kluc/podpis -> TAMPERED. Ak False (dev), vystup je dev_unverified=True (NIKDY plain VERIFIED). `anchor_rows` + `require_anchor` (FIX #4 round-3 + FIX #5): IMMUTABLE epoch_anchor je zdroj pravdy. Ak kotva referencuje inspekciu (expected_sealed_head), retazec MUSI koncit platnym chain_sealed s record_hash == ukotvena hlava + platnou inkluziou — bez ohladu na inspection_status. Inkluzia sa overuje vzdy ked is_sealed ALEBO kotva inspekciu referencuje (NIE gatovane len za is_sealed) -> zatvara global-anchor omission. `inspection_status` (FIX #4): LEN doplnkovy signal — finalny status bez sealu sa stale flagne, ale NIKDY nevie ZNIZIT/preskocit poziadavku, ktoru ustanovi kotva. Standalone verifikator NEDOVERUJE RDS inspections.status pre seal-required rozhodnutie. `attestation` + `require_attestation` (FINAL vrstva): NAJNOVSIA WORM atestacia epoch rootu (autorita = immutable WORM objekt). Cross-checkne ZIVU kotvu (anchor_rows) proti atestovanemu seq/rootu — menej riadkov / iny root -> TAMPERED (chyti tail-delete vlastneho anchor riadku DB superuserom). Anchored hlava bez kryjucej atestacie -> PENDING (sealed=False, pending=True), NIKDY VERIFIED (un-attested riadok je stale zmazatelny). """ if not rows: return VerifyResult(ok=False, reason="empty_chain") # FIX #3 — FAIL-CLOSED: vyzadujeme podpisy, ale nemame doveryhodny kluc -> nevieme # nic overit, MUSIME odmietnut (nie ticho preskocit). if require_signatures and public_key_der is None: return VerifyResult(ok=False, reason="no_trusted_signing_key") prev = GENESIS head_hash: str | None = None head_token: bytes | None = None head_event: str | None = None # PECAT SI PAMATAME TAM, KDE JE — parita s app.audit.verify. Nemusi byt # posledna: po nej legitimne pribudaju udalosti (archivacia, oprava metadat). seal_hash: str | None = None seal_token: bytes | None = None seal_seq: int | None = None # Anonymizacia (23.08.2026): payloady troch udalosti, ktore sa musia zhodovat. ingested: dict[str, Any] | None = None anonymized: dict[str, Any] | None = None anonymized_seq: int | None = None scored: dict[str, Any] | None = None for row in rows: seq = int(row["seq"]) # 0) linkovanie if row["prev_hash"] != prev: return VerifyResult(ok=False, broken_seq=seq, reason="prev_hash_mismatch") # 1) payload integrita — re-derivuj z ulozeneho payload_json payload = _as_payload(row["payload_json"]) if sha256_hex(canonical_bytes(payload)) != row["payload_sha256"]: return VerifyResult(ok=False, broken_seq=seq, reason="payload_sha256_mismatch") # 2) record integrita — prepocitaj core if sha256_hex(canonical_bytes(_build_core(row))) != row["record_hash"]: return VerifyResult(ok=False, broken_seq=seq, reason="record_hash_mismatch") # 3) autorstvo — OFFLINE KMS podpis (TRUST-GATE; FIX #1) signature = _as_bytes(row.get("signature")) if require_signatures: # Doveryhodny kluc je public_key_der (z configu, NIE z riadku). KAZDY # zaznam MUSI mat platny podpis — vynulovanie stlpcov utocnikom = padne sem. if not _offline_kms_verify(row["record_hash"], signature, public_key_der): return VerifyResult(ok=False, broken_seq=seq, reason="kms_signature_invalid") elif ( public_key_der is not None and (signature or row.get("kms_key_id")) and not _offline_kms_verify(row["record_hash"], signature, public_key_der) ): return VerifyResult(ok=False, broken_seq=seq, reason="kms_signature_invalid") prev = row["record_hash"] head_hash = row["record_hash"] head_token = _as_bytes(row.get("tsa_token")) head_event = row["event_type"] if row["event_type"] == SEAL_EVENT: seal_hash = row["record_hash"] seal_token = _as_bytes(row.get("tsa_token")) seal_seq = seq elif row["event_type"] == INGEST_EVENT and ingested is None: ingested = payload if isinstance(payload, dict) else {} elif row["event_type"] == ANONYMIZED_EVENT and anonymized is None: anonymized = payload if isinstance(payload, dict) else {} anonymized_seq = seq elif row["event_type"] == SCORED_EVENT and scored is None: scored = payload if isinstance(payload, dict) else {} # PECAT NEMUSI BYT HLAVA. Do 19.08.2026 tu stalo `head_event == SEAL_EVENT`, # takze akykolvek zapis po peceti (archivacia je najbeznejsi) spravil # is_sealed=False a pri finalnom statuse to padlo na `seal_required_for_final`, # teda TAMPERED na zdravom dokaze. app/ tu opravu dostalo 18.08., verifier/ # nie — a prave verifier je to, co dostava klient ako NEZAVISLY dokaz. is_sealed = seal_hash is not None head_seq = int(rows[-1]["seq"]) inspection_id = str(rows[0]["inspection_id"]) # 3b) anonymizacia (23.08.2026) — rovnake pravidlo ako app.audit.verify: # surovy zaznam z prijmoveho vedra sa pod pecatou objavit NESMIE a retaz # surovy hash -> render hash musi drzat pokope. anon_dovod = check_anonymization(ingested=ingested, anonymized=anonymized, anonymized_seq=anonymized_seq, scored=scored, seal_seq=seal_seq) if anon_dovod is not None: return VerifyResult(ok=False, broken_seq=anonymized_seq or seal_seq or head_seq, reason=anon_dovod, sealed=is_sealed) # Hash, ktory bol UKOTVENY: pecat, nie hlava. Jedno miesto pre vsetky dolne # vrstvy (TSA, inkluzia, WORM atestacia), aby sa nemohli rozist — presne tak # vznikla chyba opravovana 19.08. kotveny_hash = seal_hash or head_hash # FIX #4 (round-3): seal-required + anchor-required sa odvodzuju z IMMUTABLE kotvy # (anchor_rows = epoch_anchor), NIE z mutovatelneho RDS inspections.status. Standalone # verifikator NESMIE doverovat inspections.status pre rozhodnutie "musi byt zapecatena" # — utocnik s DB write ho prepne. Ak kotva referencuje inspekciu, retazec MUSI ostat # zapecateny s presne ukotvenou hlavou + platnou inkluziou, bez ohladu na status/is_sealed. import _anchor # vendored; offline parita s app.anchor anchored_head: str | None = ( _anchor.expected_sealed_head(anchor_rows or [], inspection_id) if require_anchor else None ) # INVARIANT (anchored => must stay sealed+included): kotva viaze hlavu -> retazec # MUSI koncit platnym chain_sealed s record_hash == ukotvena hlava (parita s app.audit). if anchored_head is not None: if not is_sealed: return VerifyResult( ok=False, broken_seq=head_seq, reason="seal_required_by_anchor", ) # Porovnava sa hash PECATE, nie hlavy: kotva ukotvila prave ju. # Parita s app/audit.py — tam to bolo opravene, tu nie, a na archivovanom # zazname to davalo `anchor_head_mismatch` na uplne zdravom dokaze. if seal_hash != anchored_head: return VerifyResult( ok=False, broken_seq=head_seq, reason="anchor_head_mismatch", sealed=True, ) # FIX #4 (doplnkovy signal): finalny status BEZ sealu sa stale flagne, ale NIKDY # nevie ZNIZIT poziadavku z kotvy vyssie (status sa berie len ako dodatocny signal). if inspection_status in _FINAL_INSPECTION_STATUSES and not is_sealed: return VerifyResult( ok=False, broken_seq=head_seq, reason=f"seal_required_for_final:{inspection_status}", ) # 4) existencia v case — TSA token na hlave (ak je retazec zapecateny). if verify_tsa and is_sealed and seal_hash is not None: from _tsa import verify_timestamp # vendored; ziadna app.* zavislost result = verify_timestamp( # PECAT, nie hlava: TSA token je stampnuty na seal zaznam # (app.audit.verify robi to iste so seal_record_hash/seal_token). seal_hash, seal_token, trust_pem=tsa_trust_pem, clock_skew_seconds=tsa_clock_skew_seconds, ) if not result.ok: return VerifyResult( ok=False, broken_seq=head_seq, reason=f"tsa_anchor:{result.reason}", sealed=True, ) # 5) globalna kotva — inkluzia zapecatenej hlavy v epoch roote (FIX #5). NIE # gatovane za is_sealed: spusta sa ked is_sealed ALEBO kotva inspekciu referencuje # (anchored_head) — parita s app.audit. Tym sa global-anchor omission zatvara z # oboch strán (zmazany anchor riadok pri sealed chain / zmazany seal pri anchored). if ( require_anchor and (is_sealed or anchored_head is not None) and kotveny_hash is not None ): incl = _verify_anchor_inclusion( anchor_rows or [], inspection_id, kotveny_hash, public_key_der=public_key_der, require_signatures=require_signatures, verify_tsa=verify_tsa, tsa_trust_pem=tsa_trust_pem, tsa_clock_skew_seconds=tsa_clock_skew_seconds, ) if not incl.ok: return VerifyResult( ok=False, broken_seq=head_seq, reason=f"global_anchor:{incl.reason}", sealed=is_sealed, ) # 7) FINAL vrstva — WORM+TSA ATESTACIA epoch rootu (parita s app.audit.verify krok 7). # Cross-checkne ZIVU epoch_anchor (anchor_rows) proti NAJNOVSEJ WORM atestacii: menej # riadkov / iny root -> TAMPERED (chyti tail-delete vlastneho anchor riadku superuserom). # Anchored hlava bez kryjucej atestacie -> PENDING (sealed=False), NIKDY VERIFIED. if ( require_attestation and (is_sealed or anchored_head is not None) and kotveny_hash is not None ): check = _verify_attestation( anchor_rows or [], attestation, public_key_der=public_key_der, require_signatures=require_signatures, verify_tsa=verify_tsa, tsa_trust_pem=tsa_trust_pem, tsa_clock_skew_seconds=tsa_clock_skew_seconds, ) if not check.ok: return VerifyResult( ok=False, broken_seq=head_seq, reason=f"worm_attestation:{check.reason}", sealed=True, ) import _worm # vendored entry_seq = _worm.anchor_entry_seq(anchor_rows or [], inspection_id, kotveny_hash) attested_covered = ( check.attested and entry_seq is not None and entry_seq <= check.covers_seq ) if not attested_covered: return VerifyResult( ok=True, sealed=False, pending=True, dev_unverified=not require_signatures, seal_hash=seal_hash, ) return VerifyResult(ok=True, sealed=is_sealed, dev_unverified=not require_signatures, seal_hash=seal_hash) def _verify_anchor_inclusion( # noqa: PLR0913 — offline most do _anchor.verify_inclusion anchor_rows: Sequence[dict[str, Any]], inspection_id: str, sealed_head_hash: str, *, public_key_der: bytes | None, require_signatures: bool, verify_tsa: bool, tsa_trust_pem: str, tsa_clock_skew_seconds: int, ) -> Any: # noqa: ANN401 — _anchor.InclusionResult (vendored) """Most na vendored _anchor.verify_inclusion s offline KMS + TSA callbackmi.""" import _anchor from _tsa import verify_timestamp def _kms_cb(anchor_hash: str, signature: bytes | None) -> bool: return _offline_kms_verify(anchor_hash, _as_bytes(signature), public_key_der) def _tsa_cb(anchor_hash: str, token: bytes | None) -> Any: # noqa: ANN401 return verify_timestamp( anchor_hash, _as_bytes(token), trust_pem=tsa_trust_pem, clock_skew_seconds=tsa_clock_skew_seconds, ) return _anchor.verify_inclusion( anchor_rows, inspection_id, sealed_head_hash, require_signatures=require_signatures, offline_kms_verify=_kms_cb, verify_tsa=verify_tsa, tsa_verify=_tsa_cb, ) def _verify_attestation( # noqa: PLR0913 — offline most do _worm.verify_attestation_against_live anchor_rows: Sequence[dict[str, Any]], attestation: Any, # noqa: ANN401 — _worm.Attestation | None *, public_key_der: bytes | None, require_signatures: bool, verify_tsa: bool, tsa_trust_pem: str, tsa_clock_skew_seconds: int, ) -> Any: # noqa: ANN401 — _worm.AttestationCheck (vendored) """Most na vendored _worm.verify_attestation_against_live s offline KMS + TSA callbackmi.""" import _worm from _tsa import verify_timestamp def _kms_cb(digest: str, signature: bytes | None) -> bool: return _offline_kms_verify(digest, _as_bytes(signature), public_key_der) def _tsa_cb(digest: str, token: bytes | None) -> Any: # noqa: ANN401 return verify_timestamp( digest, _as_bytes(token), trust_pem=tsa_trust_pem, clock_skew_seconds=tsa_clock_skew_seconds, ) return _worm.verify_attestation_against_live( anchor_rows, attestation, require_signatures=require_signatures, offline_kms_verify=_kms_cb, verify_tsa=verify_tsa, tsa_verify=_tsa_cb, ) # --------------------------------------------------------------------------- # # Loadery: JSON export / RDS (psycopg) # --------------------------------------------------------------------------- # def _load_rows_from_export(path: Path) -> list[dict[str, Any]]: """Nacitaj export audit_log z JSON suboru (zoznam riadkov alebo {"rows": [...]}).""" with path.open(encoding="utf-8") as f: data = json.load(f) rows: list[dict[str, Any]] = data["rows"] if isinstance(data, dict) else data return sorted(rows, key=lambda r: int(r["seq"])) def _load_rows_from_rds(dsn: str, inspection_id: str) -> list[dict[str, Any]]: """Nacitaj audit_log riadky pre inspekciu priamo z RDS (psycopg), seq ASC. Otvara vlastne read-only spojenie (verifikator je nezavisly proces). SELECT poradie stlpcov sedi s _AUDIT_COLUMNS; bytea sa vrati ako memoryview a normalizuje sa v _as_bytes pri overeni. """ import psycopg query = ( "SELECT seq, inspection_id, event_type, actor_id, actor_role, server_ts, " " payload_json, payload_sha256, prev_hash, record_hash, " " kms_key_id, signature, tsa_token " "FROM audit_log WHERE inspection_id = %(iid)s ORDER BY seq ASC" ) with psycopg.connect(dsn) as conn, conn.cursor() as cur: cur.execute(query, {"iid": inspection_id}) fetched = cur.fetchall() return [dict(zip(_AUDIT_COLUMNS, record, strict=True)) for record in fetched] def _fetch_public_key_der(kms_key_id: str, region: str) -> bytes: """KMS GetPublicKey -> DER bajty (SubjectPublicKeyInfo) na OFFLINE overenie. JEDINE volanie na KMS (raz). Verejny kluc nie je tajny; samotne overenie podpisov potom prebieha lokalne (`_offline_kms_verify`), bez kms.Verify. """ import boto3 client = boto3.client("kms", region_name=region) resp = client.get_public_key(KeyId=kms_key_id) return bytes(resp["PublicKey"]) def _load_public_key_der(path: Path) -> bytes: """Nacitaj doveryhodny verejny kluc zo suboru (PEM alebo DER) -> DER SPKI bajty. Alternativa k --kms-key-id pre uplne offline overenie bez pristupu na AWS. """ from cryptography.hazmat.primitives.serialization import ( Encoding, PublicFormat, load_der_public_key, load_pem_public_key, ) raw = path.read_bytes() key = load_pem_public_key(raw) if raw.lstrip().startswith(b"-----") else load_der_public_key( raw ) return key.public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo) def _resolve_public_key_der(config: VerifierConfig) -> bytes | None: """Resolvni doveryhodny verejny kluc: --public-key (PEM/DER) alebo KMS GetPublicKey. FIX #3 — FAIL-CLOSED: _check_source uz garantuje, ze pri require_signatures je aspon jeden zdroj kluca; tu ho len nacitame. """ if config.public_key is not None: return _load_public_key_der(config.public_key) if config.kms_key_id: return _fetch_public_key_der(config.kms_key_id, config.region) return None def _load_anchor_rows(path: Path) -> list[dict[str, Any]]: """Nacitaj epoch_anchor export (JSON zoznam / {"rows": [...]}), seq ASC.""" with path.open(encoding="utf-8") as f: data = json.load(f) rows: list[dict[str, Any]] = data["rows"] if isinstance(data, dict) else data return sorted(rows, key=lambda r: int(r["seq"])) def _load_anchor_rows_from_rds(dsn: str) -> list[dict[str, Any]]: """Nacitaj epoch_anchor riadky priamo z RDS (psycopg), seq ASC.""" import psycopg cols = ( "seq", "sealed_inspection_id", "sealed_head_hash", "prev_anchor_hash", "anchor_hash", "server_ts", "kms_key_id", "signature", "tsa_token", ) query = ( "SELECT seq, sealed_inspection_id, sealed_head_hash, prev_anchor_hash, " " anchor_hash, server_ts, kms_key_id, signature, tsa_token " "FROM epoch_anchor ORDER BY seq ASC" ) with psycopg.connect(dsn) as conn, conn.cursor() as cur: cur.execute(query) fetched = cur.fetchall() return [dict(zip(cols, rec, strict=True)) for rec in fetched] def _fetch_inspection_status(dsn: str, inspection_id: str) -> str | None: """Nacitaj inspections.status z RDS (cross-check seal-required, FIX #4).""" import psycopg with psycopg.connect(dsn) as conn, conn.cursor() as cur: cur.execute("SELECT status FROM inspections WHERE id = %(iid)s", {"iid": inspection_id}) row = cur.fetchone() return str(row[0]) if row else None def _load_latest_attestation(config: VerifierConfig) -> Any: # noqa: ANN401 — _worm.Attestation """Nacitaj NAJNOVSIU WORM atestaciu epoch rootu (autorita). Zdroj (v poradi): --attestation-path (adresar/subor WORM objektov) ALEBO --worm-bucket (S3). FINAL vrstva: ziva kotva sa cross-checkne proti tejto atestacii, takze NESMIE pochadzat z DB (DB superuser ju vie zmazat) — len z immutable WORM store. """ import _worm if config.attestation_path is not None: return _worm.load_latest_attestation_from_dir(config.attestation_path) if config.worm_bucket: return _worm.load_latest_attestation_from_s3(config.worm_bucket, config.region) return None def load_rows(config: VerifierConfig) -> list[dict[str, Any]]: """Nacitaj riadky podla konfiguracie (export ALEBO RDS).""" if config.export is not None: return _load_rows_from_export(config.export) # _check_source garantuje: dsn + inspection_id su nastavene ked nie je export. assert config.dsn is not None # noqa: S101 — invariant z VerifierConfig validatora assert config.inspection_id is not None # noqa: S101 return _load_rows_from_rds(config.dsn, config.inspection_id) # --------------------------------------------------------------------------- # # CLI # --------------------------------------------------------------------------- # def _parse_args(argv: Sequence[str] | None) -> VerifierConfig: parser = argparse.ArgumentParser( description="Walk-Check forensic audit chain verifier (independent of the app)", ) source = parser.add_mutually_exclusive_group(required=True) source.add_argument("--export", type=Path, help="JSON export audit_log") source.add_argument("--dsn", help="psycopg DSN for RDS (with --inspection-id)") parser.add_argument("--inspection-id", help="inspection UUID (required with --dsn)") parser.add_argument("--kms-key-id", help="KMS key id for GetPublicKey (offline signature verify)") parser.add_argument( "--public-key", type=Path, help="PEM/DER public key (offline, alt. to --kms-key-id)" ) parser.add_argument("--region", default="eu-central-1", help="AWS region for KMS") parser.add_argument( "--tsa-trust", type=Path, help="PEM bundle of pinned TSA certs (trust anchor)" ) parser.add_argument( "--tsa-clock-skew", type=int, default=300, help="clock tolerance (s) for genTime" ) parser.add_argument("--no-tsa", action="store_true", help="skip TSA token verification") parser.add_argument( "--anchor-export", type=Path, help="JSON export of epoch_anchor (inclusion)" ) parser.add_argument( "--require-anchor", action="store_true", help="require inclusion of the sealed head in the global anchor", ) parser.add_argument( "--attestation-path", type=Path, help="directory/file of WORM attestations of the epoch root (anchor-attestations/.json)", ) parser.add_argument( "--worm-bucket", help="S3 WORM bucket with attestations (alt. to --attestation-path)" ) parser.add_argument( "--require-attestation", action="store_true", help="cross-check the live anchor against the WORM attestation of the epoch root", ) parser.add_argument( "--allow-unsigned", action="store_true", help="DEV: do not require signatures (prints UNVERIFIED_DEV, NOT VERIFIED)", ) parser.add_argument("--json-logs", action="store_true", help="structlog JSON renderer") args = parser.parse_args(argv) return VerifierConfig( export=args.export, dsn=args.dsn, inspection_id=args.inspection_id, kms_key_id=args.kms_key_id, public_key=args.public_key, region=args.region, tsa_trust=args.tsa_trust, tsa_clock_skew_seconds=args.tsa_clock_skew, verify_tsa=not args.no_tsa, anchor_export=args.anchor_export, require_anchor=args.require_anchor, attestation_path=args.attestation_path, worm_bucket=args.worm_bucket, require_attestation=args.require_attestation, allow_unsigned=args.allow_unsigned, json_logs=args.json_logs, ) def main(argv: Sequence[str] | None = None) -> int: # noqa: C901 — verdikt mapovanie + zdroje """Entry point: nacitaj riadky, over retazec, vypis verdikt, exit 0/1. FIX #3 — FAIL-CLOSED: bez --allow-unsigned MUSI byt doveryhodny kluc (vynucuje VerifierConfig._check_source). Ak nie je, verifikator skonci s ERROR (exit 2), NIKDY neoveruje "naslepo" s public_key=None. UNVERIFIED_DEV je distinktny od VERIFIED. """ from pydantic import ValidationError try: config = _parse_args(argv) except ValidationError as exc: configure_logging() # Len vety dovodov, bez vypisu pydantic (vstupny slovnik, odkaz na dokumentaciu): # cita to clovek s PDF v ruke, nie programator (3. 10. 2026, OVEROVAC-1). dovody = "; ".join(str(e["msg"]).removeprefix("Value error, ") for e in exc.errors()) log.error("config_invalid", error=dovody) # noqa: TRY400 - clean verdikt, nie traceback vypis(f"ERROR (config): {dovody}") return 2 configure_logging(json_logs=config.json_logs) source = "rds" if config.dsn else "export" log.info("verify_start", source=source, inspection_id=config.inspection_id) # NECITATELNY SUBOR JE CHYBA PRIKAZU, KOD 2 (3. 10. 2026, OVEROVAC-1 nalez N8). # Chybajuci subor, subor, ktory nie je JSON, alebo certifikat namiesto kluca koncili # vynimkou a kodom 1, teda rovnako ako TAMPERED, hoci sa nic neoverilo. Kod 2 dostane # LEN subor, ktory sa neda precitat. Export, ktory sa precitat da, ale ma pokazeny # obsah (chybajuce pole, zly cas), ostava ako doteraz: vynimka a kod 1, lebo pre # automatizaciu je to poplach a review 3. 10. ziadalo, aby sa nezmenil. Overenie # samotne (verify) bezi mimo tychto blokov. from cryptography.exceptions import UnsupportedAlgorithm def _necitatelny(co: str, exc: Exception) -> int: log.error("input_unreadable", input=co, error=f"{type(exc).__name__}: {exc}") vypis(f"ERROR (input): cannot read {co}: {type(exc).__name__}: {exc}") return 2 necitatelny_json = (OSError, UnicodeDecodeError, json.JSONDecodeError) try: rows = load_rows(config) except necitatelny_json as exc: return _necitatelny("the export", exc) log.info("rows_loaded", records=len(rows)) try: public_key_der = _resolve_public_key_der(config) except (OSError, ValueError, UnsupportedAlgorithm) as exc: return _necitatelny("the public key", exc) if public_key_der is not None: log.info("trusted_key_loaded", source="public_key" if config.public_key else "kms") elif config.allow_unsigned: log.warning("kms_skipped", detail="--allow-unsigned: signatures are NOT verified (dev)") try: tsa_trust_pem = config.tsa_trust.read_text(encoding="utf-8") if config.tsa_trust else "" # Subor bez certifikatu by kontrolu dovery TICHO vypol (_tsa.py: prazdny trust = # bez kontroly) a chvost by tvrdil, ze sa overila (review 3. 10. 2026). if config.tsa_trust: from cryptography import x509 x509.load_pem_x509_certificates(tsa_trust_pem.encode("utf-8")) except (OSError, UnicodeDecodeError, ValueError) as exc: return _necitatelny("the timestamp authority anchor (--tsa-trust)", exc) if config.verify_tsa and not tsa_trust_pem: log.warning("tsa_trust_skipped", detail="without --tsa-trust the TSA cert is NOT validated (dev)") inspection_status = ( _fetch_inspection_status(config.dsn, config.inspection_id) if config.dsn and config.inspection_id else None ) # anchor_rows treba aj pre attestation cross-check (FINAL vrstva), nielen pre inkluziu. anchor_rows: list[dict[str, Any]] | None = None if config.require_anchor or config.require_attestation: if config.anchor_export is not None: try: anchor_rows = _load_anchor_rows(config.anchor_export) except necitatelny_json as exc: return _necitatelny("the global anchor (--anchor-export)", exc) if not anchor_rows: # Prazdny export je vrstva bez dat ako pri N3: nad nim vysiel zdravy zaznam # TAMPERED (global_anchor:anchor_empty), review 3. 10. 2026. return _necitatelny("the global anchor (--anchor-export)", ValueError("the export contains no anchor entries")) elif config.dsn is not None: anchor_rows = _load_anchor_rows_from_rds(config.dsn) # NAJNOVSIA WORM atestacia epoch rootu (autorita = immutable WORM objekt). attestation: Any = None if config.require_attestation: try: attestation = _load_latest_attestation(config) except necitatelny_json as exc: return _necitatelny("the WORM attestation (--attestation-path)", exc) if attestation is None: log.warning("attestation_missing", detail="no WORM attestation -> PENDING") result = verify( rows, public_key_der=public_key_der, verify_tsa=config.verify_tsa, require_signatures=config.require_signatures, tsa_trust_pem=tsa_trust_pem, tsa_clock_skew_seconds=config.tsa_clock_skew_seconds, inspection_status=inspection_status, anchor_rows=anchor_rows, require_anchor=config.require_anchor, attestation=attestation, require_attestation=config.require_attestation, ) status = verdict(result) # CO SA NEOVEROVALO, SA MUSI POVEDAT (31.08.2026, nocny audit). # # PDF tlaci prikaz `verify_chain.py --export … --public-key …` a hned pod nim # vetu, ze verdikt sa musi zhodovat s tym, co je vytlacene vyssie. Ten prikaz # ale nema `--tsa-trust`, `--require-anchor` ani `--require-attestation`, # takze doveryhodnost casovej peciatky, globalna kotva ani WORM atestacia sa # NEOVERUJU — a verifier napriek tomu vypisal hole slovo VERIFIED. Tretia # strana (poistovna, sud) tak citala silnejsie tvrdenie, nez sa overilo. # # Verdikt sa NEMENI (retazec naozaj sedi), ale nesie rozsah. Kto chce plne # overenie, prida vypisane prepinace. preskocene: list[str] = [] # `--no-tsa` preskoci casovu peciatku uplne; do 3. 10. 2026 to chvost nepovedal # a VERIFIED vyslo aj nad pecatou bez peciatky (OVEROVAC-1 nalez N7). if not config.verify_tsa: preskocene.append("timestamp (skipped by --no-tsa)") elif not config.tsa_trust: preskocene.append("timestamp trust (--tsa-trust)") if not config.require_anchor: preskocene.append("global anchor (--require-anchor)") if not config.require_attestation: preskocene.append("WORM attestation (--require-attestation)") rozsah = ("" if not preskocene else ", NOT CHECKED: " + ", ".join(preskocene)) # REZIM ULOZISKA (0039) sa vypisuje SAMOSTATNE a verdikt nemeni. Retazec # pilotneho zaznamu je rovnako platny ako retazec ostreho; rozdiel je len # v tom, ci sa bajty daju zmazat. Tretia strana to ma vediet od verifiera, # nie od nas — inak by musela verit vete v PDF. # # ZNENIE 27. 9. 2026 (0073): „can be deleted on request" uz nie je pravda. Pilot ani # organizacia zapecateny zaznam nezmazu, poziadaju; zaznam sa im skryje a uchova ako # dokaz. Zmazat ho moze uz len Walk-Check, vynimocne. Tretia strana ma vediet prave # toto: zamok nie je, takze zmiznut moze, ale len naso vynimocne rozhodnutie. # Pri TAMPERED sa z retaze nevypisuje nic, ani rezim uloziska: retazec, ktory nastroj # v tom istom behu vyhlasil za poruseny, nie je zdroj ziadnej vety (OVEROVAC-1 nalez N9). rezim = ulozisko_rezim_z_riadkov(rows) if rezim == REZIM_PILOT and status != STATUS_TAMPERED: log.warning("ulozisko_pilot", detail="the record is not under an irreversible lock") vypis( "STORAGE: pilot mode, the record is NOT under an irreversible lock; " "it is kept for at least four years, then in a deep archive, and only " "Walk-Check can delete it, as an exception (this does not weaken the chain)" ) # POLOHA ZARIADENIA (ADR poloha, dodatok §10). Riadky idu AZ ZA riadok # verdiktu, nie medzi STORAGE a verdikt: pri ostrom zazname je verdikt # prvy riadok stdout a tak ho citaju testy aj ludia # (`api/tests/test_verify_status.py`, `test_verifier_prod_fixture.py`). # Pri TAMPERED sa nevypise nic a verdikt poloha nemeni nikdy. Nezmysel # v retazi nesmie zhodit overovac (zmenil by sa exit kod), preto `except`. def _vypis_polohu() -> None: try: riadky = riadky_polohy(rows) except Exception: # noqa: BLE001 - poloha nikdy nemeni vysledok overenia riadky = [_POLOHA_PREDPONA + "could not be read from this chain"] for riadok in riadky: vypis(riadok) # ZAPECATENA HLAVA NA POROVNANIE S PDF (3. 10. 2026, OVEROVAC-1 nalez N5). Navod # hovori, ze verdikt a zapecatena hlava sa musia zhodovat s PDF, ale nastroj hlavu # nevypisal: iny zdravy zaznam podvrhnuty do prilohy vysiel VERIFIED a clovek nemal # s cim porovnat. Slova su tie iste ako v PDF. Pri nezapecatenej retazi riadok nie je. # Pri `--allow-unsigned` (UNVERIFIED_DEV) nie je tiez: podpis pecate sa neoveril # a zhoda s PDF by posobila ako potvrdenie. ID sa vypise LEN ako UUID: pri # nepodpisanom exporte by inak novy riadok v `inspection_id` podvrhol na vystup # samostatny riadok „VERIFIED ..." (review 3. 10. 2026). def _vypis_pecat() -> None: if result.seal_hash is None or result.dev_unverified: return iid = str(rows[0].get("inspection_id")) try: je_uuid = str(uuid.UUID(iid)) == iid except ValueError: je_uuid = False vypis( f"SEALED HEAD (record_hash): {result.seal_hash}, " f"WALKAROUND ID: {iid if je_uuid else 'not a valid UUID, see the export'}" ) if status == STATUS_PENDING: log.warning("verify_pending", records=len(rows)) # „anchored" len vtedy, ked sa kotva v tomto behu naozaj overila (nalez N4). stav = ("sealed and anchored, but NOT YET ATTESTED in WORM" if config.require_anchor else "sealed, but NOT YET ATTESTED in WORM") vypis(f"PENDING ({stav}, records={len(rows)}, not a finalized proof){rozsah}") _vypis_pecat() _vypis_polohu() return 0 if status == STATUS_UNSEALED: # Konzistentny, ale bez pecate: nie je co overovat. Exit 0, lebo # retazec nie je poskodeny; slovo VERIFIED tu ale nepadne. log.warning("verify_unsealed", records=len(rows)) vypis( f"UNSEALED (chain consistent but NOT SEALED, records={len(rows)}, " "work in progress, not a proof)" ) _vypis_polohu() return 0 if status == STATUS_UNVERIFIED_DEV: log.warning("verify_unverified_dev", sealed=result.sealed, records=len(rows)) vypis( f"UNVERIFIED_DEV (sealed={result.sealed}, records={len(rows)}, " "signatures NOT verified, not a proof)" ) _vypis_polohu() return 0 if status == STATUS_VERIFIED: log.info("verify_ok", sealed=result.sealed, records=len(rows)) vypis(f"VERIFIED (sealed={result.sealed}, records={len(rows)}){rozsah}") _vypis_pecat() _vypis_polohu() return 0 log.error( "verify_tampered", broken_seq=result.broken_seq, reason=result.reason, sealed=result.sealed, ) vypis(f"TAMPERED at seq={result.broken_seq} reason={result.reason}") return 1 if __name__ == "__main__": raise SystemExit(main())