#!/usr/bin/env python3

"""

SOF Installation MD5 Validator (validate_sof_install.py)

Validates that Sound Open Firmware (SOF) binaries, LLEXT modules, topologies,

and userspace tools from the sof-bin repository are correctly installed on a

target host's /lib/firmware directory using MD5 checksums.

Features:

- Local host validation (/lib/firmware/intel or custom path)

- Offline rootfs / NFS rootfs validation (--target-root /srv/nfs/spider-rootfs)

- Remote SSH target validation (--ssh root@spider)

- Filesystem identification mode (-i / --identify) to inspect installed versions

- Mismatch detection against historical SOF releases

- Displays version of installed files in summary table

- Fix mode (--fix, --dry-run) to automatically sync/repair installed files

- Platform filtering (-p tgl, -p mtl, -p ptl, etc.)

- Component filtering (-c fw, -c llext, -c tplg, -c tools)

- Signing flavor filtering (-f community, -f intel-signed)

- JSON output for test automation (--json)

- Standalone md5sum manifest export (--generate-md5 <file>)

"""

from __future__ import annotations

import argparse

import base64

import hashlib

import io

import json

import os

import re

import shlex

import shutil

import struct

import subprocess

import sys

import tarfile

import tempfile

from dataclasses import dataclass, field

from enum import Enum

from pathlib import Path

from typing import Any, Dict, List, Optional, Set, Tuple

class Status(Enum):

MATCH = "MATCH"

MISMATCH = "MISMATCH"

MISSING = "MISSING"

SYMLINK_MATCH = "SYMLINK_MATCH"

SYMLINK_MISMATCH = "SYMLINK_MISMATCH"

BROKEN_SYMLINK = "BROKEN_SYMLINK"

EXTRA = "EXTRA"

ERROR = "ERROR"

@dataclass

class ValidationRecord:

component: str # fw, llext, tplg, tools

platform: str # tgl, mtl, ptl, generic, etc.

rel_path: str # relative to fw_dest (or tools_dest)

expected_version: str = "" # target version requested e.g. v2.14.1

installed_version: Optional[str] = None # version found on host e.g. v2.12, v2.14.1, unknown

source_path: Optional[str] = None # in sof-bin

target_path: Optional[str] = None # on host

is_symlink: bool = False

symlink_target_expected: Optional[str] = None

symlink_target_actual: Optional[str] = None

expected_md5: Optional[str] = None

actual_md5: Optional[str] = None

expected_size: Optional[int] = None

actual_size: Optional[int] = None

matched_release: Optional[str] = None # if mismatch matches an earlier release

fix_applied: bool = False

fix_message: Optional[str] = None

status: Status = Status.MISSING

message: str = ""

expected_manifest: Optional[Dict[str, Any]] = None

actual_manifest: Optional[Dict[str, Any]] = None

upgrade_available: Optional[Dict[str, Any]] = None

def to_dict(self) -> Dict[str, Any]:

return {

"component": self.component,

"platform": self.platform,

"rel_path": self.rel_path,

"expected_version": self.expected_version,

"installed_version": self.installed_version,

"source_path": self.source_path,

"target_path": self.target_path,

"is_symlink": self.is_symlink,

"symlink_target_expected": self.symlink_target_expected,

"symlink_target_actual": self.symlink_target_actual,

"expected_md5": self.expected_md5,

"actual_md5": self.actual_md5,

"expected_size": self.expected_size,

"actual_size": self.actual_size,

"matched_release": self.matched_release,

"fix_applied": self.fix_applied,

"fix_message": self.fix_message,

"status": self.status.value,

"message": self.message,

"expected_manifest": self.expected_manifest,

"actual_manifest": self.actual_manifest,

"upgrade_available": self.upgrade_available,

}

def compute_file_md5(file_path: Path, block_size: int = 65536) -> Optional[str]:

"""Compute MD5 hash of a local file safely in all environments."""

if not file_path.is_file():

return None

try:

try:

hasher = hashlib.md5(usedforsecurity=False)

except (TypeError, ValueError):

hasher = hashlib.md5()

with open(file_path, "rb") as f:

for chunk in iter(lambda: f.read(block_size), b""):

hasher.update(chunk)

return hasher.hexdigest()

except (OSError, IOError, ValueError, Exception):

return None

def parse_binary_manifest(data: bytes) -> Dict[str, Any]:

"""

Parse embedded manifest and metadata from firmware (.ri), LLEXT (.llext/.bin),

and topology (.tplg) binaries.

Supports:

- CSS headers ($MN2): BCD build date (YYYY-MM-DD)

- ADSP FW header ($AM1): SOF major/minor/hotfix/build version, component name, modules

- Extended Manifest (XMan): FW version, ABI version, date/time strings

- Embedded build strings: Zephyr version/commit, SOF tags

- ALSA Topology Manifest (CoSA / type 8): Topology SOF ABI version

"""

