1
"""Create and validate a private, verified backup of the supported board's flash."""
2
from datetime import datetime, timezone
3
import fcntl
4
import hashlib
5
import json
6
import os
7
from pathlib import Path
8
import subprocess
9
import sys
10
import tempfile
11
 
12
from project import ROOT
13
 
14
FLASH_BYTES = 32 * 1024 * 1024
15
IMAGE_NAME = "factory-flash-32mb.bin"
16
RECORD_NAME = "factory-backup.json"
17
 
18
 
19
def backup_directory():
20
    return Path(os.environ.get("RADIO_BACKUP_DIR", ROOT / "artifacts/hardware-validation"))
21
 
22
 
23
def sha256(path):
24
    digest = hashlib.sha256()
25
    with path.open("rb") as source:
26
        for chunk in iter(lambda: source.read(1024 * 1024), b""):
27
            digest.update(chunk)
28
    return digest.hexdigest()
29
 
30
 
31
def require_backup(directory=None):
32
    directory = Path(directory) if directory is not None else backup_directory()
33
    image = directory / IMAGE_NAME
34
    try:
35
        record = json.loads((directory / RECORD_NAME).read_text())
36
        valid = (isinstance(record, dict) and
37
                 record.get("verified_against_device_flash") is True and
38
                 record.get("bytes") == FLASH_BYTES and
39
                 image.stat().st_size == FLASH_BYTES and
40
                 record.get("sha256") == sha256(image))
41
    except (OSError, ValueError):
42
        valid = False
43
    if not valid:
44
        raise ValueError(f"No valid saved backup in {directory}. Run make backup before the first flash.")
45
    return image
46
 
47
 
48
def create_backup(directory, port, run=None):
49
    """Publish a record only after esptool verifies the complete saved image."""
50
    run = subprocess.run if run is None else run
51
    directory = Path(directory)
52
    directory.mkdir(parents=True, exist_ok=True, mode=0o700)
53
    lock = os.open(directory / ".backup.lock", os.O_CREAT | os.O_RDWR, 0o600)
54
    try:
55
        try:
56
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
57
        except BlockingIOError:
58
            raise ValueError("Another backup is already running in this directory.") from None
59
        image, record = directory / IMAGE_NAME, directory / RECORD_NAME
60
        if image.exists() or record.exists():
61
            raise ValueError("A backup already exists. Use make backup-check to check it; choose a different RADIO_BACKUP_DIR for another board.")
62
        with tempfile.TemporaryDirectory(prefix=".backup-", dir=directory) as temporary:
63
            saved = Path(temporary) / IMAGE_NAME
64
            saved.touch(mode=0o600)
65
            base = [sys.executable, "-m", "esptool", "--chip", "esp32s3",
66
                    "--port", port, "--baud", "460800"]
67
            run(base + ["--before", "default-reset", "--after", "no-reset",
68
                        "read-flash", "0", str(FLASH_BYTES), str(saved)], check=True)
69
            if saved.stat().st_size != FLASH_BYTES:
70
                raise ValueError("The flash read was incomplete; no verified backup was recorded.")
71
            run(base + ["--before", "no-reset", "--after", "hard-reset",
72
                        "verify-flash", "0", str(saved)], check=True)
73
            metadata = {
74
                "bytes": FLASH_BYTES,
75
                "sha256": sha256(saved),
76
                "chip": "ESP32-S3",
77
                "port": port,
78
                "created_at": datetime.now(timezone.utc).isoformat(),
79
                "verification": "esptool verify-flash",
80
                "verified_against_device_flash": True,
81
            }
82
            pending = Path(temporary) / RECORD_NAME
83
            pending.touch(mode=0o600)
84
            pending.write_text(json.dumps(metadata, indent=2) + "\n")
85
            for path in (saved, pending):
86
                with path.open("rb") as output:
87
                    os.fsync(output.fileno())
88
            # Exclusive links also protect against a destination created outside our lock.
89
            os.link(saved, image)
90
            os.link(pending, record)
91
            directory_fd = os.open(directory, os.O_RDONLY)
92
            try:
93
                os.fsync(directory_fd)
94
            finally:
95
                os.close(directory_fd)
96
        return image
97
    finally:
98
        os.close(lock)