[SED] Wipe SSD CLI

sonic_platform_base/sed_mgmt_base.py

@@ -32,11 +32,12 @@ def _read_sed_config_value(key):
 class SedMgmtBase:
     """
     Base class for SED password management.
-    Implements change_sed_password and reset_sed_password using abstract getters.
+    Implements change_sed_password, reset_sed_password, and wipe_ssd using abstract getters.
     """
 
     SED_PW_CHANGE_SCRIPT = '/usr/local/bin/sed_pw_change.sh'
     SED_PW_RESET_SCRIPT = '/usr/local/bin/sed_pw_reset.sh'
+    SSD_ERASE_SCRIPT = '/usr/local/bin/ssd_erase.sh'
 
     def get_min_sed_password_len(self):
         """
@@ -65,6 +66,15 @@ def get_default_sed_password(self):
         """
         raise NotImplementedError
 
+    def get_psid(self):
+        """
+        Return the SED PSID (Physical Security ID) for this platform.
+
+        Returns:
+            str: PSID string, or None if it cannot be retrieved.
+        """
+        raise NotImplementedError
+
     def get_tpm_bank_a_address(self):
         """
         Return TPM bank A persistent handle for SED password (e.g. 0x81010001).
@@ -142,3 +152,38 @@ def reset_sed_password(self):
         except Exception as e:
             logger.log_error(f"Failed to reset SED password: {e}")
             return False
+
+    def wipe_ssd(self):
+        """
+        Graceful SSD wipe: crypto erase (PSID revert) + NVMe block erase.
+
+        Returns:
+            bool: True on success, False otherwise.
+        """
+        try:
+            psid = self.get_psid()
+            if not psid:
+                logger.log_error("Failed to get PSID from platform.")
+                return False
+            default_pw = self.get_default_sed_password()
+            if not default_pw:
+                logger.log_error("Failed to get default SED password.")
+                return False
+            bank_a = self.get_tpm_bank_a_address()
+            bank_b = self.get_tpm_bank_b_address()
+            if not bank_a or not bank_b:
+                logger.log_error(f"TPM bank address is not valid: bank_a: {bank_a}, bank_b: {bank_b}. Check {SED_CONFIG_PATH}.")
+                return False
+            subprocess.check_call(
+                [self.SSD_ERASE_SCRIPT,
+                 '-a', bank_a, '-b', bank_b,
+                 '-p', default_pw, '-s', psid],
+                universal_newlines=True,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.PIPE,
+                start_new_session=True,
+            )
+            return True
+        except Exception as e:
+            logger.log_error(f"Failed to wipe SSD: {e}")
+            return False

tests/sed_mgmt_base_test.py

@@ -36,6 +36,7 @@ class IncompleteSedMgmt(SedMgmtBase):
             [sed.get_min_sed_password_len],
             [sed.get_max_sed_password_len],
             [sed.get_default_sed_password],
+            [sed.get_psid],
         ]
         for method_list in not_implemented_methods:
             with pytest.raises(NotImplementedError):
@@ -71,6 +72,9 @@ def get_max_sed_password_len(self):
     def get_default_sed_password(self):
         return 'default_secret'
 
+    def get_psid(self):
+        return 'PSID1234567890ABCDEF1234567890ABCDEF'
+
     def get_tpm_bank_a_address(self):
         return '0x81010001'
 
@@ -159,6 +163,57 @@ def test_reset_sed_password_script_fails(self, mock_check_call):
         assert sed.reset_sed_password() is False
 
 
+class TestSedMgmtBaseWipeSsd:
+    """Test SedMgmtBase.wipe_ssd."""
+
+    @mock.patch('subprocess.check_call')
+    def test_wipe_ssd_success(self, mock_check_call):
+        """wipe_ssd calls ssd_erase.sh with correct args and returns True."""
+        sed = ConcreteSedMgmt()
+        assert sed.wipe_ssd() is True
+        mock_check_call.assert_called_once()
+        call_args = mock_check_call.call_args[0][0]
+        assert call_args[0] == SedMgmtBase.SSD_ERASE_SCRIPT
+        assert call_args[call_args.index('-a') + 1] == '0x81010001'
+        assert call_args[call_args.index('-b') + 1] == '0x81010002'
+        assert call_args[call_args.index('-p') + 1] == 'default_secret'
+        assert call_args[call_args.index('-s') + 1] == 'PSID1234567890ABCDEF1234567890ABCDEF'
+        # start_new_session=True is critical: wipe must survive SSH drop.
+        assert mock_check_call.call_args[1].get('start_new_session') is True
+
+    def test_wipe_ssd_no_psid(self):
+        """wipe_ssd returns False when PSID is not available."""
+        sed = ConcreteSedMgmt()
+        with mock.patch.object(sed, 'get_psid', return_value=None):
+            assert sed.wipe_ssd() is False
+
+    def test_wipe_ssd_no_default_password(self):
+        """wipe_ssd returns False when default password is not available."""
+        sed = ConcreteSedMgmt()
+        with mock.patch.object(sed, 'get_default_sed_password', return_value=None):
+            assert sed.wipe_ssd() is False
+
+    def test_wipe_ssd_missing_bank_a(self):
+        """wipe_ssd returns False when bank_a is missing."""
+        sed = ConcreteSedMgmt()
+        with mock.patch.object(sed, 'get_tpm_bank_a_address', return_value=None):
+            assert sed.wipe_ssd() is False
+
+    def test_wipe_ssd_missing_bank_b(self):
+        """wipe_ssd returns False when bank_b is missing."""
+        sed = ConcreteSedMgmt()
+        with mock.patch.object(sed, 'get_tpm_bank_b_address', return_value=''):
+            assert sed.wipe_ssd() is False
+
+    @mock.patch('subprocess.check_call')
+    def test_wipe_ssd_script_fails(self, mock_check_call):
+        """wipe_ssd returns False when ssd_erase.sh exits non-zero."""
+        import subprocess
+        mock_check_call.side_effect = subprocess.CalledProcessError(11, 'ssd_erase.sh')
+        sed = ConcreteSedMgmt()
+        assert sed.wipe_ssd() is False
+
+
 class TestReadSedConfigValue:
     """Test _read_sed_config_value helper."""