[fi] Add RNG command handler

This commit adds the command handler for the RNG FI tests. Currently
the csrng test is supported.

Signed-off-by: Pascal Nasahl <[email protected]>

fault_injection/configs/pen.global_fi.rng.cw310.yaml

@@ -0,0 +1,44 @@
+target:
+  target_type: cw310
+  fpga_bitstream: "../objs/lowrisc_systems_chip_earlgrey_cw310_0.1.bit"
+  force_program_bitstream: False
+  fw_bin: "../objs/sca_ujson_fpga_cw310.bin"
+  output_len_bytes: 16
+  target_clk_mult: 0.24
+  target_freq: 24000000
+  baudrate: 115200
+  protocol: "ujson"
+  port: "/dev/ttyACM4"
+fisetup:
+  fi_gear: "husky"
+  fi_type: "voltage_glitch"
+  parameter_generation: "random"
+  # Voltage glitch width in cycles.
+  glitch_width_min: 5
+  glitch_width_max: 150
+  glitch_width_step: 3
+  # Range for trigger delay in cycles.
+  trigger_delay_min: 0
+  trigger_delay_max: 500
+  trigger_step: 10
+  # Number of iterations for the parameter sweep.
+  num_iterations: 100
+fiproject:
+  # Project database type and memory threshold.
+  project_db: "ot_fi_project"
+  project_mem_threshold: 10000
+  # Store FI plot.
+  show_plot: True
+  num_plots: 10
+  plot_x_axis: "trigger_delay"
+  plot_x_axis_legend: "[cycles]"
+  plot_y_axis: "glitch_width"
+  plot_y_axis_legend: "[cycles]"
+test:
+  which_test: "rng_csrng_bias"
+  expected_result: '{"res":0,"rand":[932170270,3480632584,387346064,186012424,899661374,2795183089,336687633,3222931513,1490543709,3319795384,3464147855,1850271046,1239323641,2292604615,3314177342,1567494162],"alerts":[0,0,0],"err_status":0}'
+  # Set to true if the test should ignore alerts returned by the test. As the
+  # alert handler on the device could sometime fire alerts that are not
+  # related to the FI, ignoring is by default set to true. A manual analysis
+  # still can be performed as the alerts are stored in the database.
+  ignore_alerts: True

fault_injection/fi_rng.py

