diff --git a/sonic_platform_base/sed_mgmt_base.py b/sonic_platform_base/sed_mgmt_base.py index cd571a0a4..b56e37ff6 100644 --- a/sonic_platform_base/sed_mgmt_base.py +++ b/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 diff --git a/tests/sed_mgmt_base_test.py b/tests/sed_mgmt_base_test.py index cb358d8ec..1a9f41f32 100644 --- a/tests/sed_mgmt_base_test.py +++ b/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."""