97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Hash-bound, offline-only appinfo snapshot query for Phase 1.0DH."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
import hashlib
|
|
from pathlib import Path
|
|
import re
|
|
import sqlite3
|
|
from urllib.parse import quote
|
|
|
|
|
|
PHASE = "PHASE_1_0DH_HASH_BOUND_SNAPSHOT_QUERY"
|
|
TITLE_ID = "PPSA01659"
|
|
MAX_SNAPSHOT_BYTES = 64 * 1024 * 1024
|
|
SHA256 = re.compile(r"^[0-9a-f]{64}$")
|
|
|
|
|
|
class SnapshotError(ValueError):
|
|
"""The supplied snapshot or query contract is invalid."""
|
|
|
|
|
|
class Outcome(Enum):
|
|
PRESENT = "PRESENT"
|
|
ABSENT = "ABSENT"
|
|
UNKNOWN = "UNKNOWN"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SnapshotBinding:
|
|
phase: str
|
|
title_id: str
|
|
size: int
|
|
sha256: str
|
|
|
|
|
|
def _digest(path: Path) -> tuple[int, str]:
|
|
total = 0
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
while block := stream.read(64 * 1024):
|
|
total += len(block)
|
|
if total > MAX_SNAPSHOT_BYTES:
|
|
raise SnapshotError("snapshot exceeds size ceiling")
|
|
digest.update(block)
|
|
return total, digest.hexdigest()
|
|
|
|
|
|
def validate_binding(binding: SnapshotBinding) -> None:
|
|
if type(binding) is not SnapshotBinding or binding.phase != PHASE:
|
|
raise SnapshotError("phase binding mismatch")
|
|
if binding.title_id != TITLE_ID:
|
|
raise SnapshotError("title binding mismatch")
|
|
if not isinstance(binding.size, int) or isinstance(binding.size, bool) or \
|
|
binding.size <= 0 or binding.size > MAX_SNAPSHOT_BYTES:
|
|
raise SnapshotError("snapshot size is invalid")
|
|
if not isinstance(binding.sha256, str) or not SHA256.fullmatch(binding.sha256):
|
|
raise SnapshotError("snapshot hash is invalid")
|
|
|
|
|
|
def query_snapshot(path: Path, binding: SnapshotBinding) -> Outcome:
|
|
"""Query a byte-exact local snapshot without creating SQLite sidecars."""
|
|
validate_binding(binding)
|
|
path = path.resolve(strict=True)
|
|
if not path.is_file() or _digest(path) != (binding.size, binding.sha256):
|
|
raise SnapshotError("snapshot identity mismatch")
|
|
if path.read_bytes()[:16] != b"SQLite format 3\x00":
|
|
raise SnapshotError("snapshot is not SQLite 3")
|
|
|
|
uri = f"file:{quote(path.as_posix(), safe='/:')}?mode=ro&immutable=1"
|
|
try:
|
|
connection = sqlite3.connect(uri, uri=True)
|
|
try:
|
|
connection.execute("PRAGMA query_only = ON")
|
|
columns = {
|
|
row[1] for row in connection.execute(
|
|
"PRAGMA table_info(tbl_appinfo)")
|
|
}
|
|
if "titleId" not in columns:
|
|
return Outcome.UNKNOWN
|
|
rows = connection.execute(
|
|
"SELECT 1 FROM tbl_appinfo WHERE titleId = ? LIMIT 2",
|
|
(TITLE_ID,),
|
|
).fetchall()
|
|
if len(rows) == 1:
|
|
return Outcome.PRESENT
|
|
if len(rows) == 0:
|
|
return Outcome.ABSENT
|
|
return Outcome.UNKNOWN
|
|
finally:
|
|
connection.close()
|
|
except sqlite3.Error:
|
|
return Outcome.UNKNOWN
|