@@ -0,0 +1,252 @@
+#!/usr/bin/env python3
+# Copyright lowRISC contributors.
+# Licensed under the Apache License, Version 2.0, see LICENSE for details.
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+import logging
+from datetime import datetime
+from pathlib import Path
+
+import yaml
+from fi_gear.fi_gear import FIGear
+from project_library.project import FIProject, FISuccess, ProjectConfig
+from tqdm import tqdm
+
+import util.helpers as helpers
+from target.communication.fi_rng_commands import OTFIRng
+from target.targets import Target, TargetConfig
+from util import plot
+
+logger = logging.getLogger()
+
+
+def setup(cfg: dict, project: Path):
+    """ Setup target, FI gear, and project.
+
+    Args:
+        cfg: The configuration for the current experiment.
+        project: The path for the project file.
+
+    Returns:
+        The target, FI gear, and project.
+    """
+    # Calculate pll_frequency of the target.
+    # target_freq = pll_frequency * target_clk_mult
+    # target_clk_mult is a hardcoded constant in the FPGA bitstream.
+    cfg["target"]["pll_frequency"] = cfg["target"]["target_freq"] / cfg["target"]["target_clk_mult"]
+
+    # Create target config & setup target.
+    logger.info(f"Initializing target {cfg['target']['target_type']} ...")
+    target_cfg = TargetConfig(
+        target_type = cfg["target"]["target_type"],
+        fw_bin = cfg["target"]["fw_bin"],
+        protocol = cfg["target"]["protocol"],
+        pll_frequency = cfg["target"]["pll_frequency"],
+        bitstream = cfg["target"].get("fpga_bitstream"),
+        force_program_bitstream = cfg["target"].get("force_program_bitstream"),
+        baudrate = cfg["target"].get("baudrate"),
+        port = cfg["target"].get("port"),
+        output_len = cfg["target"].get("output_len_bytes"),
+        usb_serial = cfg["target"].get("usb_serial")
+    )
+    target = Target(target_cfg)
+
+    # Init FI gear.
+    fi_gear = FIGear(cfg)
+
+    # Init project.
+    project_cfg = ProjectConfig(type = cfg["fiproject"]["project_db"],
+                                path = project,
+                                overwrite = True,
+                                fi_threshold = cfg["fiproject"].get("project_mem_threshold")
+                                )
+    project = FIProject(project_cfg)
+    project.create_project()
+
+    return target, fi_gear, project
+
+
+def print_fi_statistic(fi_results: list) -> None:
+    """ Print FI Statistic.
+
+    Prints the number of FISuccess.SUCCESS, FISuccess.EXPRESPONSE, and
+    FISuccess.NORESPONSE.
+
+    Args:
+        fi_results: The FI results.
+    """
+    num_total = len(fi_results)
+    num_succ = round((fi_results.count(FISuccess.SUCCESS) / num_total) * 100, 2)
+    num_exp = round((fi_results.count(FISuccess.EXPRESPONSE) / num_total) * 100, 2)
+    num_no = round((fi_results.count(FISuccess.NORESPONSE) / num_total) * 100, 2)
+    logger.info(f"{num_total} faults, {fi_results.count(FISuccess.SUCCESS)}"
+                f"({num_succ}%) successful, {fi_results.count(FISuccess.EXPRESPONSE)}"
+                f"({num_exp}%) expected, and {fi_results.count(FISuccess.NORESPONSE)}"
+                f"({num_no}%) no response.")
+
+
+def fi_parameter_sweep(cfg: dict, target: Target, fi_gear,
+                       project: FIProject, ot_communication: OTFIRng) -> None:
+    """ Fault parameter sweep.
+
+    Sweep through the fault parameter space.
+
+    Args:
+        cfg: The FI project configuration.
+        target: The OpenTitan target.
+        fi_gear: The FI gear to use.
+        project: The project to store the results.
+        ot_communication: The OpenTitan RNG FI communication interface.
+    Returns:
+        device_id: The ID of the target device.
+    """
+    # Configure the RNG FI code on the target.
+    device_id = ot_communication.init(cfg["test"]["which_test"])
+    # Store results in array for a quick access.
+    fi_results = []
+    # Start the parameter sweep.
+    remaining_iterations = fi_gear.get_num_fault_injections()
+    with tqdm(total=remaining_iterations, desc="Injecting", ncols=80,
+              unit=" different faults") as pbar:
+        while remaining_iterations > 0:
+            # Get fault parameters (e.g., trigger delay, glitch voltage).
+            fault_parameters = fi_gear.generate_fi_parameters()
+
+            # Arm the FI gear.
+            fi_gear.arm_trigger(fault_parameters)
+
+            # Start test on OpenTitan.
+            ot_communication.start_test(cfg)
+
+            # Read response.
+            response = ot_communication.read_response(max_tries=30)
+            response_compare = response
+            expected_response = cfg["test"]["expected_result"]
+
+            # Compare response. If no response is received, the device mostly
+            # crashed or was resetted.
+            if response_compare == "":
+                # No UART response received.
+                fi_result = FISuccess.NORESPONSE
+                # Resetting OT as it most likely crashed.
+                ot_communication = target.reset_target(com_reset = True)
+                # Re-establish UART connection.
+                ot_communication = OTFIRng(target)
+                # Configure the RNG FI code on the target.
+                ot_communication.init(cfg["test"]["which_test"])
+                # Reset FIGear if necessary.
+                fi_gear.reset()
+            else:
+                # If the test decides to ignore alerts triggered by the alert
+                # handler, remove it from the received and expected response.
+                # In the database, the received alert is still available for
+                # further diagnosis.
+                if cfg["test"]["ignore_alerts"]:
+                    resp_json = json.loads(response_compare)
+                    exp_json = json.loads(expected_response)
+                    if "alerts" in resp_json:
+                        del resp_json["alerts"]
+                        response_compare = json.dumps(resp_json,
+                                                      separators=(',', ':'))
+                    if "alerts" in exp_json:
+                        del exp_json["alerts"]
+                        expected_response = json.dumps(exp_json,
+                                                       separators=(',', ':'))
+
+                # Check if result is expected result (FI failed) or unexpected
+                # result (FI successful).
+                fi_result = FISuccess.SUCCESS
+                if response_compare == expected_response:
+                    # Expected result received. No FI effect.
+                    fi_result = FISuccess.EXPRESPONSE
+
+            # Store result into FIProject.
+            project.append_firesult(
+                response = response,
+                fi_result = fi_result,
+                trigger_delay = fault_parameters.get("trigger_delay"),
+                glitch_voltage = fault_parameters.get("glitch_voltage"),
+                glitch_width = fault_parameters.get("glitch_width"),
+                x_pos = fault_parameters.get("x_pos"),
+                y_pos = fault_parameters.get("y_pos")
+            )
+            fi_results.append(fi_result)
+
+            remaining_iterations -= 1
+            pbar.update(1)
+    print_fi_statistic(fi_results)
+    return device_id
+
+
+def print_plot(project: FIProject, config: dict, file: Path) -> None:
+    """ Print plot of traces.
+
+    Printing the plot helps to narrow down the fault injection parameters.
+
+    Args:
+        project: The project containing the traces.
+        config: The capture configuration.
+        file: The file path.
+    """
+    if config["fiproject"]["show_plot"]:
+        plot.save_fi_plot_to_file(config, project, file)
+        logger.info("Created plot.")
+        logger.info(f'Created plot: '
+                    f'{Path(str(file) + ".html").resolve()}')
+
+
+def main(argv=None):
+    # Configure the logger.
+    logger.setLevel(logging.INFO)
+    console = logging.StreamHandler()
+    logger.addHandler(console)
+
+    # Parse the provided arguments.
+    args = helpers.parse_arguments(argv)
+
+    # Load configuration from file.
+    with open(args.cfg) as f:
+        cfg = yaml.load(f, Loader=yaml.FullLoader)
+
+    # Setup the target, FI gear, and project.
+    target, fi_gear, project = setup(cfg, args.project)
+
+    # Establish communication interface with OpenTitan.
+    ot_communication = OTFIRng(target)
+
+    # FI parameter sweep.
+    device_id = fi_parameter_sweep(cfg, target, fi_gear, project, ot_communication)
+
+    # Print plot.
+    print_plot(project.get_firesults(start=0, end=cfg["fiproject"]["num_plots"]),
+               cfg, args.project)
+
+    # Save metadata.
+    metadata = {}
+    metadata["device_id"] = device_id
+    metadata["datetime"] = datetime.now().strftime("%m/%d/%Y, %H:%M:%S")
+    # Store bitstream information.
+    metadata["fpga_bitstream_path"] = cfg["target"].get("fpga_bitstream")
+    if cfg["target"].get("fpga_bitstream") is not None:
+        metadata["fpga_bitstream_crc"] = helpers.file_crc(cfg["target"]["fpga_bitstream"])
+    if args.save_bitstream:
+        metadata["fpga_bitstream"] = helpers.get_binary_blob(cfg["target"]["fpga_bitstream"])
+    # Store binary information.
+    metadata["fw_bin_path"] = cfg["target"]["fw_bin"]
+    metadata["fw_bin_crc"] = helpers.file_crc(cfg["target"]["fw_bin"])
+    if args.save_binary:
+        metadata["fw_bin"] = helpers.get_binary_blob(cfg["target"]["fw_bin"])
+    # Store user provided notes.
+    metadata["notes"] = args.notes
+    # Store the Git hash.
+    metadata["git_hash"] = helpers.get_git_hash()
+    # Write metadata into project database.
+    project.write_metadata(metadata)
+
+    # Save and close project.
+    project.save()
+
+
+if __name__ == "__main__":
+    main()

