Merge pull request #29 from SamD2021/add_fixboard

Add fixboard to firmware reflash

dpu-tools

@@ -204,7 +204,10 @@ class BFTools(DPUTools):
         firmware_up_parser.set_defaults(subcommand="firmware_up")
 
         firmware_up_parser.add_argument(
-            "--version", action="store_true", help="BF Version to Upgrade"
+            "-v",
+            "--version",
+            type=str,
+            help="BF Version to Upgrade",
         )
 
         firmware_subparsers.add_parser(

utils/common.py

@@ -181,3 +181,23 @@ def detect_dpu_type() -> Result:
             returncode=-1,
         )
     return Result(next(iter(kinds)), "", 0)
+
+
+def list_http_directory(url: str) -> list[str]:
+    """
+    Fetch the directory listing from an HTTP server.
+    """
+    response = requests.get(url)
+    response.raise_for_status()
+    # Use a simple regex to extract links
+    return re.findall(r'href=["\'](.*?)["\']', response.text)
+
+
+def ssh_run(cmd: str, address: str, dry_run: bool) -> Result:
+    """
+    Takes a command and runs it on the remote location
+    """
+    return run(
+        f"ssh -o 'StrictHostKeyChecking=no' -o 'UserKnownHostsFile=/dev/null' {address} '{cmd}'",
+        dry_run=dry_run,
+    )

utils/fwutils.py

@@ -1,7 +1,11 @@
 #!/usr/bin/env python3
 import logging
+from os import makedirs
 import sys
 import pexpect
+import json
+import re
+import tempfile
 from typing import Optional
 from utils.minicom import minicom_cmd, pexpect_child_wait, configure_minicom
 from utils.common_ipu import (
@@ -12,7 +16,14 @@
     minicom_get_version,
 )
 from utils.common_bf import find_bf_pci_addresses_or_quit, mst_flint, bf_version
-from utils.common import extract_tar_gz, download_file, run, Result
+from utils.common import (
+    extract_tar_gz,
+    download_file,
+    run,
+    Result,
+    list_http_directory,
+    ssh_run,
+)
 from utils.remote_api import RemoteAPI
 
 
@@ -37,8 +48,9 @@ def __init__(
                 "clean_up_imc",
                 "flash_ssd_image",
                 "flash_spi_image",
+                "apply_fixboard",
             ]
