bellsystems-cp/backend/firmware/elf_store.py

"""Firmware ELF symbol store + server-side crash decoding (firmware F-070).

Crash reports carry `elf_sha256`: the first 16 hex chars of the SHA-256 of the
crashing build's firmware.elf. ELFs uploaded here are stored under that same
key, so a crash can be matched to its exact build and its PC/backtrace
addresses decoded to function + file:line with xtensa addr2line.

Decoding is optional: it needs a matching ELF AND the xtensa-esp32-elf-addr2line
binary in the backend container (settings.addr2line_path, or on PATH). When
either is missing the console falls back to a copy-paste addr2line command.
"""

import hashlib
import json
import re
import shutil
import subprocess
from datetime import datetime, timezone
from pathlib import Path

from config import settings
from shared.exceptions import NotFoundError, ValidationError

ADDR2LINE_NAME = "xtensa-esp32-elf-addr2line"
_SHA_PREFIX = re.compile(r"^[0-9a-f]{16}$")
_MAX_ADDRESSES = 40


def _elf_dir() -> Path:
    return Path(settings.firmware_storage_path) / "elf"


def elf_sha_prefix(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()[:16]


def _check_sha(sha: str) -> str:
    sha = (sha or "").lower()
    if not _SHA_PREFIX.match(sha):
        raise ValidationError("elf_sha256 must be 16 hex characters.")
    return sha


def addr2line_path() -> str | None:
    configured = getattr(settings, "addr2line_path", "") or ""
    if configured and Path(configured).is_file():
        return configured
    return shutil.which(ADDR2LINE_NAME)


def validate_elf(data: bytes) -> None:
    if not data.startswith(b"\x7fELF"):
        raise ValidationError("Not an ELF file — upload the firmware.elf from the build, not the .bin.")


def save_elf(data: bytes, filename: str | None, uploaded_by: str | None,
             firmware: dict | None = None) -> dict:
    """Store an ELF keyed by its sha256 prefix. Re-uploading the same file is a
    no-op overwrite. `firmware` optionally records which release it belongs to."""
    validate_elf(data)
    sha = elf_sha_prefix(data)
    d = _elf_dir()
    d.mkdir(parents=True, exist_ok=True)
    (d / f"{sha}.elf").write_bytes(data)
    meta = {
        "elf_sha256": sha,
        "filename": filename or "firmware.elf",
        "size": len(data),
        "uploaded_at": datetime.now(timezone.utc).isoformat(),
        "uploaded_by": uploaded_by,
        "firmware": firmware,
    }
    (d / f"{sha}.json").write_text(json.dumps(meta), encoding="utf-8")
    return meta


def get_elf_info(sha: str) -> dict:
    """Whether a matching ELF exists and whether the decoder is installed —
    the crash view uses this to pick decode-in-console vs. copy-command."""
    sha = _check_sha(sha)
    meta_path = _elf_dir() / f"{sha}.json"
    meta = None
    if (_elf_dir() / f"{sha}.elf").is_file():
        try:
            meta = json.loads(meta_path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            meta = {"elf_sha256": sha}
    return {
        "elf_sha256": sha,
        "exists": meta is not None,
        "meta": meta,
        "decoder_available": addr2line_path() is not None,
    }


def decode(sha: str, addresses: list[int]) -> list[dict]:
    """Run addr2line -pfiaC over the addresses. Returns one entry per address:
    {"address": "0x400D8A3F", "frames": ["func at file:line", "(inlined by) ..."]}."""
    sha = _check_sha(sha)
    elf = _elf_dir() / f"{sha}.elf"
    if not elf.is_file():
        raise NotFoundError("ELF")
    tool = addr2line_path()
    if tool is None:
        raise ValidationError(f"{ADDR2LINE_NAME} is not installed on the server.")

    addrs = [a for a in addresses if isinstance(a, int) and 0 <= a <= 0xFFFFFFFF][:_MAX_ADDRESSES]
    if not addrs:
        return []
    hex_addrs = [f"0x{a:08X}" for a in addrs]
    proc = subprocess.run(
        [tool, "-pfiaC", "-e", str(elf), *hex_addrs],
        capture_output=True, text=True, timeout=20,
    )
    if proc.returncode != 0:
        raise ValidationError(f"addr2line failed: {proc.stderr.strip()[:300]}")

    # With -a each address starts a new block "0x400d8a3f: func at file:line";
    # -i adds " (inlined by) ..." continuation lines under it.
    results: list[dict] = []
    for line in proc.stdout.splitlines():
        if not line.strip():
            continue
        m = re.match(r"^(0x[0-9a-fA-F]+):\s*(.*)$", line)
        if m:
            results.append({"address": f"0x{int(m.group(1), 16):08X}", "frames": [m.group(2).strip()]})
        elif results:
            results[-1]["frames"].append(line.strip())
    return results