if not data:

return {}

info: Dict[str, Any] = {}

# 1. CSS Header ($MN2)

pos_mn2 = data.find(b"$MN2")

if pos_mn2 != -1 and pos_mn2 >= 8:

try:

date_raw = struct.unpack_from("<I", data, pos_mn2 - 8)[0]

year = ((date_raw >> 28) & 0xf) * 1000 + ((date_raw >> 24) & 0xf) * 100 + ((date_raw >> 20) & 0xf) * 10 + ((date_raw >> 16) & 0xf)

month = ((date_raw >> 12) & 0xf) * 10 + ((date_raw >> 8) & 0xf)

day = ((date_raw >> 4) & 0xf) * 10 + (date_raw & 0xf)

if 1970 <= year <= 2099 and 1 <= month <= 12 and 1 <= day <= 31:

info["build_date"] = f"{year:04d}-{month:02d}-{day:02d}"

except Exception:

pass

# 2. ADSP FW Manifest ($AM1)

pos_am1 = data.find(b"$AM1")

if pos_am1 != -1 and pos_am1 + 52 <= len(data):

try:

hdr_id, hdr_len, name, preload_pages, flags, feat_mask, maj, min_v, hotfix, bld, num_mods = struct.unpack_from(

"<4sI8sIIIHHHHI", data, pos_am1

)

if maj == 0:

ver_str = f"v{min_v}.{hotfix}.{bld}"

elif bld in (0, 1):

ver_str = f"v{maj}.{min_v}.{hotfix}"

else:

ver_str = f"v{maj}.{min_v}.{hotfix}.{bld}"

info["manifest_version"] = ver_str

name_str = name.decode("ascii", "replace").strip("\x00")

if name_str:

info["component_name"] = name_str

info["modules_count"] = num_mods & 0xffff

except Exception:

pass

# 3. Extended Manifest (XMan)

pos_xman = data.find(b"XMan")

if pos_xman != -1 and pos_xman + 16 <= len(data):

try:

magic, full_sz, hdr_sz, hdr_ver = struct.unpack_from("<IIII", data, pos_xman)

offset = pos_xman + hdr_sz

end = min(pos_xman + full_sz, len(data))

while offset + 8 <= end:

elem_type, elem_sz = struct.unpack_from("<II", data, offset)

if elem_sz == 0:

break

if elem_type == 0 and offset + 16 + 44 <= len(data): # FW_VERSION

v_off = offset + 16

major, minor, micro, build = struct.unpack_from("<HHHH", data, v_off)

date_b = data[v_off+8:v_off+20].split(b"\x00")[0].decode("ascii", "replace").strip()

time_b = data[v_off+20:v_off+30].split(b"\x00")[0].decode("ascii", "replace").strip()

tag_b = data[v_off+30:v_off+36].split(b"\x00")[0].decode("ascii", "replace").strip()

abi_ver, src_hash = struct.unpack_from("<II", data, v_off+36)

if not info.get("manifest_version"):

if major == 0:

info["manifest_version"] = f"v{minor}.{micro}.{build}"

elif build in (0, 1):

info["manifest_version"] = f"v{major}.{minor}.{micro}"

else:

info["manifest_version"] = f"v{major}.{minor}.{micro}.{build}"

if date_b and not info.get("build_date"):

info["build_date"] = date_b

if time_b:

info["build_time"] = time_b

if tag_b:

info["sof_tag"] = tag_b

if abi_ver:

info["abi_version"] = f"0x{abi_ver:x}"

offset += elem_sz

except Exception:

pass

# 4. Embedded strings (Zephyr / SOF version banner)

try:

zephyr_match = re.search(rb"Booting Zephyr OS build ([^\s\*\r\n\x00]+)", data)

if zephyr_match:

info["zephyr_build"] = zephyr_match.group(1).decode("ascii", "replace").strip()

fw_tag_match = re.search(rb"tags SOF:([^\s\x00]+) zephyr:([^\s\x00]+)", data)

if fw_tag_match:

info["sof_tag"] = fw_tag_match.group(1).decode("ascii", "replace")

info["zephyr_tag"] = fw_tag_match.group(2).decode("ascii", "replace")

except Exception:

pass

# 5. ALSA Topology Manifest (CoSA)

pos_cosa = data.find(b"CoSA")

if pos_cosa != -1:

try:

offset = pos_cosa

while offset + 36 <= len(data):

if data[offset:offset+4] != b"CoSA":

break

