blob: 2c40ef4e952458ffa1672771ee378b9d63c017e8 [file]
#!/usr/bin/env fuchsia-vendored-python
# Copyright 2026 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Linux-specific host hardware and filesystem capability profiler.
Outputs a structured, single-record JSON snapshot of mounts, disks, CPUs, RAM, and network interfaces.
Security & PII Boundaries:
- Workstation usernames in directory/mount paths are automatically sanitized and redacted with '$USER'.
- Sensitive network configurations—including physical MAC addresses and IP addresses—are completely
unread and never collected by this utility.
"""
import argparse
import getpass
import json
import os
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Any, Iterator, Optional
# --- Immutable Dataclass Data Structures representing parsed states ---
@dataclass(frozen=True, slots=True)
class MountInfo:
"""Represents file system mount configuration details.
Attributes:
device: The source block device or virtual filesystem (e.g., '/dev/sda1').
mount_point: The path where the filesystem is mounted (e.g., '/').
fstype: The filesystem type (e.g., 'ext4', 'tmpfs').
options: The active comma-separated mount flags (e.g., 'rw,noatime').
"""
device: str
mount_point: str
fstype: str
options: str
def sanitize(self, username: str) -> "MountInfo":
"""Returns a new MountInfo with matching path components redacted."""
if not username:
return self
return replace(
self,
device=_sanitize_path(self.device, username),
mount_point=_sanitize_path(self.mount_point, username),
)
def to_dict(self) -> dict[str, str]:
"""Converts the mount info record to a clean JSON-serializable dictionary."""
return {
"device": self.device,
"mount_point": self.mount_point,
"fstype": self.fstype,
"options": self.options,
}
@dataclass(frozen=True, slots=True)
class DiskInfo:
"""Represents a physical or virtual storage block device's properties.
Attributes:
rotational: True if the disk is a spinning platter HDD, False if SSD/NVMe.
size_gb: The storage capacity of the disk in decimal Gigabytes.
"""
rotational: Optional[bool] = None
size_gb: Optional[float] = None
def to_dict(self) -> dict[str, Any]:
"""Converts the disk info record to a clean JSON-serializable dictionary."""
res: dict[str, Any] = {}
if self.rotational is not None:
res["rotational"] = self.rotational
if self.size_gb is not None:
res["size_gb"] = self.size_gb
return res
@dataclass(frozen=True, slots=True)
class CpuInfo:
"""Represents the host processor hardware specs.
Attributes:
model_name: The brand/model name of the CPU (e.g., 'AMD EPYC 7B12').
logical_processors: The total count of active logical threads/cores.
physical_cores: The count of distinct physical cores on the CPU.
"""
model_name: str
logical_processors: int
physical_cores: int
def to_dict(self) -> dict[str, Any]:
"""Converts the CPU info record to a clean JSON-serializable dictionary."""
return {
"model_name": self.model_name,
"logical_processors": self.logical_processors,
"physical_cores": self.physical_cores,
}
@dataclass(frozen=True, slots=True)
class MemoryInfo:
"""Represents the total physical system memory available on the host.
Attributes:
total_ram_gb: Total installed RAM capacity in Gigabytes.
"""
total_ram_gb: float
def to_dict(self) -> dict[str, Any]:
"""Converts the memory info record to a clean JSON-serializable dictionary."""
return {
"total_ram_gb": self.total_ram_gb,
}
@dataclass(frozen=True, slots=True)
class NetworkInterfaceInfo:
"""Represents an active or configured network interface card on the host.
Attributes:
name: The interface device name (e.g., 'eth0', 'wlan0').
operstate: The current operational state (e.g., 'up', 'down').
speed_mbps: Optional active link bandwidth capacity in Megabits per second.
"""
name: str
operstate: str
speed_mbps: Optional[int] = None
def to_dict(self) -> dict[str, Any]:
"""Converts the network interface info to a clean JSON-serializable dictionary."""
res: dict[str, Any] = {
"name": self.name,
"operstate": self.operstate,
}
if self.speed_mbps is not None:
res["speed_mbps"] = self.speed_mbps
return res
# --- System Hardware Profiling Helpers ---
def _sanitize_path(path_str: str, username: str) -> str:
"""Replaces any directory path component exactly equal to the username with $USER."""
if not username or not path_str:
return path_str
# Retrieve the path parts natively (e.g. ('/', 'home', 'fangism'))
parts = Path(path_str).parts
sanitized_parts = ["$USER" if p == username else p for p in parts]
return str(Path(*sanitized_parts))
def _is_workspace_mount(mount_point: str, workspace_dir: str) -> bool:
"""Checks if a mount point is a parent directory of or equals the workspace directory.
Examples:
workspace_dir = "/home/fuchsia/src"
_is_workspace_mount("/", workspace_dir) -> True
_is_workspace_mount("/home", workspace_dir) -> True
_is_workspace_mount("/home/fuchsia/src", workspace_dir) -> True
_is_workspace_mount("/run", workspace_dir) -> False
_is_workspace_mount("/hom", workspace_dir) -> False (prefix matches, but not directory-bound)
"""
if workspace_dir == mount_point:
return True
# Enforce trailing slash to guarantee directory-level matching
# (prevents "/hom" from falsely matching "/home/fuchsia")
prefix = mount_point if mount_point.endswith("/") else mount_point + "/"
return workspace_dir.startswith(prefix)
def _parse_mounts_content(
content: str, workspace_dir: str
) -> Iterator[MountInfo]:
"""Purely parses the raw /proc/mounts content string.
Filters and yields MountInfo records that are parent to or equal the workspace_dir.
Assumes the standard Linux /proc/mounts format, which is whitespace-separated
and maps columns sequentially as follows:
Col 0: Device node or virtual filesystem source (e.g. '/dev/sda1' or 'tmpfs')
Col 1: Mount point absolute path (e.g. '/' or '/home/fuchsia')
Col 2: Filesystem type identifier (e.g. 'ext4' or 'tmpfs')
Col 3: Comma-separated mount options flags (e.g. 'rw,noatime,discard')
Col 4: Dummy integer for dump (usually '0')
Col 5: Dummy integer for fsck pass (usually '0')
Example /proc/mounts line:
/dev/sda1 / ext4 rw,noatime,discard,errors=remount-ro 0 0
"""
for line in content.splitlines():
parts = line.split()
if len(parts) >= 4:
mount_point = parts[1]
if _is_workspace_mount(mount_point, workspace_dir):
yield MountInfo(
device=parts[0],
mount_point=mount_point,
fstype=parts[2],
options=parts[3],
)
def _parse_mount_options(
proc_dir: str, workspace_dir: str
) -> Iterator[MountInfo]:
"""Reads mounts from proc_dir and delegates to _parse_mounts_content."""
proc_mounts = Path(proc_dir) / "mounts"
if not proc_mounts.exists():
return
try:
content = proc_mounts.read_text(encoding="utf-8")
yield from _parse_mounts_content(content, workspace_dir)
except OSError:
pass
def _query_block_device_info(dev_dir: Path) -> DiskInfo:
"""Reads size and rotational state of a block device, failing gracefully.
Expected file formats and metrics:
1. dev_dir / "queue" / "rotational":
- Content: "0" (SSD / Solid State / NVMe / Non-rotational)
- Content: "1" (HDD / Magnetic rotational hard drive)
2. dev_dir / "size":
- Content: size in 512-byte sectors as an integer string (e.g., "488281250" for 250 GB)
"""
rotational = None
rot_file = dev_dir / "queue" / "rotational"
if rot_file.exists():
try:
# 1 = Magnetic HDD, 0 = Solid State SSD/NVMe
rotational = rot_file.read_text(encoding="utf-8").strip() == "1"
except OSError:
pass
size_gb = None
size_file = dev_dir / "size"
if size_file.exists():
try:
raw_size = size_file.read_text(encoding="utf-8").strip()
# Size is stored in 512-byte sectors; convert to GB
size_gb = round((int(raw_size) * 512) / 1e9, 2)
except OSError:
pass
except ValueError:
pass
return DiskInfo(rotational=rotational, size_gb=size_gb)
def _parse_disk_properties(sys_dir: str) -> dict[str, DiskInfo]:
"""Scans and extracts properties for physical and virtual block devices.
Scans directories inside sys_dir/block representing active storage devices:
- SCSI/SATA drives: "sda", "sdb", etc.
- NVMe SSD drives: "nvme0n1", "nvme1n1", etc.
- Virtualised VirtIO partitions: "vda", "vdb", "xvda", etc.
"""
disks: dict[str, DiskInfo] = {}
sys_block = Path(sys_dir) / "block"
if not sys_block.exists():
return disks
# Match physical SCSI (sd), NVMe (nvme), and virtualised VirtIO (vd / xvd) devices
valid_prefixes = ("sd", "nvme", "vd", "xvd")
try:
for dev in sys_block.iterdir():
if dev.name.startswith(valid_prefixes):
try:
info = _query_block_device_info(dev)
# Only append if we successfully collected at least some info
if info.rotational is not None or info.size_gb is not None:
disks[dev.name] = info
except OSError:
pass
except OSError:
pass
return disks
def _parse_cpuinfo_content(content: str) -> CpuInfo:
"""Purely parses /proc/cpuinfo content and returns a CpuInfo record.
Example /proc/cpuinfo snippet:
processor : 0
model name : Intel(R) Xeon(R) CPU @ 2.20GHz
cpu cores : 2
"""
model_name = "Unknown"
logical_processors = 0
physical_cores = 0
for line in content.splitlines():
if ":" in line:
key, val = [parts.strip() for parts in line.split(":", 1)]
if key == "processor":
logical_processors += 1
elif key == "model name" and model_name == "Unknown":
model_name = val
elif key == "cpu cores" and physical_cores == 0:
try:
physical_cores = int(val)
except ValueError:
pass
# Fallback in case cpu cores isn't reported on some architectures (like virtual VMs)
if physical_cores == 0:
physical_cores = logical_processors
return CpuInfo(
model_name=model_name,
logical_processors=logical_processors,
physical_cores=physical_cores,
)
def _get_cpu_info(proc_dir: str) -> CpuInfo:
"""Reads /proc/cpuinfo and returns a CpuInfo record."""
proc_cpuinfo = Path(proc_dir) / "cpuinfo"
if not proc_cpuinfo.exists():
return CpuInfo("Unknown", 1, 1)
try:
content = proc_cpuinfo.read_text(encoding="utf-8")
return _parse_cpuinfo_content(content)
except OSError:
return CpuInfo("Unknown", 1, 1)
def _parse_meminfo_total(content: str) -> MemoryInfo:
"""Purely parses MemTotal from /proc/meminfo content.
Example /proc/meminfo line:
MemTotal: 65873920 kB
"""
total_ram_gb = 0.0
for line in content.splitlines():
if line.startswith("MemTotal:"):
parts = line.split()
if len(parts) >= 2:
try:
# Convert kB to bytes, then to GB
total_ram_gb = round((int(parts[1]) * 1024) / 1e9, 2)
except ValueError:
pass
break
return MemoryInfo(total_ram_gb=total_ram_gb)
def _get_system_memory_info(proc_dir: str) -> MemoryInfo:
"""Reads /proc/meminfo and returns total physical memory."""
proc_meminfo = Path(proc_dir) / "meminfo"
if not proc_meminfo.exists():
return MemoryInfo(0.0)
try:
content = proc_meminfo.read_text(encoding="utf-8")
return _parse_meminfo_total(content)
except OSError:
return MemoryInfo(0.0)
def _query_net_interface_info(iface_dir: Path) -> NetworkInterfaceInfo:
"""Reads operstate and speed of a network interface, failing gracefully."""
operstate = "unknown"
oper_file = iface_dir / "operstate"
if oper_file.exists():
try:
operstate = oper_file.read_text(encoding="utf-8").strip()
except OSError:
pass
speed_mbps = None
speed_file = iface_dir / "speed"
if speed_file.exists():
try:
speed_mbps = int(speed_file.read_text(encoding="utf-8").strip())
except (OSError, ValueError):
pass
return NetworkInterfaceInfo(
name=iface_dir.name, operstate=operstate, speed_mbps=speed_mbps
)
def _parse_network_interfaces(sys_dir: str) -> Iterator[NetworkInterfaceInfo]:
"""Scans and extracts properties for physical and active network interfaces."""
sys_net = Path(sys_dir) / "class" / "net"
if not sys_net.exists():
return
try:
for iface in sys_net.iterdir():
# Skip loopback interface 'lo'
if iface.name == "lo":
continue
try:
yield _query_net_interface_info(iface)
except OSError:
pass
except OSError:
pass
def collect_hardware_profile(
proc_dir: str, sys_dir: str, workspace_dir: str, username: str
) -> dict[str, Any]:
"""Gathers a lightweight system, disk, CPU, RAM, and net snapshot dict."""
return {
"cpu": _get_cpu_info(proc_dir).to_dict(),
"memory": _get_system_memory_info(proc_dir).to_dict(),
"disks": {
k: v.to_dict() for k, v in _parse_disk_properties(sys_dir).items()
},
"network": [
net.to_dict() for net in _parse_network_interfaces(sys_dir)
],
"mounts": [
m.sanitize(username).to_dict()
for m in _parse_mount_options(proc_dir, workspace_dir)
],
}
def capture_hardware_profile(
output_path: Path,
proc_dir: str,
sys_dir: str,
workspace_dir: str,
username: str,
) -> None:
"""Collects and writes the hardware profile JSON to disk."""
try:
profile = collect_hardware_profile(
proc_dir, sys_dir, workspace_dir, username
)
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, "w") as f:
json.dump(profile, f, indent=2)
except Exception:
# Fail silently to guarantee we never interrupt the user's build
pass
def main() -> None:
"""Core command-line execution entry point for the utility."""
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--output",
type=Path,
required=True,
help="Path to write hardware profile JSON snapshot",
)
parser.add_argument(
"--proc-dir",
default="/proc",
help="Custom path for proc filesystem (for testing)",
)
parser.add_argument(
"--sys-dir",
default="/sys",
help="Custom path for sys filesystem (for testing)",
)
args = parser.parse_args()
try:
username = getpass.getuser()
except KeyError:
username = ""
capture_hardware_profile(
args.output,
proc_dir=args.proc_dir,
sys_dir=args.sys_dir,
workspace_dir=os.getcwd(),
username=username,
)
if __name__ == "__main__":
main()