diff --git a/plugins/rootfsmanager/README.md b/plugins/rootfsmanager/README.md new file mode 100644 index 00000000..25660f81 --- /dev/null +++ b/plugins/rootfsmanager/README.md @@ -0,0 +1,17 @@ +# rootfs Manager for Surfactant + +A plugin that manages `rootfs` files for Surfactant based on ELF metadata. + +## Quickstart + +In the same virtual environment that Surfactant was installed in, install this plugin with `pip install .`. +If pipx was used to install Surfactant, install this plugin with +`pipx inject surfactant git+https://github.com/LLNL/Surfactant#subdirectory=plugins/rootfsmanager`. + +For developers making changes to this plugin, install it with `pip install -e .`. + +## Uninstalling + +This plugin can be uninstalled with `pip uninstall surfactantplugin-rootfsmanager`. +If pipx was used, it can be uninstalled with +`pipx uninject surfactant surfactantplugin-rootfsmanager` diff --git a/plugins/rootfsmanager/pyproject.toml b/plugins/rootfsmanager/pyproject.toml new file mode 100644 index 00000000..0abf3d60 --- /dev/null +++ b/plugins/rootfsmanager/pyproject.toml @@ -0,0 +1,34 @@ +[build-system] +requires = ["setuptools>=77.0.3", "setuptools-scm"] +build-backend = "setuptools.build_meta" + +[project] +name = "rootfsmanager" +authors = [ + {name = "Kendall Harter", email = "harter8@llnl.gov"}, +] +description = "Automatic rootfs downloading based on ELF information." +readme = "README.md" +requires-python = ">=3.10" +keywords = ["surfactant"] +license = "MIT" +classifiers = [ + "Programming Language :: Python :: 3", + "Environment :: Console", + "Operating System :: MacOS", + "Operating System :: Microsoft :: Windows", + "Operating System :: POSIX :: Linux", +] +dependencies = [ + "surfactant", + "loguru==0.7.*", + "requests>=2.32.3", + "beautifulsoup4==4.*", +] +dynamic = ["version"] + +[project.entry-points."surfactant"] +"surfactantplugin_rootfsmanager" = "surfactantplugin_rootfsmanager" + +[tool.setuptools] +py-modules=["surfactantplugin_rootfsmanager"] diff --git a/plugins/rootfsmanager/surfactantplugin_rootfsmanager.py b/plugins/rootfsmanager/surfactantplugin_rootfsmanager.py new file mode 100644 index 00000000..54cb8bfb --- /dev/null +++ b/plugins/rootfsmanager/surfactantplugin_rootfsmanager.py @@ -0,0 +1,186 @@ +# Copyright 2026 Lawrence Livermore National Security, LLC +# See the top-level LICENSE file for details. +# +# SPDX-License-Identifier: MIT + +# This has a bunch of false positives (because := statements can't be combined) +# so disable it +# ruff: noqa SIM102 + +# Currently using Ubuntu downloads at: https://partner-images.canonical.com/oci/ +# Another possible source (for other distros): https://images.linuxcontainers.org/ +# See also: https://github.com/llnl/Surfactant/pull/646#issuecomment-4870762467 + +import dataclasses +import datetime +import gzip +import io +import tarfile +from collections.abc import Generator + +import requests +from bs4 import BeautifulSoup, Tag +from loguru import logger + +import surfactant.configmanager +import surfactant.plugin +from surfactant.sbomtypes import SBOM, Relationship, Software + + +@dataclasses.dataclass +class DownloadedInfo: + distro: str + version: str + date: datetime.date + + def __hash__(self) -> int: + return hash((self.distro, self.version, self.date)) + + +class RootfsManager: + __config = surfactant.configmanager.ConfigManager() + __webpage_archive: dict[str, str] = {} + __downloaded_info: set[DownloadedInfo] = set() + # The version info extracted from a webpage + # Not a dict so the order on the webpage can be retained, which puts + # the newest version at the end + __version_archive: list[tuple[str, list[tuple[str, datetime.date]]]] = [] + # Mapping from ELF arch names to download arch names + __ELF_ARCH_TO_DOWNLOAD_ARCH: dict[str, str] = { + "EM_X86_64": "amd64", + "EM_AARCH64": "arm64", + "EM_ARM": "armhf", + "EM_PPC64": "ppc64el", + "EM_RISCV": "riscv64", + "EM_S390": "s390x", + } + __UBUNTU_BASE_DOWNLOAD_URL: str = "https://partner-images.canonical.com/oci/" + + def __init__(self): + # Create the data directory if needed + self.data_dir = self.__config.get_data_dir_path() / "rootfs_downloads" + if not self.data_dir.exists(): + self.data_dir.mkdir(parents=True) + else: + for child in self.data_dir.iterdir(): + if child.is_dir(): + if info := self.try_decode_filename(child.name): + self.__downloaded_info.add(info) + else: + logger.warning(f"Could not parse filename of downloaded rootfs: {child}") + + def try_decode_filename(self, name: str) -> DownloadedInfo | None: + split = name.split("@") + if len(split) != 3: + return None + datesplit = split[2].split("-") + if len(datesplit) != 3: + return None + return DownloadedInfo(split[0], split[1], datetime.date(*(int(x) for x in datesplit))) + + def get_web_subpage(self, subpage: str) -> str | None: + if subpage in self.__webpage_archive: + return self.__webpage_archive[subpage] + + r = requests.get(f"{self.__UBUNTU_BASE_DOWNLOAD_URL}{subpage}") + if r.status_code != 200: + return None + + self.__webpage_archive[subpage] = r.text + return r.text + + # Each table has the same format, so this can be used to iterate + # over meaningful entries in the table + def webpage_table_info(self, subpage: str) -> Generator[list[Tag], None, None]: + if root_page := self.get_web_subpage(subpage): + html = BeautifulSoup(root_page, features="html.parser") + raw_info: list[list[Tag]] = [] + # Extract each row of the table into a list + if body := html.find("body"): + if table := body.find("table"): + for tr in (tr for tr in table if isinstance(tr, Tag)): + raw_info.append([td for td in tr if isinstance(td, Tag)]) + # Information starts at the fourth row + yield from raw_info[3:] + + # Iterates over the directories on a webpage + def dirs_on_webpage(self, subpage: str) -> Generator[tuple[str, datetime.date], None, None]: + for row in self.webpage_table_info(subpage): + # Directories have 5 elements and the first column is an image with a [DIR] alt img + if len(row) == 5: + if img := row[0].find("img"): + if "alt" in img.attrs and img.attrs["alt"] == "[DIR]": + if a := row[1].find("a"): + raw_date = row[2].text.split(" ")[0] + yield (a.text, datetime.date(*(int(x) for x in raw_date.split("-")))) + + # Iterates over the .tar.gz files on a webpage + # This is needed since, for some reason, the number of "ubuntu" strings is inconsistent + # Just returns the name of the files + def tar_gz_on_webpage(self, subpage: str) -> Generator[str, None, None]: + # This is very similar to the dirs_on_webpage; they could maybe be combined? + for row in self.webpage_table_info(subpage): + if len(row) == 5: + if img := row[0].find("img"): + if "src" in img.attrs and img.attrs["src"] == "/icons/compressed.gif": + if a := row[1].find("a"): + yield a.text + + # Potential future work: + # - More considerations than just architecture + # - More distros than just Ubuntu + # - Download older versions, etc. + def download_if_needed(self, architecture: str): + # Download the version information if needed + if len(self.__version_archive) == 0: + # This takes a long time so print some status messages on what's happening + logger.info("Downloading Ubuntu version info") + for ver_name, _ in self.dirs_on_webpage(""): + logger.info(f"Parsing Ubuntu version {ver_name}") + self.__version_archive.append( + (ver_name, [x for x in self.dirs_on_webpage(ver_name) if x[0] != "current/"]) + ) + + # The latest version is always at the end so use that + version_name, version_downloads = self.__version_archive[-1] + # Also use the latest version download + download_dir, modified_date = max(version_downloads, key=lambda x: x[1]) + # Download it if needed + arch = self.__ELF_ARCH_TO_DOWNLOAD_ARCH[architecture] + dir_name = f"ubuntu-{version_name[:-1]}@{arch}@{modified_date.strftime('%Y-%m-%d')}" + if dir_ver := self.try_decode_filename(dir_name): + if dir_ver not in self.__downloaded_info: + for f in self.tar_gz_on_webpage(f"{version_name}{download_dir}"): + if f"{arch}-root" in f: + r = requests.get( + f"{self.__UBUNTU_BASE_DOWNLOAD_URL}{version_name}{download_dir}{f}", + stream=True, + ) + + if r.status_code != 200: + continue + + logger.info(f"Downloading {f}") + + try: + (self.data_dir / dir_name).mkdir(exist_ok=True) + with ( + gzip.GzipFile(fileobj=io.BytesIO(r.raw.read())) as gfile, + tarfile.TarFile(fileobj=gfile) as tfile, + ): + tfile.extractall(self.data_dir / dir_name) + self.__downloaded_info.add(dir_ver) + break + except (gzip.BadGzipFile, tarfile.TarError) as e: + logger.warning(f"Error extracting {f} - {e}") + + +rootfs_manager = RootfsManager() + + +@surfactant.plugin.hookimpl +def establish_relationships( + sbom: SBOM, software: Software, metadata: object +) -> list[Relationship] | None: + if isinstance(metadata, dict) and "elfArchitecture" in metadata: + rootfs_manager.download_if_needed(metadata["elfArchitecture"])