magic, abi, ver, ttype, sz, vendor_type, payload_sz, idx, count = struct.unpack_from("<IIIIIIIII", data, offset)

if ttype == 8: # MANIFEST

man_data = data[offset+36 : offset+36+payload_sz]

if len(man_data) >= 112:

priv_sz = struct.unpack_from("<I", man_data, 108)[0]

priv_data = man_data[112:112+priv_sz]

if len(priv_data) >= 6:

abi_maj, abi_min, abi_patch = struct.unpack_from("<HHH", priv_data, 0)

info["tplg_abi"] = f"{abi_maj}.{abi_min}.{abi_patch}"

break

offset += 36 + payload_sz

except Exception:

pass

return info

def format_manifest_summary(man: Optional[Dict[str, Any]]) -> str:

"""Format manifest dictionary into a concise single-line description."""

if not man:

return ""

parts = []

if man.get("manifest_version"):

parts.append(f"ver {man['manifest_version']}")

if man.get("build_date"):

dt = man["build_date"]

if man.get("build_time"):

dt += f" {man['build_time']}"

parts.append(f"built {dt}")

if man.get("tplg_abi"):

parts.append(f"ABI {man['tplg_abi']}")

elif man.get("abi_version"):

parts.append(f"ABI {man['abi_version']}")

if man.get("sof_tag") and man.get("sof_tag") != man.get("manifest_version"):

parts.append(f"tag {man['sof_tag']}")

if man.get("zephyr_build"):

parts.append(f"zephyr {man['zephyr_build']}")

return ", ".join(parts)

def parse_version_tuple(ver_str: Optional[str]) -> Optional[Tuple[int, ...]]:

"""Parse version string like 'v2.14.1', '2.14', 'v2.14.1.1', 'v2.14.1 (manifest)' into an integer tuple."""

if not ver_str or ver_str in ("-", "unknown", "broken", "broken link", "target mismatch", "error", "(missing)"):

return None

m = re.search(r"v?(\d+(?:\.\d+)*)", ver_str)

if m:

nums = m.group(1).split(".")

return tuple(int(n) for n in nums if n.isdigit())

nums = re.findall(r"\d+", ver_str)

if nums:

return tuple(int(n) for n in nums)

return None

class SofBinRepo:

"""Interface to inspect the sof-bin repository."""

def __init__(self, repo_dir: Path):

self.repo_dir = repo_dir.resolve()

if not self.repo_dir.is_dir():

raise FileNotFoundError(f"sof-bin repository directory not found: {self.repo_dir}")

self._md5_db: Optional[Dict[str, List[Dict[str, str]]]] = None

self._filename_md5_cache: Dict[str, Dict[str, List[Dict[str, str]]]] = {}

@classmethod

def find_repo(cls, hint_dir: Optional[str] = None) -> "SofBinRepo":

"""Auto-detect the sof-bin repository location."""

candidates = []

if hint_dir:

candidates.append(Path(hint_dir))

if "SOF_BIN_DIR" in os.environ:

candidates.append(Path(os.environ["SOF_BIN_DIR"]))

# Script location

script_dir = Path(__file__).resolve().parent

candidates.append(script_dir)

if (script_dir / "sof-bin").is_dir():

candidates.append(script_dir / "sof-bin")

# Standard work locations

home = Path.home()

candidates.extend([

home / "work" / "sof-ptl" / "sof-bin",

home / "work" / "sof-tgl" / "sof-bin",

home / "work" / "sof-arl" / "sof-bin",

home / "work" / "sof-bin",

Path.cwd(),

Path.cwd() / "sof-bin",

])

for c in candidates:

if c.is_dir() and (cls._is_sof_bin_dir(c)):

return cls(c)

raise FileNotFoundError(

"Could not locate sof-bin repository. Please specify --sof-bin-dir."

)

@classmethod

def _is_sof_bin_dir(cls, path: Path) -> bool:

"""Check if a directory contains SOF release directories (e.g. v2.14.x, v2.12.x)."""

has_version_dirs = any(

d.is_dir() and (re.match(r"^v?\d+\.\d+", d.name) or d.name.startswith("v20"))

for d in path.iterdir()

)

return has_version_dirs

def get_version_dirs(self) -> List[Path]:

"""Return sorted list of version directories in the repo (newest first)."""

dirs = [

d for d in self.repo_dir.iterdir()

if d.is_dir() and (re.match(r"^v?\d+\.\d+", d.name) or d.name.startswith("v20"))

]

def sort_key(p: Path) -> Tuple[int, ...]:

nums = re.findall(r"\d+", p.name)

return tuple(int(n) for n in nums) if nums else (0,)

return sorted(dirs, key=sort_key, reverse=True)