target/communication/fi_rng_commands.py

@@ -0,0 +1,75 @@
+# Copyright lowRISC contributors.
+# Licensed under the Apache License, Version 2.0, see LICENSE for details.
+# SPDX-License-Identifier: Apache-2.0
+"""Communication interface for OpenTitan RNG FI framework.
+
+Communication with OpenTitan happens over the uJSON command interface.
+"""
+import json
+import time
+from typing import Optional
+
+
+class OTFIRng:
+    def __init__(self, target) -> None:
+        self.target = target
+
+    def _ujson_rng_cmd(self) -> None:
+        time.sleep(0.01)
+        self.target.write(json.dumps("RngFi").encode("ascii"))
+        time.sleep(0.01)
+
+    def init(self, test: str) -> None:
+        """ Initialize the RNG FI code on the chip.
+
+        Args:
+            test: The selected test.
+        Returns:
+            The device ID of the device.
+        """
+        # RngFi command.
+        self._ujson_rng_cmd()
+        # Init command.
+        time.sleep(0.01)
+        if "csrng" in test:
+            self.target.write(json.dumps("CsrngInit").encode("ascii"))
+        # Read back device ID from device.
+        return self.read_response(max_tries=30)
+
+    def rng_csrng_bias(self) -> None:
+        """ Starts the csrng test.
+        """
+        # RngFi command.
+        time.sleep(0.05)
+        self._ujson_rng_cmd()
+        # Csrng command.
+        time.sleep(0.05)
+        self.target.write(json.dumps("Csrng").encode("ascii"))
+
+    def start_test(self, cfg: dict) -> None:
+        """ Start the selected test.
+
+        Call the function selected in the config file. Uses the getattr()
+        construct to call the function.
+
+        Args:
+            cfg: Config dict containing the selected test.
+        """
+        test_function = getattr(self, cfg["test"]["which_test"])
+        test_function()
+
+    def read_response(self, max_tries: Optional[int] = 1) -> str:
+        """ Read response from RNG FI framework.
+        Args:
+            max_tries: Maximum number of attempts to read from UART.
+
+        Returns:
+            The JSON response of OpenTitan.
+        """
+        it = 0
+        while it != max_tries:
+            read_line = str(self.target.readline())
+            if "RESP_OK" in read_line:
+                return read_line.split("RESP_OK:")[1].split(" CRC:")[0]
+            it += 1
+        return ""