-        self.steps_to_run = steps_to_run
+        self.steps_to_run = steps_to_run if not dry_run else []
         if self.dry_run:
             self.logger.info(
                 "DRY RUN, This is just a preview of the actions that will be taken"
@@ -127,6 +139,15 @@ def reflash_ipu(self) -> None:
         else:
             self.logger.info("Skipping flash_spi_image")
 
+        self.logger.info("Step 5: apply_fixboard")
+        if self.should_run("apply_fixboard"):
+            if self.fixboard_is_needed():
+                self.logger.info("Applying fixboard!")
+                self.apply_fixboard()
+            else:
+                self.logger.info("Fixboard not needed!")
+        else:
+            self.logger.info("Skipping applying_fixboard")
         # Step 5: Reboot IMC
         self.logger.info("Done!")
         self.logger.info(f"Please cold reboot IMC at {self.imc_address}")
@@ -234,6 +255,139 @@ def get_images(self) -> tuple[str, str]:
 
         return ssd_bin_file, recovery_bin_file
 
+    def ensure_fixboard_image_on_imc(self) -> str:
+        """
+        Download and extract the SSD image and recovery firmware for the given version.
+        Return the paths for both files.
+        """
+        # Regex to capture the number after the first `-` and a word
+        pattern = r"^[a-zA-Z0-9]+-[a-zA-Z]+(\d+)"
+
+        # Search for the number in the hostname
+        match = re.search(pattern, self.imc_address)
+
+        if match:
+            number = match.group(1)
+            self.logger.debug(f"Extracted number: {number}")
+        else:
+            self.logger.error(
+                "No number found in the hostname. Can't proceed with detecting pre-built images"
+            )
+            exit(1)
+
+        base_url = f"http://{self.repo_url}/fixboard"
+        server_dirs = list_http_directory(base_url)
+        self.logger.debug(f"server_dirs:{server_dirs}")
+        if any(number in dir for dir in server_dirs):
+            base_url = f"http://{self.repo_url}/fixboard/{number}"
+            with tempfile.TemporaryDirectory() as temp_dir:
+                download_dir = f"{temp_dir}/{self.repo_url}/fixboard/{number}"  # Or any preferred directory for temp storage
+                makedirs(download_dir, exist_ok=True)
+                fixboard_files = list_http_directory(base_url)
+                fixboard_local_file_paths: list[str] = []
+                for file in fixboard_files:
+                    fixboard_local_file_paths.append(
+                        download_file(f"{base_url}/{file}", download_dir)
+                    )
+
+                # Find the required .bin files
+                self.logger.debug(
+                    f"fixboard_local_file_paths: {fixboard_local_file_paths}"
+                )
+                for fixboard_file in fixboard_local_file_paths:
+                    if fixboard_file.endswith(".bin.board_config"):
+
+                        full_address = f"root@{self.imc_address}"
+
+                        file_name = fixboard_file.split("/")[-1]
+                        result = run(
+                            f"scp -o 'StrictHostKeyChecking=no' -o 'UserKnownHostsFile=/dev/null' {fixboard_file} {full_address}:/tmp/{file_name}",
+                            dry_run=self.dry_run,
+                        )
+                        if result.returncode:
+                            self.logger.error(
+                                f"Couldn't transfer file using scp, Error: {result.err}"
+                            )
+                            exit(1)
+                        return f"/tmp/{file_name}"
+
+                self.logger.error("Couldn't find the board_config file, exitting...")
+                exit(1)
+        else:
+            self.logger.error(
+                f"server {self.imc_address} with number {number} doesn't have pre built fixboard images yet, please add the necessary files to the repo"
+            )
+            exit(1)
+
+    def apply_fixboard(self) -> None:
+        fixboard_bin_board_config_file = self.ensure_fixboard_image_on_imc()
+        full_address = f"root@{self.imc_address}"
+        result = ssh_run(
+            "flash_erase /dev/mtd0 0x30000 1",
+            full_address,
+            dry_run=self.dry_run,
+        )
+        if result.returncode:
+            self.logger.error(f"Couldn't flash_erase, Error: {result.err}")
+            exit(1)
+
+        result = ssh_run(
+            f"nandwrite --start=0x30000 --input-size=0x1000 -p /dev/mtd0 {fixboard_bin_board_config_file}",
+            full_address,
+            dry_run=self.dry_run,
+        )
+        if result.returncode:
+            self.logger.error(f"Couldn't nandwrite, Error: {result.err}")
+            exit(1)
+        self.logger.info("Rebooting IMC now!")
+        result = ssh_run(
+            "reboot",
+            full_address,
+            dry_run=self.dry_run,
+        )
+
+    def fixboard_is_needed(self) -> bool:
+        full_address = f"root@{self.imc_address}"
+        try:
+            result = ssh_run(
+                "iset-cli get-board-config",
+                full_address,
+                dry_run=self.dry_run,
+            )
+            if result.returncode:
+                self.logger.error(
+                    f"Couldn't retrieve get board config using iset-cli, Error: {result.err}"
+                )
+                exit(1)
+            # Parse the JSON from the command output
+            self.logger.debug(f"Board Config command output: {result.out}")
+            board_config: dict[str, str] = json.loads(result.out)
+            self.logger.debug(f"Json Board Config: {board_config}")
+            for key, value in board_config.items():
+                if "MAC Address" in key:
+                    if value in ["00:00:00:00:00:00", "FF:FF:FF:FF:FF:FF"]:
+                        return True
+                if "PBA" in key:
+                    if value in [
+                        "000000000000000000000000",
+                        "FFFFFFFFFFFFFFFFFFFFFFFF",
+                    ]:
+                        self.logger.debug(f"PBA check failed for {key}: {value}")
+                        return True
+                if "Serial Number" in key:
+                    if value == "":
+                        return True
+
+            return False
+
+        except json.JSONDecodeError as e:
+            self.logger.error(f"Failed to parse JSON: {e}")
+            exit(1)
+
+        except Exception as e:
+            self.logger.error(f"Unexpected error: {e}")
+            exit(1)
+
 
 class BFFirmware:
     def __init__(self, id: int, version_to_flash: Optional[str] = None):