def list_available_versions(self) -> List[Dict[str, Any]]:

"""Return metadata for all available releases in sof-bin."""

versions = []

for vdir in self.get_version_dirs():

subdirs = [p.name for p in vdir.iterdir() if p.is_dir()]

fw_sub = [s for s in subdirs if s.startswith(("sof-ipc4-v", "sof-ipc3-zephyr-v", "sof-v"))]

tplg_sub = [s for s in subdirs if s.startswith(("sof-ipc4-tplg-v", "sof-ace-tplg-v", "sof-tplg-v"))]

lib_sub = [s for s in subdirs if s.startswith("sof-ipc4-lib-v")]

tools_sub = [s for s in subdirs if s.startswith("tools-v")]

fw_version = fw_sub[0] if fw_sub else None

tplg_version = tplg_sub[0] if tplg_sub else None

lib_version = lib_sub[0] if lib_sub else None

platforms = set()

for root, _, files in os.walk(vdir):

for f in files:

if f.endswith(".ri"):

m = re.search(r"sof-([a-z0-9_]+)\.ri", f)

if m:

platforms.add(m.group(1))

versions.append({

"version_dir": vdir.name,

"directory": vdir.name,

"path": str(vdir),

"fw_component": fw_version,

"tplg_component": tplg_version,

"lib_component": lib_version,

"tools_component": tools_sub[0] if tools_sub else None,

"platforms": sorted(platforms),

})

return versions

def resolve_version_components(self, target_version: str) -> Dict[str, Path]:

"""

Given a version string (e.g. 'v2.14.1', '2.14', 'V2.14.1', 'v2.12', 'v2.2.2'),

locate the matching component directories in sof-bin.

"""

norm_ver = target_version.strip().lower()

if not norm_ver.startswith("v") and not norm_ver.startswith("20"):

norm_ver = "v" + norm_ver

vdir: Optional[Path] = None

m = re.match(r"^v?(\d+)\.(\d+)(\.|\b|-)", norm_ver)

if m:

expected_vdir = f"v{m.group(1)}.{m.group(2)}.x"

candidate = self.repo_dir / expected_vdir

if candidate.is_dir():

vdir = candidate

if not vdir:

for d in self.get_version_dirs():

d_name_lower = d.name.lower()

if norm_ver in d_name_lower or any(norm_ver in p.name.lower() for p in d.iterdir()):

vdir = d

break

if not vdir:

for d in self.get_version_dirs():

if d.name.lower().startswith(norm_ver):

vdir = d

break

if not vdir or not vdir.is_dir():

available = [d.name for d in self.get_version_dirs()]

raise ValueError(

f"Version '{target_version}' not found in sof-bin. Available major versions: {', '.join(available)}"

)

subdirs = [p for p in vdir.iterdir() if p.is_dir()]

def find_best_match(comp_name: str, prefixes: Tuple[str, ...], ver: str) -> Optional[Path]:

matching = [p for p in subdirs if p.name.lower().startswith(prefixes)]

if not matching:

return None

for p in matching:

p_lower = p.name.lower()

if p_lower.endswith("-" + ver) or p_lower.endswith(ver):

return p

ver_clean = ver.lstrip("v")

for p in matching:

p_lower = p.name.lower()

if p_lower.endswith("-" + ver_clean) or p_lower.endswith(ver_clean):

return p

def ver_key(p: Path) -> List[int]:

nums = re.findall(r"\d+", p.name)

return [int(n) for n in nums]

newest = sorted(matching, key=ver_key)[-1]

sys.stderr.write(

f"Note: Component '{comp_name}' exact version '{ver}' not found in {vdir.name}; using newest available subversion '{newest.name}'\n"

)

return newest

fw_dir = find_best_match("fw", ("sof-ipc4-v", "sof-ipc3-zephyr-v", "sof-v"), norm_ver)

llext_dir = find_best_match("llext", ("sof-ipc4-lib-v",), norm_ver)

tplg_dir = find_best_match("tplg", ("sof-ipc4-tplg-v", "sof-ace-tplg-v", "sof-tplg-v"), norm_ver)

tools_dir = find_best_match("tools", ("tools-v",), norm_ver)

components: Dict[str, Path] = {"vdir": vdir}

if fw_dir:

components["fw"] = fw_dir

if llext_dir:

components["llext"] = llext_dir

if tplg_dir:

components["tplg"] = tplg_dir

if tools_dir:

components["tools"] = tools_dir

return components

def build_global_md5_database(self) -> Dict[str, List[Dict[str, str]]]:

"""

Build an index of all regular files in all version directories by MD5 checksum.

Returns: md5_hash -> list of {version_dir, component, filename, rel_path}

"""

if self._md5_db is not None:

return self._md5_db

db: Dict[str, List[Dict[str, str]]] = {}

for vdir in self.get_version_dirs():

for root, _, files in os.walk(vdir):

for f in files:

file_p = Path(root) / f

if file_p.is_file() and not file_p.is_symlink():

md5_val = compute_file_md5(file_p)

if md5_val:

rel_to_vdir = file_p.relative_to(vdir)

comp_name = rel_to_vdir.parts[0] if rel_to_vdir.parts else "root"

entry = {

"version_dir": vdir.name,

"component": comp_name,

"filename": f,

"rel_path": str(rel_to_vdir),

"full_path": str(file_p),

}

if md5_val not in db:

db[md5_val] = []

db[md5_val].append(entry)

self._md5_db = db

return db

def lookup_md5(self, md5_hash: Optional[str], target_filename: Optional[str] = None) -> Optional[str]:

"""

Look up an MD5 hash across all releases and return a human-readable release identifier.

Uses fast on-demand caching per target filename to avoid scanning the entire repository.

"""

if not md5_hash:

return None

clean_hash = md5_hash.lower()

if target_filename:

if target_filename not in self._filename_md5_cache:

fn_db: Dict[str, List[Dict[str, str]]] = {}

for vdir in self.get_version_dirs():

for root, _, files in os.walk(vdir):

if target_filename in files:

file_p = Path(root) / target_filename

if file_p.is_file() and not file_p.is_symlink():

h = compute_file_md5(file_p)

if h:

rel_to_vdir = file_p.relative_to(vdir)

comp_name = rel_to_vdir.parts[0] if rel_to_vdir.parts else "root"

entry = {

"version_dir": vdir.name,

"component": comp_name,

"filename": target_filename,

"rel_path": str(rel_to_vdir),

"full_path": str(file_p),

}

fn_db.setdefault(h, []).append(entry)

self._filename_md5_cache[target_filename] = fn_db

matches = self._filename_md5_cache[target_filename].get(clean_hash)

if not matches and self._md5_db is not None:

matches = [m for m in self._md5_db.get(clean_hash, []) if m["filename"] == target_filename]

else:

db = self.build_global_md5_database()

matches = db.get(clean_hash)

if not matches:

return None

descriptions = []

for m in matches:

comp = m["component"]

ver_match = re.search(r"v\d+(\.\d+)*(-[a-zA-Z0-9]+)?", comp)

ver_label = ver_match.group(0) if ver_match else m["version_dir"]

descriptions.append(f"{ver_label} ({comp})")

return ", ".join(dict.fromkeys(descriptions))

def find_topology_upgrade(

self,

filename: str,

installed_version: Optional[str] = None,

installed_md5: Optional[str] = None,

fw_version: Optional[str] = None,

) -> Optional[Dict[str, Any]]:

"""

Find if a topology file can be upgraded from sof-bin to match installed FW version or newest available release.

Returns upgrade info dict if an upgrade is available, else None.

"""

candidates: List[Dict[str, Any]] = []

target_vdirs = self.get_version_dirs()

if fw_version:

norm_fw = fw_version.lstrip("v")

parts = norm_fw.split(".")

major_minor_prefix = f"v{parts[0]}.{parts[1]}" if len(parts) >= 2 else f"v{parts[0]}"

matching_vdirs = [v for v in target_vdirs if v.name.startswith(major_minor_prefix)]

if matching_vdirs:

target_vdirs = matching_vdirs + [v for v in target_vdirs if v not in matching_vdirs]

for vdir in target_vdirs:

for item in vdir.iterdir():

if item.is_dir() and item.name.startswith(("sof-ipc4-tplg", "sof-ace-tplg", "sof-tplg")):

file_p = item / filename

if file_p.is_file():

m = re.search(r"v\d+(\.\d+)*(-[a-zA-Z0-9]+)?", item.name)

cand_ver = m.group(0) if m else vdir.name

cand_tup = parse_version_tuple(cand_ver)

md5_val = compute_file_md5(file_p)

candidates.append({

"version_dir": vdir.name,

"component": item.name,

"version": cand_ver,

"version_tuple": cand_tup or (0,),

"filename": filename,

"rel_path": f"sof-ipc4-tplg/{filename}" if item.name.startswith("sof-ipc4") else f"sof-tplg/{filename}",

"full_path": str(file_p),

"md5": md5_val,

})

if not candidates:

return None

# Sort candidates by version tuple descending

candidates.sort(key=lambda x: (x["version_tuple"], x["version"]), reverse=True)

best_cand = candidates[0]

# If installed file already matches the best candidate checksum, no upgrade needed

if installed_md5 and best_cand["md5"] and installed_md5.lower() == best_cand["md5"].lower():

return None

inst_tup = parse_version_tuple(installed_version)

cand_tup = best_cand["version_tuple"]

is_newer = False

if inst_tup and cand_tup:

max_len = max(len(inst_tup), len(cand_tup), 3)

norm_inst = inst_tup + (0,) * (max_len - len(inst_tup))

norm_cand = cand_tup + (0,) * (max_len - len(cand_tup))

if norm_cand > norm_inst:

is_newer = True

elif norm_cand == norm_inst and installed_version != best_cand["version"]:

is_newer = True

elif not inst_tup or installed_version in ("-", "unknown", "broken link"):

is_newer = True

if is_newer:

reason = f"Upgrade available to match FW {fw_version}" if fw_version else "Newer topology available in sof-bin"

return {

"target_version": best_cand["version"],

"component": best_cand["component"],

"filename": filename,

"rel_path": best_cand["rel_path"],

"source_path": best_cand["full_path"],

"md5": best_cand["md5"],

"fw_version": fw_version,

"reason": reason,

}

return None

class HostTarget:

"""Interface to read, check, and fix files on target host (local or remote)."""

def is_remote(self) -> bool:

return False

def batch_query(self, paths: List[str]) -> Dict[str, Tuple[bool, bool, Optional[int], Optional[str], Optional[str], Optional[Dict[str, Any]]]]:

raise NotImplementedError

def fix_file(

self,

source_path: str,

target_path: str,

is_symlink: bool,

symlink_target: Optional[str],

dry_run: bool = False,

) -> Tuple[bool, str]:

raise NotImplementedError

def scan_installed_files(self, fw_dest: str = "/lib/firmware/intel", tools_dest: str = "/usr/local/bin") -> List[Dict[str, Any]]:

raise NotImplementedError

class LocalHostTarget(HostTarget):

"""Local filesystem host target."""

def __init__(self, target_root: Optional[Path] = None):

self.target_root = target_root.resolve() if target_root else None

def _resolve(self, path: str) -> Path:

p = Path(path)

if self.target_root and p.is_absolute():

rel = str(p).lstrip("/")

return self.target_root / rel

elif self.target_root:

return self.target_root / p

return p

def is_remote(self) -> bool:

return False

def batch_query(self, paths: List[str]) -> Dict[str, Tuple[bool, bool, Optional[int], Optional[str], Optional[str], Optional[Dict[str, Any]]]]:

results = {}

for p_str in paths:

p = self._resolve(p_str)

is_sym = p.is_symlink()

exists = p.exists() or is_sym

size = None

target = None

md5_val = None

man_info = None

if is_sym:

try:

target = os.readlink(p)

except OSError:

target = None

try:

resolved = p.resolve()

if resolved.is_file():

size = resolved.stat().st_size

raw_data = resolved.read_bytes()

md5_val = hashlib.md5(raw_data).hexdigest()

man_info = parse_binary_manifest(raw_data)

except OSError:

pass

elif exists and p.is_file():

try:

size = p.stat().st_size

raw_data = p.read_bytes()

md5_val = hashlib.md5(raw_data).hexdigest()

man_info = parse_binary_manifest(raw_data)

except OSError:

pass

results[p_str] = (exists, is_sym, size, target, md5_val, man_info)

return results

def fix_file(

self,

source_path: str,

target_path: str,

is_symlink: bool,

symlink_target: Optional[str],

dry_run: bool = False,

) -> Tuple[bool, str]:

real_tgt = self._resolve(target_path)

if dry_run:

if is_symlink:

return True, f"[DRY-RUN] Would create symlink {real_tgt} -> {symlink_target}"

return True, f"[DRY-RUN] Would copy {source_path} -> {real_tgt}"

try:

real_tgt.parent.mkdir(parents=True, exist_ok=True)

if real_tgt.exists() or real_tgt.is_symlink():

real_tgt.unlink()

if is_symlink:

if not symlink_target:

return False, f"Cannot create symlink at {real_tgt}: missing symlink target"

os.symlink(symlink_target, real_tgt)

return True, f"Created symlink {real_tgt} -> {symlink_target}"

else:

shutil.copy2(source_path, real_tgt)

return True, f"Copied {source_path} -> {real_tgt}"

except PermissionError:

return False, f"Permission denied writing to {real_tgt}. Please re-run with sudo or check permissions."

except Exception as e:

return False, f"Error fixing {real_tgt}: {e}"

def scan_installed_files(self, fw_dest: str = "/lib/firmware/intel", tools_dest: str = "/usr/local/bin") -> List[Dict[str, Any]]:

results = []

real_fw = self._resolve(fw_dest)

if real_fw.is_dir():

for root, _, files in os.walk(real_fw):

for name in files:

p = Path(root) / name

rel = str(p.relative_to(real_fw))

is_sym = p.is_symlink()

sym_target = os.readlink(p) if is_sym else None

md5_val = None

size = None

man_info = None

if is_sym:

try:

resolved = p.resolve()

if resolved.is_file():

size = resolved.stat().st_size

raw_data = resolved.read_bytes()

md5_val = hashlib.md5(raw_data).hexdigest()

man_info = parse_binary_manifest(raw_data)

except OSError:

pass

elif p.is_file():

try:

size = p.stat().st_size

raw_data = p.read_bytes()

md5_val = hashlib.md5(raw_data).hexdigest()

man_info = parse_binary_manifest(raw_data)

except OSError:

pass

results.append({

"path_type": "fw",

"rel_path": rel,

"full_path": f"{fw_dest.rstrip('/')}/{rel}",

"filename": name,

"is_symlink": is_sym,

"symlink_target": sym_target,

"size": size,

"md5": md5_val,

"manifest": man_info,

})

real_tools = self._resolve(tools_dest)

if real_tools.is_dir():

for name in os.listdir(real_tools):

p = real_tools / name

if name.startswith(("sof-", "mtrace-")) or name in ("sof-logger", "sof-probes", "sof-coredump", "sof-ctl"):

is_sym = p.is_symlink()

sym_target = os.readlink(p) if is_sym else None

md5_val = None

size = None

man_info = None

if is_sym:

try:

resolved = p.resolve()

if resolved.is_file():

size = resolved.stat().st_size

raw_data = resolved.read_bytes()

md5_val = hashlib.md5(raw_data).hexdigest()

man_info = parse_binary_manifest(raw_data)

except OSError:

pass

elif p.is_file():

try:

size = p.stat().st_size

raw_data = p.read_bytes()

md5_val = hashlib.md5(raw_data).hexdigest()

man_info = parse_binary_manifest(raw_data)

except OSError:

pass

results.append({

"path_type": "tools",

"rel_path": f"tools/{name}",

"full_path": f"{tools_dest.rstrip('/')}/{name}",

"filename": name,

"is_symlink": is_sym,

"symlink_target": sym_target,

"size": size,

"md5": md5_val,

"manifest": man_info,

})

return results

def extract_deb(deb_path: Path, dest_dir: Path) -> Dict[str, Any]:

"""Extract Debian package (.deb) into dest_dir and return package metadata."""

meta: Dict[str, Any] = {"package_type": "deb", "file": deb_path.name, "path": str(deb_path)}

extracted = False

if shutil.which("dpkg-deb"):

try:

subprocess.run(["dpkg-deb", "-x", str(deb_path), str(dest_dir)], check=True, capture_output=True)

extracted = True

except subprocess.SubprocessError:

extracted = False

if not extracted:

data = deb_path.read_bytes()

if not data.startswith(b"!<arch>\n"):

raise ValueError(f"Invalid Debian package (missing ar magic): {deb_path}")

pos = 8

members: Dict[str, bytes] = {}

while pos + 60 <= len(data):

hdr = data[pos : pos + 60]

name = hdr[:16].decode("ascii", "replace").strip().rstrip("/")

size_str = hdr[48:58].decode("ascii", "replace").strip()

size = int(size_str) if size_str.isdigit() else 0

member_data = data[pos + 60 : pos + 60 + size]

members[name] = member_data

pos += 60 + size + (1 if size % 2 == 1 else 0)

data_arch_name = next((k for k in members if k.startswith("data.tar")), None)

if not data_arch_name:

raise ValueError(f"No data.tar member found in Debian package: {deb_path}")

data_payload = members[data_arch_name]

try:

with tarfile.open(fileobj=io.BytesIO(data_payload), mode="r:*") as tar:

tar.extractall(dest_dir)

except Exception:

p = subprocess.Popen(["tar", "-xf", "-", "-C", str(dest_dir)], stdin=subprocess.PIPE)

p.communicate(input=data_payload)

if p.returncode != 0:

raise RuntimeError(f"Failed to extract {data_arch_name} from {deb_path}")

ctrl_arch_name = next((k for k in members if k.startswith("control.tar")), None)

if ctrl_arch_name:

try:

with tarfile.open(fileobj=io.BytesIO(members[ctrl_arch_name]), mode="r:*") as tar:

for member in tar.getmembers():

if member.name.endswith("control") or member.name == "control":

f = tar.extractfile(member)

if f:

for line in f.read().decode("utf-8", "replace").splitlines():

if ":" in line:

k, v = line.split(":", 1)

meta[k.strip().lower()] = v.strip()

except Exception:

pass

if "package" not in meta and shutil.which("dpkg-deb"):

try:

res = subprocess.run(["dpkg-deb", "-f", str(deb_path)], capture_output=True, text=True)

for line in res.stdout.splitlines():

if ":" in line:

k, v = line.split(":", 1)

meta[k.strip().lower()] = v.strip()

except Exception:

pass

return meta

def extract_rpm(rpm_path: Path, dest_dir: Path) -> Dict[str, Any]:

"""Extract RPM package (.rpm) into dest_dir and return package metadata."""

meta: Dict[str, Any] = {"package_type": "rpm", "file": rpm_path.name, "path": str(rpm_path)}

extracted = False

if shutil.which("rpm2cpio") and shutil.which("cpio"):

try:

p1 = subprocess.Popen(["rpm2cpio", str(rpm_path)], stdout=subprocess.PIPE)

p2 = subprocess.Popen(["cpio", "-idmv"], stdin=p1.stdout, cwd=str(dest_dir), stdout=subprocess.PIPE, stderr=subprocess.PIPE)

if p1.stdout:

p1.stdout.close()

p2.communicate()

p1.wait()

if p2.returncode == 0:

extracted = True

except Exception:

extracted = False

if not extracted and shutil.which("rpm2archive") and shutil.which("tar"):

try:

with open(rpm_path, "rb") as fh:

p1 = subprocess.Popen(["rpm2archive", "-n"], stdin=fh, stdout=subprocess.PIPE)

p2 = subprocess.Popen(["tar", "-C", str(dest_dir), "-xf", "-"], stdin=p1.stdout)

if p1.stdout:

p1.stdout.close()

p2.communicate()

p1.wait()

if p2.returncode == 0:

extracted = True

except Exception:

extracted = False

if not extracted:

raise RuntimeError(f"Could not extract RPM package {rpm_path}: requires rpm2cpio + cpio or rpm2archive + tar")

if shutil.which("rpm"):

try:

res = subprocess.run(

["rpm", "-qp", "--queryformat", "%{NAME}\n%{VERSION}\n%{RELEASE}\n%{ARCH}\n%{SUMMARY}", str(rpm_path)],

capture_output=True,

text=True,

)

lines = res.stdout.strip().splitlines()

if len(lines) >= 4:

meta["package"] = lines[0].strip()

meta["version"] = f"{lines[1].strip()}-{lines[2].strip()}"

meta["architecture"] = lines[3].strip()

if len(lines) >= 5:

meta["description"] = lines[4].strip()

except Exception:

pass

return meta

def extract_package(pkg_path: Path, dest_dir: Path) -> Dict[str, Any]:

"""Auto-detect package format (.deb or .rpm) and extract into dest_dir."""

try:

with open(pkg_path, "rb") as f:

magic = f.read(8)

except OSError as e:

raise FileNotFoundError(f"Cannot read package file {pkg_path}: {e}")

if magic.startswith(b"!<arch>\n") or pkg_path.suffix == ".deb":

return extract_deb(pkg_path, dest_dir)

elif magic.startswith(b"\xed\xab\xee\xdb") or pkg_path.suffix == ".rpm":

return extract_rpm(pkg_path, dest_dir)

else:

raise ValueError(f"Unsupported package format: {pkg_path.name} (magic: {magic[:4]!r})")

def collect_packages(pkg_args: List[str]) -> List[Path]:

"""Collect and resolve list of package paths from arguments (handles directories and commas)."""

collected = []

for item in pkg_args:

for p_str in item.split(","):

p_str = p_str.strip()

if not p_str:

continue

p = Path(p_str).resolve()

if p.is_dir():

found = sorted(list(p.glob("*.deb")) + list(p.glob("*.rpm")))

if not found:

raise FileNotFoundError(f"No .deb or .rpm packages found in directory: {p}")

collected.extend(found)

elif p.is_file():

collected.append(p)

else:

raise FileNotFoundError(f"Package file or directory not found: {p_str}")

return collected

class PackageTarget(HostTarget):

"""Target host backed by extracted .deb or .rpm packages in a managed temporary directory."""

def __init__(self, packages: List[Path], temp_dir: Optional[tempfile.TemporaryDirectory] = None):

self.packages = [Path(p).resolve() for p in packages]

self._temp_dir_obj = temp_dir or tempfile.TemporaryDirectory(prefix="sof_pkg_")

self.target_root = Path(self._temp_dir_obj.name)

self.package_metadata: List[Dict[str, Any]] = []

self._delegate = LocalHostTarget(target_root=self.target_root)

self._extract_all()

def _extract_all(self):

for pkg in self.packages:

if not pkg.is_file():

raise FileNotFoundError(f"Package file not found: {pkg}")