Merge ~ltrager/maas:lp1835954_2.8 into maas:2.8

[Lee Trager]

This is the NVMe secure erase implementation (+ one more patch, to fix/improve quick erase).
I tried to explain/detail everything in the commit messages; also, code/tests were validated against Flake8 to prevent style issues.

Backport of 9633810

[Lee Trager (ltrager)]

Reviewed in https://code.launchpad.net/~gpiccoli/maas/+git/maas/+merge/387316

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b lp1835954_2.8 lp:~ltrager/maas/+git/maas into -b 2.8 lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: ddf53cddd6079dd0b1fb4a128cb134f8c4f88d40

review: Approve

diff --git a/src/maasserver/compose_preseed.py b/src/maasserver/compose_preseed.py
index 8acf2f5..4baa65a 100644
--- a/src/maasserver/compose_preseed.py
+++ b/src/maasserver/compose_preseed.py
@@ -391,6 +391,9 @@ def get_base_preseed(node=None):
             # jq is used during enlistment to read the JSON string containing
             # the system_id of the newly created machine.
             cloud_config["packages"] += ["jq"]
+        # On disk erasing, we need nvme-cli
+        if node is not None and node.status == NODE_STATUS.DISK_ERASING:
+            cloud_config["packages"] += ["nvme-cli"]
 
     return cloud_config
 
diff --git a/src/metadataserver/user_data/templates/snippets/maas_wipe.py b/src/metadataserver/user_data/templates/snippets/maas_wipe.py
old mode 100644
new mode 100755
index 317c62b..f7e611a
--- a/src/metadataserver/user_data/templates/snippets/maas_wipe.py
+++ b/src/metadataserver/user_data/templates/snippets/maas_wipe.py
@@ -29,12 +29,95 @@ def list_disks():
     return disks
 
 
-def get_disk_security_info(disk):
-    """Get the disk security information.
+def get_nvme_security_info(disk):
+    """Gather NVMe information from the NVMe disks using the
+    nvme-cli tool. Info from id-ctrl and id-ns is needed for
+    secure erase (nvme format) and write zeroes."""
+
+    # Grab the relevant info from nvme id-ctrl. We need to check the
+    # following bits:
+    #
+    # OACS (Optional Admin Command Support) bit 1: Format supported
+    # ONCS (Optional NVM Command Support) bit 3: Write Zeroes supported
+    # FNA (Format NVM Attributes) bit 2: Cryptographic format supported
+
+    security_info = {
+        "format_supported": False,
+        "writez_supported": False,
+        "crypto_format": False,
+        "nsze": 0,
+        "lbaf": 0,
+        "ms": 0,
+    }
+
+    try:
+        output = subprocess.check_output(["nvme", "id-ctrl", DEV_PATH % disk])
+    except subprocess.CalledProcessError as exc:
+        print_flush("Error on nvme id-ctrl (%s)" % exc.returncode)
+        return security_info
+    except OSError as exc:
+        print_flush("OS error when running nvme-cli (%s)" % exc.strerror)
+        return security_info
+
+    output = output.decode()
+
+    for line in output.split("\n"):
+        if "oacs" in line:
+            oacs = line.split(":")[1]
+            if int(oacs, 16) & 0x2:
+                security_info["format_supported"] = True
+
+        if "oncs" in line:
+            oncs = line.split(":")[1]
+            if int(oncs, 16) & 0x8:
+                security_info["writez_supported"] = True
+
+        if "fna" in line:
+            fna = line.split(":")[1]
+            if int(fna, 16) & 0x4:
+                security_info["crypto_format"] = True
+
+    # Next step: collect LBAF (LBA Format), MS (Metadata Setting) and
+    # NSZE (Namespace Size) from id-ns. According to NVMe spec, bits 0:3
+    # from FLBAS corresponds to the LBAF value, whereas bit 4 is MS.
+
+    try:
+        output = subprocess.check_output(["nvme", "id-ns", DEV_PATH % disk])
+    except subprocess.CalledProcessError as exc:
+        print_flush("Error on nvme id-ns (%s)" % exc.returncode)
+        security_info["format_supported"] = False
+        security_info["writez_supported"] = False
+        return security_info
+    except OSError as exc:
+        print_flush("OS error when running nvme-cli (%s)" % exc.strerror)
+        security_info["format_supported"] = False
+        security_info["writez_supported"] = False
+        return security_info
+
+    output = output.decode()
+
+    for line in output.split("\n"):
+        if "nsze" in line:
+            # According to spec., this should be used as 0-based value.
+            nsze = line.split(":")[1]
+            security_info["nsze"] = int(nsze, 16) - 1
+
+        if "flbas" in line:
+            flbas = line.split(":")[1]
+            flbas = int(flbas, 16)
+            security_info["lbaf"] = flbas & 0xF
+
+            if flbas & 0x10:
+                security_info["ms"] = 1
+
+    return security_info
 
-    Uses `hdparam` to get security information about the disk. Sadly hdparam
-    doesn't provide an output that makes it easy to parse.
+
+def get_hdparm_security_info(disk):
+    """Get SCSI/ATA disk security info from hdparm.
+    Sadly hdparam doesn't provide an output that makes it easy to parse.
     """
+
     # Grab the security section for hdparam.
     security_section = []
     output = subprocess.check_output([b"hdparm", b"-I", DEV_PATH % disk])
@@ -62,57 +145,25 @@ def get_disk_security_info(disk):
     return security_info
 
 
+def get_disk_security_info(disk):
+    """Get the disk security information.
+
+    Uses `hdparam` to get security information about the SCSI/ATA disks.
+    If NVMe, nvme-cli is used instead.
+    """
+
+    if b"nvme" in disk:
+        return get_nvme_security_info(disk)
+
+    return get_hdparm_security_info(disk)
+
+
 def get_disk_info():
     """Return dictionary of wipeable disks and thier security information."""
     return {kname: get_disk_security_info(kname) for kname in list_disks()}
 
 
-def try_secure_erase(kname, info):
-    """Try to wipe the disk with secure erase."""
-    if info[b"supported"]:
-        if info[b"frozen"]:
-            print_flush(
-                "%s: not using secure erase; "
-                "drive is currently frozen." % kname.decode("ascii")
-            )
-            return False
-        elif info[b"locked"]:
-            print_flush(
-                "%s: not using secure erase; "
-                "drive is currently locked." % kname.decode("ascii")
-            )
-            return False
-        elif info[b"enabled"]:
-            print_flush(
-                "%s: not using secure erase; "
-                "drive security is already enabled." % kname.decode("ascii")
-            )
-            return False
-        else:
-            # Wiping using secure erase.
-            try:
-                secure_erase(kname)
-            except Exception as e:
-                print_flush(
-                    "%s: failed to be securely erased: %s"
-                    % (kname.decode("ascii"), e)
-                )
-                return False
-            else:
-                print_flush(
-                    "%s: successfully securely erased."
-                    % (kname.decode("ascii"))
-                )
-                return True
-    else:
-        print_flush(
-            "%s: drive does not support secure erase."
-            % (kname.decode("ascii"))
-        )
-        return False
-
-
-def secure_erase(kname):
+def secure_erase_hdparm(kname):
     """Securely wipe the device."""
     # First write 1 MiB of known data to the beginning of the block device.
     # This is used to check at the end of the secure erase that it worked
@@ -142,7 +193,7 @@ def secure_erase(kname):
 
     # Now that the user password is set the device should have its
     # security mode enabled.
-    info = get_disk_security_info(kname)
+    info = get_hdparm_security_info(kname)
     if not info[b"enabled"]:
         # If not enabled that means the password did not take, so it does not
         # need to be cleared.
@@ -167,7 +218,7 @@ def secure_erase(kname):
         failed_exc = exc
 
     # Make sure that the device is now not enabled.
-    info = get_disk_security_info(kname)
+    info = get_hdparm_security_info(kname)
     if info[b"enabled"]:
         # Wipe failed since security is still enabled.
         subprocess.check_output(
@@ -184,23 +235,197 @@ def secure_erase(kname):
         )
 
 
-def wipe_quickly(kname):
-    """Quickly wipe the disk by zeroing the beginning and end of the disk.
+def try_secure_erase_hdparm(kname, info):
+    """Try to wipe the disk with secure erase."""
+    if info[b"supported"]:
+        if info[b"frozen"]:
+            print_flush(
+                "%s: not using secure erase; "
+                "drive is currently frozen." % kname.decode("ascii")
+            )
+            return False
+        elif info[b"locked"]:
+            print_flush(
+                "%s: not using secure erase; "
+                "drive is currently locked." % kname.decode("ascii")
+            )
+            return False
+        elif info[b"enabled"]:
+            print_flush(
+                "%s: not using secure erase; "
+                "drive security is already enabled." % kname.decode("ascii")
+            )
+            return False
+        else:
+            # Wiping using secure erase.
+            try:
+                secure_erase_hdparm(kname)
+            except Exception as e:
+                print_flush(
+                    "%s: failed to be securely erased: %s"
+                    % (kname.decode("ascii"), e)
+                )
+                return False
+            else:
+                print_flush(
+                    "%s: successfully securely erased."
+                    % (kname.decode("ascii"))
+                )
+                return True
+    else:
+        print_flush(
+            "%s: drive does not support secure erase."
+            % (kname.decode("ascii"))
+        )
+        return False
+
+
+def try_secure_erase_nvme(kname, info):
+    """Perform a secure-erase on NVMe disk if that feature is
+    available. Prefer cryptographic erase, when available."""
+
+    if not info["format_supported"]:
+        print_flush(
+            "Device %s does not support formatting" % kname.decode("ascii")
+        )
+        return False
+
+    if info["crypto_format"]:
+        ses = 2
+    else:
+        ses = 1
+
+    try:
+        subprocess.check_output(
+            [
+                "nvme",
+                "format",
+                "-s",
+                str(ses),
+                "-l",
+                str(info["lbaf"]),
+                "-m",
+                str(info["ms"]),
+                DEV_PATH % kname,
+            ]
+        )
+    except subprocess.CalledProcessError as exc:
+        print_flush("Error with format command (%s)" % exc.returncode)
+        return False
+    except OSError as exc:
+        print_flush("OS error when running nvme-cli (%s)" % exc.strerror)
+        return False
+
+    print_flush(
+        "Secure erase was successful on NVMe drive %s" % kname.decode("ascii")
+    )
+    return True
+
+
+def try_secure_erase(kname, info):
+    """Entry-point for secure-erase for SCSI/ATA or NVMe disks."""
+
+    if b"nvme" in kname:
+        return try_secure_erase_nvme(kname, info)
+
+    return try_secure_erase_hdparm(kname, info)
 
-    This is not a secure erase but does make it harder to get the data from
-    the device.
+
+def wipe_quickly(kname):
+    """Quickly wipe the disk by using wipefs and zeroing the beginning
+    and end of the disk. This is not a secure erase but does make it
+    harder to get the data from the device and also clears previous layouts.
     """
+
+    wipe_error = 0
     print_flush("%s: starting quick wipe." % kname.decode("ascii"))
+    try:
+        subprocess.check_output(["wipefs", "-f", "-a", DEV_PATH % kname])
+        wipe_error -= 1
+    except subprocess.CalledProcessError as exc:
+        print_flush(
+            "%s: wipefs failed (%s)" % (kname.decode("ascii"), exc.returncode)
+        )
+        wipe_error += 1
+
     buf = b"\0" * 1024 * 1024 * 2  # 2 MiB
-    with open(DEV_PATH % kname, "wb") as fp:
+    try:
+        fp = open(DEV_PATH % kname, "wb")
         fp.write(buf)
         fp.seek(-len(buf), 2)
         fp.write(buf)
-    print_flush("%s: successfully quickly wiped." % kname.decode("ascii"))
+        wipe_error -= 1
+    except OSError as exc:
+        print_flush(
+            "%s: OS error while wiping beginning/end of disk (%s)"
+            % (kname.decode("ascii"), exc.strerror)
+        )
+        wipe_error += 1
+
+    if wipe_error > 0:
+        print_flush("%s: failed to be quickly wiped." % kname.decode("ascii"))
+    else:
+        print_flush("%s: successfully quickly wiped." % kname.decode("ascii"))
+
+
+def nvme_write_zeroes(kname, info):
+    """Perform a write-zeroes operation on NVMe device instead of
+    dd'ing 0 to the entire disk if secure erase is not available.
+    Write-zeroes is a faster way to clean a NVMe disk."""
+
+    fallback = False
+
+    if not info["writez_supported"]:
+        print(
+            "NVMe drive %s does not support write-zeroes"
+            % kname.decode("ascii")
+        )
+        fallback = True
+
+    if info["nsze"] <= 0:
+        print(
+            "Bad namespace information collected on NVMe drive %s"
+            % kname.decode("ascii")
+        )
+        fallback = True
+
+    if fallback:
+        print_flush("Will fallback to regular drive zeroing.")
+        return False
+
+    try:
+        subprocess.check_output(
+            [
+                "nvme",
+                "write-zeroes",
+                "-f",
+                "-s",
+                "0",
+                "-c",
+                str(hex(info["nsze"])[2:]),
+                DEV_PATH % kname,
+            ]
+        )
+    except subprocess.CalledProcessError as exc:
+        print_flush("Error with write-zeroes command (%s)" % exc.returncode)
+        return False
+    except OSError as exc:
+        print_flush("OS error when running nvme-cli (%s)" % exc.strerror)
+        return False
+
+    print_flush(
+        "%s: successfully zeroed (using write-zeroes)." % kname.decode("ascii")
+    )
+    return True
+
+
+def zero_disk(kname, info):
+    """Zero the entire disk, trying write-zeroes first if NVMe disk."""
 
+    if b"nvme" in kname:
+        if nvme_write_zeroes(kname, info):
+            return
 
-def zero_disk(kname):
-    """Zero the entire disk."""
     # Get the total size of the device.
     size = 0
     with open(DEV_PATH % kname, "rb") as fp:
@@ -266,8 +491,9 @@ def main():
         action="store_true",
         default=False,
         help=(
-            "Wipe 1MiB at the start and at the end of the drive to make data "
-            "recovery inconvenient and unlikely to happen by accident. This "
+            "Wipe 2MiB at the start and at the end of the drive to make data "
+            "recovery inconvenient and unlikely to happen by accident. Also, "
+            "it runs wipefs to clear known partition/layout signatures. This "
             "is not secure."
         ),
     )
@@ -288,7 +514,7 @@ def main():
             if args.quick_erase:
                 wipe_quickly(kname)
             else:
-                zero_disk(kname)
+                zero_disk(kname, info)
 
     print_flush("All disks have been successfully wiped.")
 
diff --git a/src/metadataserver/user_data/templates/snippets/tests/test_maas_wipe.py b/src/metadataserver/user_data/templates/snippets/tests/test_maas_wipe.py
index 995c0b5..191cd02 100644
--- a/src/metadataserver/user_data/templates/snippets/tests/test_maas_wipe.py
+++ b/src/metadataserver/user_data/templates/snippets/tests/test_maas_wipe.py
@@ -6,6 +6,7 @@
 __all__ = []
 
 import argparse
+import builtins
 import subprocess
 from textwrap import dedent
 from unittest.mock import call, MagicMock
@@ -22,148 +23,29 @@ from snippets.maas_wipe import (
     get_disk_info,
     get_disk_security_info,
     list_disks,
-    secure_erase,
+    nvme_write_zeroes,
+    secure_erase_hdparm,
     try_secure_erase,
     wipe_quickly,
     WipeError,
     zero_disk,
 )
-
-HDPARM_BEFORE_SECURITY = b"""\
-/dev/sda:
-
-ATA device, with non-removable media
-    Model Number:       INTEL SSDSC2CT240A4
-    Serial Number:      CVKI3206029X240DGN
-    Firmware Revision:  335u
-    Transport:          Serial, ATA8-AST, SATA 1.0a, SATA II Extensions
-Standards:
-    Used: unknown (minor revision code 0xffff)
-    Supported: 9 8 7 6 5
-    Likely used: 9
-Configuration:
-    Logical		max	current
-    cylinders	16383	16383
-    heads		16	16
-    sectors/track	63	63
-    --
-    CHS current addressable sectors:   16514064
-    LBA    user addressable sectors:  268435455
-    LBA48  user addressable sectors:  468862128
-    Logical  Sector size:                   512 bytes
-    Physical Sector size:                   512 bytes
-    Logical Sector-0 offset:                  0 bytes
-    device size with M = 1024*1024:      228936 MBytes
-    device size with M = 1000*1000:      240057 MBytes (240 GB)
-    cache/buffer size  = unknown
-    Nominal Media Rotation Rate: Solid State Device
-Capabilities:
-    LBA, IORDY(can be disabled)
-    Queue depth: 32
-    Standby timer values: spec'd by Standard, no device specific minimum
-    R/W multiple sector transfer: Max = 16	Current = 16
-    Advanced power management level: 254
-    DMA: mdma0 mdma1 mdma2 udma0 udma1 udma2 udma3 udma4 udma5 *udma6
-         Cycle time: min=120ns recommended=120ns
-    PIO: pio0 pio1 pio2 pio3 pio4
-         Cycle time: no flow control=120ns  IORDY flow control=120ns
-Commands/features:
-    Enabled	Supported:
-       *	SMART feature set
-            Security Mode feature set
-       *	Power Management feature set
-       *	Write cache
-       *	Look-ahead
-       *	Host Protected Area feature set
-       *	WRITE_BUFFER command
-       *	READ_BUFFER command
-       *	NOP cmd
-       *	DOWNLOAD_MICROCODE
-       *	Advanced Power Management feature set
-            Power-Up In Standby feature set
-       *	48-bit Address feature set
-       *	Mandatory FLUSH_CACHE
-       *	FLUSH_CACHE_EXT
-       *	SMART error logging
-       *	SMART self-test
-       *	General Purpose Logging feature set
-       *	WRITE_{DMA|MULTIPLE}_FUA_EXT
-       *	64-bit World wide name
-       *	IDLE_IMMEDIATE with UNLOAD
-       *	WRITE_UNCORRECTABLE_EXT command
-       *	{READ,WRITE}_DMA_EXT_GPL commands
-       *	Segmented DOWNLOAD_MICROCODE
-       *	Gen1 signaling speed (1.5Gb/s)
-       *	Gen2 signaling speed (3.0Gb/s)
-       *	Gen3 signaling speed (6.0Gb/s)
-       *	Native Command Queueing (NCQ)
-       *	Host-initiated interface power management
-       *	Phy event counters
-       *	DMA Setup Auto-Activate optimization
-            Device-initiated interface power management
-       *	Software settings preservation
-       *	SMART Command Transport (SCT) feature set
-       *	SCT Data Tables (AC5)
-       *	reserved 69[4]
-       *	Data Set Management TRIM supported (limit 1 block)
-       *	Deterministic read data after TRIM
-"""
-
-HDPARM_AFTER_SECURITY = b"""\
-Logical Unit WWN Device Identifier: 55cd2e40002643cf
-    NAA		: 5
-    IEEE OUI	: 5cd2e4
-    Unique ID	: 0002643cf
-Checksum: correct
-"""
-
-HDPARM_SECURITY_NOT_SUPPORTED = b"""\
-Security:
-    Master password revision code = 65534
-    not supported
-    not enabled
-    not locked
-    not frozen
-    not	expired: security count
-        supported: enhanced erase
-    4min for SECURITY ERASE UNIT. 2min for ENHANCED SECURITY ERASE UNIT.
-"""
-
-HDPARM_SECURITY_SUPPORTED_NOT_ENABLED = b"""\
-Security:
-    Master password revision code = 65534
-        supported
-    not enabled
-    not locked
-    not frozen
-    not	expired: security count
-        supported: enhanced erase
-    4min for SECURITY ERASE UNIT. 2min for ENHANCED SECURITY ERASE UNIT.
-"""
-
-HDPARM_SECURITY_SUPPORTED_ENABLED = b"""\
-Security:
-    Master password revision code = 65534
-        supported
-        enabled
-    not locked
-    not frozen
-    not	expired: security count
-        supported: enhanced erase
-    4min for SECURITY ERASE UNIT. 2min for ENHANCED SECURITY ERASE UNIT.
-"""
-
-HDPARM_SECURITY_ALL_TRUE = b"""\
-Security:
-    Master password revision code = 65534
-        supported
-        enabled
-        locked
-        frozen
-    not	expired: security count
-        supported: enhanced erase
-    4min for SECURITY ERASE UNIT. 2min for ENHANCED SECURITY ERASE UNIT.
-"""
+from snippets.tests.test_maas_wipe_defs import (
+    HDPARM_AFTER_SECURITY,
+    HDPARM_BEFORE_SECURITY,
+    HDPARM_SECURITY_ALL_TRUE,
+    HDPARM_SECURITY_NOT_SUPPORTED,
+    HDPARM_SECURITY_SUPPORTED_ENABLED,
+    HDPARM_SECURITY_SUPPORTED_NOT_ENABLED,
+    NVME_IDCTRL_EPILOGUE,
+    NVME_IDCTRL_FNA_CRYPTFORMAT_SUPPORTED,
+    NVME_IDCTRL_FNA_CRYPTFORMAT_UNSUPPORTED,
+    NVME_IDCTRL_OACS_FORMAT_SUPPORTED,
+    NVME_IDCTRL_OACS_FORMAT_UNSUPPORTED,
+    NVME_IDCTRL_ONCS_WRITEZ_SUPPORTED,
+    NVME_IDCTRL_ONCS_WRITEZ_UNSUPPORTED,
+    NVME_IDCTRL_PROLOGUE,
+)
 
 
 class TestMAASWipe(MAASTestCase):
@@ -196,11 +78,13 @@ class TestMAASWipe(MAASTestCase):
             sdb   disk  1
             sr0   rom   0
             sr1   rom   0
+            nvme0n1   disk  0
+            nvme1n1   disk  1
             """
         ).encode("ascii")
-        self.assertEqual([b"sda"], list_disks())
+        self.assertEqual([b"sda", b"nvme0n1"], list_disks())
 
-    def test_get_disk_security_info_missing(self):
+    def test_get_disk_security_info_missing_hdparm(self):
         hdparm_output = HDPARM_BEFORE_SECURITY + HDPARM_AFTER_SECURITY
         mock_check_output = self.patch(subprocess, "check_output")
         mock_check_output.return_value = hdparm_output
@@ -220,7 +104,7 @@ class TestMAASWipe(MAASTestCase):
             observered,
         )
 
-    def test_get_disk_security_info_not_supported(self):
+    def test_get_disk_security_info_not_supported_hdparm(self):
         hdparm_output = (
             HDPARM_BEFORE_SECURITY
             + HDPARM_SECURITY_NOT_SUPPORTED
@@ -244,7 +128,7 @@ class TestMAASWipe(MAASTestCase):
             observered,
         )
 
-    def test_get_disk_security_info_supported_not_enabled(self):
+    def test_get_disk_security_info_supported_not_enabled_hdparm(self):
         hdparm_output = (
             HDPARM_BEFORE_SECURITY
             + HDPARM_SECURITY_SUPPORTED_NOT_ENABLED
@@ -268,7 +152,7 @@ class TestMAASWipe(MAASTestCase):
             observered,
         )
 
-    def test_get_disk_security_info_supported_enabled(self):
+    def test_get_disk_security_info_supported_enabled_hdparm(self):
         hdparm_output = (
             HDPARM_BEFORE_SECURITY
             + HDPARM_SECURITY_SUPPORTED_ENABLED
@@ -292,7 +176,7 @@ class TestMAASWipe(MAASTestCase):
             observered,
         )
 
-    def test_get_disk_security_info_all_true(self):
+    def test_get_disk_security_info_all_true_hdparm(self):
         hdparm_output = (
             HDPARM_BEFORE_SECURITY
             + HDPARM_SECURITY_ALL_TRUE
@@ -316,7 +200,7 @@ class TestMAASWipe(MAASTestCase):
             observered,
         )
 
-    def test_get_disk_info(self):
+    def test_get_disk_info_hdparm(self):
         disk_names = [
             factory.make_name("disk").encode("ascii") for _ in range(3)
         ]
@@ -331,14 +215,243 @@ class TestMAASWipe(MAASTestCase):
             for _ in range(3)
         ]
         self.patch(
-            maas_wipe, "get_disk_security_info"
+            maas_wipe, "get_hdparm_security_info"
+        ).side_effect = security_info
+        observed = get_disk_info()
+        self.assertEqual(
+            {disk_names[i]: security_info[i] for i in range(3)}, observed
+        )
+
+    def test_get_disk_security_info_crypt_format_writez_nvme(self):
+        nvme_cli_output = (
+            NVME_IDCTRL_PROLOGUE
+            + NVME_IDCTRL_OACS_FORMAT_SUPPORTED
+            + NVME_IDCTRL_ONCS_WRITEZ_SUPPORTED
+            + NVME_IDCTRL_FNA_CRYPTFORMAT_SUPPORTED
+            + NVME_IDCTRL_EPILOGUE
+        )
+        mock_check_output = self.patch(subprocess, "check_output")
+        mock_check_output.return_value = nvme_cli_output
+        disk_name = factory.make_name("nvme").encode("ascii")
+        observered = get_disk_security_info(disk_name)
+        self.assertThat(
+            mock_check_output,
+            MockCallsMatch(
+                call(["nvme", "id-ctrl", maas_wipe.DEV_PATH % disk_name]),
+                call(["nvme", "id-ns", maas_wipe.DEV_PATH % disk_name]),
+            ),
+        )
+        self.assertEqual(
+            {
+                "format_supported": True,
+                "writez_supported": True,
+                "crypto_format": True,
+                "nsze": 0,
+                "lbaf": 0,
+                "ms": 0,
+            },
+            observered,
+        )
+
+    def test_get_disk_security_info_nocrypt_format_writez_nvme(self):
+        nvme_cli_output = (
+            NVME_IDCTRL_PROLOGUE
+            + NVME_IDCTRL_OACS_FORMAT_SUPPORTED
+            + NVME_IDCTRL_ONCS_WRITEZ_SUPPORTED
+            + NVME_IDCTRL_FNA_CRYPTFORMAT_UNSUPPORTED
+            + NVME_IDCTRL_EPILOGUE
+        )
+        mock_check_output = self.patch(subprocess, "check_output")
+        mock_check_output.return_value = nvme_cli_output
+        disk_name = factory.make_name("nvme").encode("ascii")
+        observered = get_disk_security_info(disk_name)
+        self.assertThat(
+            mock_check_output,
+            MockCallsMatch(
+                call(["nvme", "id-ctrl", maas_wipe.DEV_PATH % disk_name]),
+                call(["nvme", "id-ns", maas_wipe.DEV_PATH % disk_name]),
+            ),
+        )
+        self.assertEqual(
+            {
+                "format_supported": True,
+                "writez_supported": True,
+                "crypto_format": False,
+                "nsze": 0,
+                "lbaf": 0,
+                "ms": 0,
+            },
+            observered,
+        )
+
+    def test_get_disk_security_info_crypt_format_nowritez_nvme(self):
+        nvme_cli_output = (
+            NVME_IDCTRL_PROLOGUE
+            + NVME_IDCTRL_OACS_FORMAT_SUPPORTED
+            + NVME_IDCTRL_ONCS_WRITEZ_UNSUPPORTED
+            + NVME_IDCTRL_FNA_CRYPTFORMAT_SUPPORTED
+            + NVME_IDCTRL_EPILOGUE
+        )
+        mock_check_output = self.patch(subprocess, "check_output")
+        mock_check_output.return_value = nvme_cli_output
+        disk_name = factory.make_name("nvme").encode("ascii")
+        observered = get_disk_security_info(disk_name)
+        self.assertThat(
+            mock_check_output,
+            MockCallsMatch(
+                call(["nvme", "id-ctrl", maas_wipe.DEV_PATH % disk_name]),
+                call(["nvme", "id-ns", maas_wipe.DEV_PATH % disk_name]),
+            ),
+        )
+        self.assertEqual(
+            {
+                "format_supported": True,
+                "writez_supported": False,
+                "crypto_format": True,
+                "nsze": 0,
+                "lbaf": 0,
+                "ms": 0,
+            },
+            observered,
+        )
+
+    def test_get_disk_security_info_noformat_writez_nvme(self):
+        nvme_cli_output = (
+            NVME_IDCTRL_PROLOGUE
+            + NVME_IDCTRL_OACS_FORMAT_UNSUPPORTED
+            + NVME_IDCTRL_ONCS_WRITEZ_SUPPORTED
+            + NVME_IDCTRL_FNA_CRYPTFORMAT_SUPPORTED
+            + NVME_IDCTRL_EPILOGUE
+        )
+        mock_check_output = self.patch(subprocess, "check_output")
+        mock_check_output.return_value = nvme_cli_output
+        disk_name = factory.make_name("nvme").encode("ascii")
+        observered = get_disk_security_info(disk_name)
+        self.assertThat(
+            mock_check_output,
+            MockCallsMatch(
+                call(["nvme", "id-ctrl", maas_wipe.DEV_PATH % disk_name]),
+                call(["nvme", "id-ns", maas_wipe.DEV_PATH % disk_name]),
+            ),
+        )
+        self.assertEqual(
+            {
+                "format_supported": False,
+                "writez_supported": True,
+                "crypto_format": True,
+                "nsze": 0,
+                "lbaf": 0,
+                "ms": 0,
+            },
+            observered,
+        )
+
+    def test_get_disk_security_info_noformat_nowritez_nvme(self):
+        nvme_cli_output = (
+            NVME_IDCTRL_PROLOGUE
+            + NVME_IDCTRL_OACS_FORMAT_UNSUPPORTED
+            + NVME_IDCTRL_ONCS_WRITEZ_UNSUPPORTED
+            + NVME_IDCTRL_FNA_CRYPTFORMAT_UNSUPPORTED
+            + NVME_IDCTRL_EPILOGUE
+        )
+        mock_check_output = self.patch(subprocess, "check_output")
+        mock_check_output.return_value = nvme_cli_output
+        disk_name = factory.make_name("nvme").encode("ascii")
+        observered = get_disk_security_info(disk_name)
+        self.assertThat(
+            mock_check_output,
+            MockCallsMatch(
+                call(["nvme", "id-ctrl", maas_wipe.DEV_PATH % disk_name]),
+                call(["nvme", "id-ns", maas_wipe.DEV_PATH % disk_name]),
+            ),
+        )
+        self.assertEqual(
+            {
+                "format_supported": False,
+                "writez_supported": False,
+                "crypto_format": False,
+                "nsze": 0,
+                "lbaf": 0,
+                "ms": 0,
+            },
+            observered,
+        )
+
+    def test_get_disk_security_info_failed_cmd_nvme(self):
+        mock_check_output = self.patch(subprocess, "check_output")
+        mock_check_output.side_effect = subprocess.CalledProcessError(
+            1, "nvme id-ctrl"
+        )
+        disk_name = factory.make_name("nvme").encode("ascii")
+        observered = get_disk_security_info(disk_name)
+
+        self.assertThat(
+            self.print_flush,
+            MockCalledOnceWith("Error on nvme id-ctrl (%s)" % "1"),
+        )
+        self.assertEqual(
+            {
+                "format_supported": False,
+                "writez_supported": False,
+                "crypto_format": False,
+                "nsze": 0,
+                "lbaf": 0,
+                "ms": 0,
+            },
+            observered,
+        )
+
+    def test_get_disk_security_info_failed_os_nvme(self):
+        mock_check_output = self.patch(subprocess, "check_output")
+        mock_check_output.side_effect = OSError(
+            -2, "No such file or directory"
+        )
+        disk_name = factory.make_name("nvme").encode("ascii")
+        observered = get_disk_security_info(disk_name)
+
+        self.assertThat(
+            self.print_flush,
+            MockCalledOnceWith(
+                "OS error when running nvme-cli (No such file or directory)"
+            ),
+        )
+        self.assertEqual(
+            {
+                "format_supported": False,
+                "writez_supported": False,
+                "crypto_format": False,
+                "nsze": 0,
+                "lbaf": 0,
+                "ms": 0,
+            },
+            observered,
+        )
+
+    def test_get_disk_info_nvme(self):
+        disk_names = [
+            factory.make_name("nvme").encode("ascii") for _ in range(3)
+        ]
+        self.patch(maas_wipe, "list_disks").return_value = disk_names
+        security_info = [
+            {
+                "format_supported": True,
+                "writez_supported": True,
+                "crypto_format": True,
+                "nsze": 0,
+                "lbaf": 0,
+                "ms": 0,
+            }
+            for _ in range(3)
+        ]
+        self.patch(
+            maas_wipe, "get_nvme_security_info"
         ).side_effect = security_info
         observed = get_disk_info()
         self.assertEqual(
             {disk_names[i]: security_info[i] for i in range(3)}, observed
         )
 
-    def test_try_secure_erase_not_supported(self):
+    def test_try_secure_erase_not_supported_hdparm(self):
         disk_name = factory.make_name("disk").encode("ascii")
         disk_info = {
             b"supported": False,
@@ -355,7 +468,7 @@ class TestMAASWipe(MAASTestCase):
             ),
         )
 
-    def test_try_secure_erase_frozen(self):
+    def test_try_secure_erase_frozen_hdparm(self):
         disk_name = factory.make_name("disk").encode("ascii")
         disk_info = {
             b"supported": True,
@@ -372,7 +485,7 @@ class TestMAASWipe(MAASTestCase):
             ),
         )
 
-    def test_try_secure_erase_locked(self):
+    def test_try_secure_erase_locked_hdparm(self):
         disk_name = factory.make_name("disk").encode("ascii")
         disk_info = {
             b"supported": True,
@@ -389,7 +502,7 @@ class TestMAASWipe(MAASTestCase):
             ),
         )
 
-    def test_try_secure_erase_enabled(self):
+    def test_try_secure_erase_enabled_hdparm(self):
         disk_name = factory.make_name("disk").encode("ascii")
         disk_info = {
             b"supported": True,
@@ -406,7 +519,7 @@ class TestMAASWipe(MAASTestCase):
             ),
         )
 
-    def test_try_secure_erase_failed_erase(self):
+    def test_try_secure_erase_failed_erase_hdparm(self):
         disk_name = factory.make_name("disk").encode("ascii")
         disk_info = {
             b"supported": True,
@@ -415,7 +528,7 @@ class TestMAASWipe(MAASTestCase):
             b"frozen": False,
         }
         exception = factory.make_exception()
-        self.patch(maas_wipe, "secure_erase").side_effect = exception
+        self.patch(maas_wipe, "secure_erase_hdparm").side_effect = exception
         self.assertFalse(try_secure_erase(disk_name, disk_info))
         self.assertThat(
             self.print_flush,
@@ -425,7 +538,7 @@ class TestMAASWipe(MAASTestCase):
             ),
         )
 
-    def test_try_secure_erase_successful_erase(self):
+    def test_try_secure_erase_successful_erase_hdparm(self):
         disk_name = factory.make_name("disk").encode("ascii")
         disk_info = {
             b"supported": True,
@@ -433,7 +546,7 @@ class TestMAASWipe(MAASTestCase):
             b"locked": False,
             b"frozen": False,
         }
-        self.patch(maas_wipe, "secure_erase")
+        self.patch(maas_wipe, "secure_erase_hdparm")
         self.assertTrue(try_secure_erase(disk_name, disk_info))
         self.assertThat(
             self.print_flush,
@@ -443,7 +556,223 @@ class TestMAASWipe(MAASTestCase):
             ),
         )
 
-    def test_secure_erase_writes_known_data(self):
+    def test_try_secure_erase_not_supported_nvme(self):
+        disk_name = factory.make_name("nvme").encode("ascii")
+        sec_info = {
+            "format_supported": False,
+            "writez_supported": True,
+            "crypto_format": True,
+            "nsze": 0,
+            "lbaf": 0,
+            "ms": 0,
+        }
+        self.assertFalse(try_secure_erase(disk_name, sec_info))
+        self.assertThat(
+            self.print_flush,
+            MockCalledOnceWith(
+                "Device %s does not support formatting"
+                % disk_name.decode("ascii")
+            ),
+        )
+
+    def test_try_secure_erase_successful_cryto_nvme(self):
+        disk_name = factory.make_name("nvme").encode("ascii")
+        sec_info = {
+            "format_supported": True,
+            "writez_supported": True,
+            "crypto_format": True,
+            "nsze": 0,
+            "lbaf": 0,
+            "ms": 0,
+        }
+        mock_check_output = self.patch(subprocess, "check_output")
+        self.assertTrue(try_secure_erase(disk_name, sec_info))
+        self.assertThat(
+            mock_check_output,
+            MockCalledOnceWith(
+                [
+                    "nvme",
+                    "format",
+                    "-s",
+                    "2",
+                    "-l",
+                    "0",
+                    "-m",
+                    "0",
+                    maas_wipe.DEV_PATH % disk_name,
+                ]
+            ),
+        )
+        self.assertThat(
+            self.print_flush,
+            MockCalledOnceWith(
+                "Secure erase was successful on NVMe drive %s"
+                % disk_name.decode("ascii")
+            ),
+        )
+
+    def test_try_secure_erase_successful_nocryto_nvme(self):
+        disk_name = factory.make_name("nvme").encode("ascii")
+        sec_info = {
+            "format_supported": True,
+            "writez_supported": True,
+            "crypto_format": False,
+            "nsze": 0,
+            "lbaf": 0,
+            "ms": 0,
+        }
+        mock_check_output = self.patch(subprocess, "check_output")
+        self.assertTrue(try_secure_erase(disk_name, sec_info))
+        self.assertThat(
+            mock_check_output,
+            MockCalledOnceWith(
+                [
+                    "nvme",
+                    "format",
+                    "-s",
+                    "1",
+                    "-l",
+                    "0",
+                    "-m",
+                    "0",
+                    maas_wipe.DEV_PATH % disk_name,
+                ]
+            ),
+        )
+        self.assertThat(
+            self.print_flush,
+            MockCalledOnceWith(
+                "Secure erase was successful on NVMe drive %s"
+                % disk_name.decode("ascii")
+            ),
+        )
+
+    def test_try_secure_erase_failed_nvme(self):
+        disk_name = factory.make_name("nvme").encode("ascii")
+        sec_info = {
+            "format_supported": True,
+            "writez_supported": True,
+            "crypto_format": True,
+            "nsze": 0,
+            "lbaf": 0,
+            "ms": 0,
+        }
+        mock_check_output = self.patch(subprocess, "check_output")
+        mock_check_output.side_effect = subprocess.CalledProcessError(
+            1, "nvme format"
+        )
+
+        self.assertFalse(try_secure_erase(disk_name, sec_info))
+        self.assertThat(
+            self.print_flush,
+            MockCalledOnceWith("Error with format command (%s)" % "1"),
+        )
+
+    def test_try_write_zeroes_not_supported_nvme(self):
+        disk_name = factory.make_name("nvme").encode("ascii")
+        sec_info = {
+            "format_supported": False,
+            "writez_supported": False,
+            "crypto_format": False,
+            "nsze": 1,
+            "lbaf": 0,
+            "ms": 0,
+        }
+        mock_print = self.patch(builtins, "print")
+        self.assertFalse(nvme_write_zeroes(disk_name, sec_info))
+        self.assertThat(
+            mock_print,
+            MockCalledOnceWith(
+                "NVMe drive %s does not support write-zeroes"
+                % disk_name.decode("ascii")
+            ),
+        )
+        self.assertThat(
+            self.print_flush,
+            MockCalledOnceWith("Will fallback to regular drive zeroing."),
+        )
+
+    def test_try_write_zeroes_supported_invalid_nsze_nvme(self):
+        disk_name = factory.make_name("nvme").encode("ascii")
+        sec_info = {
+            "format_supported": False,
+            "writez_supported": True,
+            "crypto_format": False,
+            "nsze": 0,
+            "lbaf": 0,
+            "ms": 0,
+        }
+        mock_print = self.patch(builtins, "print")
+        self.assertFalse(nvme_write_zeroes(disk_name, sec_info))
+        self.assertThat(
+            mock_print,
+            MockCalledOnceWith(
+                "Bad namespace information collected on NVMe drive %s"
+                % disk_name.decode("ascii")
+            ),
+        )
+        self.assertThat(
+            self.print_flush,
+            MockCalledOnceWith("Will fallback to regular drive zeroing."),
+        )
+
+    def test_try_write_zeroes_successful_nvme(self):
+        disk_name = factory.make_name("nvme").encode("ascii")
+        sec_info = {
+            "format_supported": False,
+            "writez_supported": True,
+            "crypto_format": False,
+            "nsze": 0x100A,
+            "lbaf": 0,
+            "ms": 0,
+        }
+        mock_check_output = self.patch(subprocess, "check_output")
+        self.assertTrue(nvme_write_zeroes(disk_name, sec_info))
+        self.assertThat(
+            mock_check_output,
+            MockCalledOnceWith(
+                [
+                    "nvme",
+                    "write-zeroes",
+                    "-f",
+                    "-s",
+                    "0",
+                    "-c",
+                    "100a",
+                    maas_wipe.DEV_PATH % disk_name,
+                ]
+            ),
+        )
+        self.assertThat(
+            self.print_flush,
+            MockCalledOnceWith(
+                "%s: successfully zeroed (using write-zeroes)."
+                % disk_name.decode("ascii")
+            ),
+        )
+
+    def test_try_write_zeroes_failed_nvme(self):
+        disk_name = factory.make_name("nvme").encode("ascii")
+        sec_info = {
+            "format_supported": False,
+            "writez_supported": True,
+            "crypto_format": False,
+            "nsze": 100,
+            "lbaf": 0,
+            "ms": 0,
+        }
+        mock_check_output = self.patch(subprocess, "check_output")
+        mock_check_output.side_effect = subprocess.CalledProcessError(
+            1, "nvme write-zeroes"
+        )
+
+        self.assertFalse(nvme_write_zeroes(disk_name, sec_info))
+        self.assertThat(
+            self.print_flush,
+            MockCalledOnceWith("Error with write-zeroes command (%s)" % "1"),
+        )
+
+    def test_secure_erase_writes_known_data_hdparm(self):
         tmp_dir = self.make_dir()
         dev_path = (tmp_dir + "/%s").encode("ascii")
         self.patch(maas_wipe, "DEV_PATH", dev_path)
@@ -455,7 +784,7 @@ class TestMAASWipe(MAASTestCase):
         mock_check_output = self.patch(subprocess, "check_output")
         mock_check_output.side_effect = factory.make_exception()
 
-        self.assertRaises(WipeError, secure_erase, dev_name)
+        self.assertRaises(WipeError, secure_erase_hdparm, dev_name)
         expected_buf = b"M" * 1024 * 1024
         with open(file_path, "rb") as fp:
             read_buf = fp.read(len(expected_buf))
@@ -463,7 +792,7 @@ class TestMAASWipe(MAASTestCase):
             expected_buf, read_buf, "First 1 MiB of file was not written."
         )
 
-    def test_secure_erase_sets_security_password(self):
+    def test_secure_erase_sets_security_password_hdparm(self):
         tmp_dir = self.make_dir()
         dev_path = (tmp_dir + "/%s").encode("ascii")
         self.patch(maas_wipe, "DEV_PATH", dev_path)
@@ -476,10 +805,10 @@ class TestMAASWipe(MAASTestCase):
         # Fail to get disk info just to exit early.
         exception_type = factory.make_exception_type()
         self.patch(
-            maas_wipe, "get_disk_security_info"
+            maas_wipe, "get_hdparm_security_info"
         ).side_effect = exception_type()
 
-        self.assertRaises(exception_type, secure_erase, dev_name)
+        self.assertRaises(exception_type, secure_erase_hdparm, dev_name)
         self.assertThat(
             mock_check_output,
             MockCalledOnceWith(
@@ -494,7 +823,7 @@ class TestMAASWipe(MAASTestCase):
             ),
         )
 
-    def test_secure_erase_fails_if_not_enabled(self):
+    def test_secure_erase_fails_if_not_enabled_hdparm(self):
         tmp_dir = self.make_dir()
         dev_path = (tmp_dir + "/%s").encode("ascii")
         self.patch(maas_wipe, "DEV_PATH", dev_path)
@@ -503,16 +832,16 @@ class TestMAASWipe(MAASTestCase):
         self.make_empty_file(file_path)
 
         self.patch(subprocess, "check_output")
-        self.patch(maas_wipe, "get_disk_security_info").return_value = {
+        self.patch(maas_wipe, "get_hdparm_security_info").return_value = {
             b"enabled": False
         }
 
-        error = self.assertRaises(WipeError, secure_erase, dev_name)
+        error = self.assertRaises(WipeError, secure_erase_hdparm, dev_name)
         self.assertEqual(
             "Failed to enable security to perform secure erase.", str(error)
         )
 
-    def test_secure_erase_fails_when_still_enabled(self):
+    def test_secure_erase_fails_when_still_enabled_hdparm(self):
         tmp_dir = self.make_dir()
         dev_path = (tmp_dir + "/%s").encode("ascii")
         self.patch(maas_wipe, "DEV_PATH", dev_path)
@@ -521,14 +850,14 @@ class TestMAASWipe(MAASTestCase):
         self.make_empty_file(file_path)
 
         mock_check_output = self.patch(subprocess, "check_output")
-        self.patch(maas_wipe, "get_disk_security_info").return_value = {
+        self.patch(maas_wipe, "get_hdparm_security_info").return_value = {
             b"enabled": True
         }
         exception = factory.make_exception()
         mock_check_call = self.patch(subprocess, "check_call")
         mock_check_call.side_effect = exception
 
-        error = self.assertRaises(WipeError, secure_erase, dev_name)
+        error = self.assertRaises(WipeError, secure_erase_hdparm, dev_name)
         self.assertThat(
             mock_check_call,
             MockCalledOnceWith(
@@ -561,7 +890,7 @@ class TestMAASWipe(MAASTestCase):
         self.assertEqual("Failed to securely erase.", str(error))
         self.assertEqual(exception, error.__cause__)
 
-    def test_secure_erase_fails_when_buffer_not_different(self):
+    def test_secure_erase_fails_when_buffer_not_different_hdparm(self):
         tmp_dir = self.make_dir()
         dev_path = (tmp_dir + "/%s").encode("ascii")
         self.patch(maas_wipe, "DEV_PATH", dev_path)
@@ -570,13 +899,13 @@ class TestMAASWipe(MAASTestCase):
         self.make_empty_file(file_path)
 
         self.patch(subprocess, "check_output")
-        self.patch(maas_wipe, "get_disk_security_info").side_effect = [
+        self.patch(maas_wipe, "get_hdparm_security_info").side_effect = [
             {b"enabled": True},
             {b"enabled": False},
         ]
         mock_check_call = self.patch(subprocess, "check_call")
 
-        error = self.assertRaises(WipeError, secure_erase, dev_name)
+        error = self.assertRaises(WipeError, secure_erase_hdparm, dev_name)
         self.assertThat(
             mock_check_call,
             MockCalledOnceWith(
@@ -595,7 +924,7 @@ class TestMAASWipe(MAASTestCase):
             str(error),
         )
 
-    def test_secure_erase_fails_success(self):
+    def test_secure_erase_fails_success_hdparm(self):
         tmp_dir = self.make_dir()
         dev_path = (tmp_dir + "/%s").encode("ascii")
         self.patch(maas_wipe, "DEV_PATH", dev_path)
@@ -604,7 +933,7 @@ class TestMAASWipe(MAASTestCase):
         self.make_empty_file(file_path)
 
         self.patch(subprocess, "check_output")
-        self.patch(maas_wipe, "get_disk_security_info").side_effect = [
+        self.patch(maas_wipe, "get_hdparm_security_info").side_effect = [
             {b"enabled": True},
             {b"enabled": False},
         ]
@@ -620,9 +949,9 @@ class TestMAASWipe(MAASTestCase):
         mock_check_call.side_effect = wipe_buffer
 
         # No error should be raised.
-        secure_erase(dev_name)
+        secure_erase_hdparm(dev_name)
 
-    def test_wipe_quickly(self):
+    def test_wipe_quickly_successful(self):
         tmp_dir = self.make_dir()
         dev_path = (tmp_dir + "/%s").encode("ascii")
         self.patch(maas_wipe, "DEV_PATH", dev_path)
@@ -630,7 +959,14 @@ class TestMAASWipe(MAASTestCase):
         file_path = dev_path % dev_name
         self.make_empty_file(file_path, content=b"T")
 
+        mock_check_output = self.patch(subprocess, "check_output")
         wipe_quickly(dev_name)
+        self.assertThat(
+            mock_check_output,
+            MockCalledOnceWith(
+                ["wipefs", "-f", "-a", maas_wipe.DEV_PATH % dev_name]
+            ),
+        )
 
         buf_size = 1024 * 1024
         with open(file_path, "rb") as fp:
@@ -641,8 +977,18 @@ class TestMAASWipe(MAASTestCase):
         zero_buf = b"\0" * 1024 * 1024
         self.assertEqual(zero_buf, first_buf, "First 1 MiB was not wiped.")
         self.assertEqual(zero_buf, last_buf, "Last 1 MiB was not wiped.")
+        self.assertThat(
+            self.print_flush,
+            MockCallsMatch(
+                call("%s: starting quick wipe." % dev_name.decode("ascii")),
+                call(
+                    "%s: successfully quickly wiped."
+                    % dev_name.decode("ascii")
+                ),
+            ),
+        )
 
-    def test_zero_disk(self):
+    def test_wipe_quickly_successful_but_wipefs_failed(self):
         tmp_dir = self.make_dir()
         dev_path = (tmp_dir + "/%s").encode("ascii")
         self.patch(maas_wipe, "DEV_PATH", dev_path)
@@ -650,13 +996,83 @@ class TestMAASWipe(MAASTestCase):
         file_path = dev_path % dev_name
         self.make_empty_file(file_path, content=b"T")
 
+        mock_check_output = self.patch(subprocess, "check_output")
+        mock_check_output.side_effect = subprocess.CalledProcessError(
+            1, "wipefs"
+        )
+        wipe_quickly(dev_name)
+
+        buf_size = 1024 * 1024
+        with open(file_path, "rb") as fp:
+            first_buf = fp.read(buf_size)
+            fp.seek(-buf_size, 2)
+            last_buf = fp.read(buf_size)
+
+        zero_buf = b"\0" * 1024 * 1024
+        self.assertEqual(zero_buf, first_buf, "First 1 MiB was not wiped.")
+        self.assertEqual(zero_buf, last_buf, "Last 1 MiB was not wiped.")
+        self.assertThat(
+            self.print_flush,
+            MockCallsMatch(
+                call("%s: starting quick wipe." % dev_name.decode("ascii")),
+                call("%s: wipefs failed (1)" % dev_name.decode("ascii")),
+                call(
+                    "%s: successfully quickly wiped."
+                    % dev_name.decode("ascii")
+                ),
+            ),
+        )
+
+    def test_wipe_quickly_failed(self):
+        dev_name = factory.make_name("disk").encode("ascii")
+
+        mock_check_output = self.patch(subprocess, "check_output")
+        mock_check_output.side_effect = subprocess.CalledProcessError(
+            1, "wipefs"
+        )
+
+        mock_os_open = self.patch(builtins, "open")
+        mock_os_open.side_effect = OSError(-2, "No such file or directory")
+
+        wipe_quickly(dev_name)
+
+        self.assertThat(
+            self.print_flush,
+            MockCallsMatch(
+                call("%s: starting quick wipe." % dev_name.decode("ascii")),
+                call("%s: wipefs failed (1)" % dev_name.decode("ascii")),
+                call(
+                    "%s: OS error while wiping beginning/end of disk (No such file or directory)"
+                    % dev_name.decode("ascii")
+                ),
+                call(
+                    "%s: failed to be quickly wiped."
+                    % dev_name.decode("ascii")
+                ),
+            ),
+        )
+
+    def test_zero_disk_hdd(self):
+        tmp_dir = self.make_dir()
+        dev_path = (tmp_dir + "/%s").encode("ascii")
+        self.patch(maas_wipe, "DEV_PATH", dev_path)
+        dev_name = factory.make_name("disk").encode("ascii")
+        file_path = dev_path % dev_name
+        self.make_empty_file(file_path, content=b"T")
+        disk_info = {
+            b"supported": True,
+            b"enabled": False,
+            b"locked": False,
+            b"frozen": False,
+        }
+
         # Add a little size to the file making it not evenly
         # divisable by 1 MiB.
         extra_end = 512
         with open(file_path, "a+b") as fp:
             fp.write(b"T" * extra_end)
 
-        zero_disk(dev_name)
+        zero_disk(dev_name, disk_info)
 
         zero_buf = b"\0" * 1024 * 1024
         with open(file_path, "rb") as fp:
@@ -682,7 +1098,7 @@ class TestMAASWipe(MAASTestCase):
         parser.parse_args.return_value = args
         self.patch(argparse, "ArgumentParser").return_value = parser
 
-    def test_main_calls_try_secure_erase_for_all_disks(self):
+    def test_main_calls_try_secure_erase_for_all_hdd(self):
         self.patch_args(True, False)
         disks = {
             factory.make_name("disk").encode("ascii"): {} for _ in range(3)
@@ -698,7 +1114,7 @@ class TestMAASWipe(MAASTestCase):
         self.assertThat(mock_try, MockCallsMatch(*calls))
         self.assertThat(mock_zero, MockNotCalled())
 
-    def test_main_calls_zero_disk_if_no_secure_erase(self):
+    def test_main_calls_zero_disk_if_no_secure_erase_hdd(self):
         self.patch_args(True, False)
         disks = {
             factory.make_name("disk").encode("ascii"): {} for _ in range(3)
@@ -711,11 +1127,10 @@ class TestMAASWipe(MAASTestCase):
         maas_wipe.main()
 
         try_calls = [call(disk, info) for disk, info in disks.items()]
-        wipe_calls = [call(disk) for disk in disks.keys()]
         self.assertThat(mock_try, MockCallsMatch(*try_calls))
-        self.assertThat(mock_zero, MockCallsMatch(*wipe_calls))
+        self.assertThat(mock_zero, MockCallsMatch(*try_calls))
 
-    def test_main_calls_wipe_quickly_if_no_secure_erase(self):
+    def test_main_calls_wipe_quickly_if_no_secure_erase_hdd(self):
         self.patch_args(True, True)
         disks = {
             factory.make_name("disk").encode("ascii"): {} for _ in range(3)
@@ -760,6 +1175,6 @@ class TestMAASWipe(MAASTestCase):
         mock_try.return_value = False
         maas_wipe.main()
 
-        wipe_calls = [call(disk) for disk in disks.keys()]
+        wipe_calls = [call(disk, info) for disk, info in disks.items()]
         self.assertThat(mock_try, MockNotCalled())
         self.assertThat(zero_disk, MockCallsMatch(*wipe_calls))
diff --git a/src/metadataserver/user_data/templates/snippets/tests/test_maas_wipe_defs.py b/src/metadataserver/user_data/templates/snippets/tests/test_maas_wipe_defs.py
new file mode 100644
index 0000000..f9bb440
--- /dev/null
+++ b/src/metadataserver/user_data/templates/snippets/tests/test_maas_wipe_defs.py
@@ -0,0 +1,232 @@
+#!/usr/bin/python3
+# Copyright 2020 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+#
+# hdparm / nvme-cli outputs used on maas_wipe testing
+
+HDPARM_BEFORE_SECURITY = b"""\
+/dev/sda:
+
+ATA device, with non-removable media
+    Model Number:       INTEL SSDSC2CT240A4
+    Serial Number:      CVKI3206029X240DGN
+    Firmware Revision:  335u
+    Transport:          Serial, ATA8-AST, SATA 1.0a, SATA II Extensions
+Standards:
+    Used: unknown (minor revision code 0xffff)
+    Supported: 9 8 7 6 5
+    Likely used: 9
+Configuration:
+    Logical		max	current
+    cylinders	16383	16383
+    heads		16	16
+    sectors/track	63	63
+    --
+    CHS current addressable sectors:   16514064
+    LBA    user addressable sectors:  268435455
+    LBA48  user addressable sectors:  468862128
+    Logical  Sector size:                   512 bytes
+    Physical Sector size:                   512 bytes
+    Logical Sector-0 offset:                  0 bytes
+    device size with M = 1024*1024:      228936 MBytes
+    device size with M = 1000*1000:      240057 MBytes (240 GB)
+    cache/buffer size  = unknown
+    Nominal Media Rotation Rate: Solid State Device
+Capabilities:
+    LBA, IORDY(can be disabled)
+    Queue depth: 32
+    Standby timer values: spec'd by Standard, no device specific minimum
+    R/W multiple sector transfer: Max = 16	Current = 16
+    Advanced power management level: 254
+    DMA: mdma0 mdma1 mdma2 udma0 udma1 udma2 udma3 udma4 udma5 *udma6
+         Cycle time: min=120ns recommended=120ns
+    PIO: pio0 pio1 pio2 pio3 pio4
+         Cycle time: no flow control=120ns  IORDY flow control=120ns
+Commands/features:
+    Enabled	Supported:
+       *	SMART feature set
+            Security Mode feature set
+       *	Power Management feature set
+       *	Write cache
+       *	Look-ahead
+       *	Host Protected Area feature set
+       *	WRITE_BUFFER command
+       *	READ_BUFFER command
+       *	NOP cmd
+       *	DOWNLOAD_MICROCODE
+       *	Advanced Power Management feature set
+            Power-Up In Standby feature set
+       *	48-bit Address feature set
+       *	Mandatory FLUSH_CACHE
+       *	FLUSH_CACHE_EXT
+       *	SMART error logging
+       *	SMART self-test
+       *	General Purpose Logging feature set
+       *	WRITE_{DMA|MULTIPLE}_FUA_EXT
+       *	64-bit World wide name
+       *	IDLE_IMMEDIATE with UNLOAD
+       *	WRITE_UNCORRECTABLE_EXT command
+       *	{READ,WRITE}_DMA_EXT_GPL commands
+       *	Segmented DOWNLOAD_MICROCODE
+       *	Gen1 signaling speed (1.5Gb/s)
+       *	Gen2 signaling speed (3.0Gb/s)
+       *	Gen3 signaling speed (6.0Gb/s)
+       *	Native Command Queueing (NCQ)
+       *	Host-initiated interface power management
+       *	Phy event counters
+       *	DMA Setup Auto-Activate optimization
+            Device-initiated interface power management
+       *	Software settings preservation
+       *	SMART Command Transport (SCT) feature set
+       *	SCT Data Tables (AC5)
+       *	reserved 69[4]
+       *	Data Set Management TRIM supported (limit 1 block)
+       *	Deterministic read data after TRIM
+"""
+
+HDPARM_AFTER_SECURITY = b"""\
+Logical Unit WWN Device Identifier: 55cd2e40002643cf
+    NAA		: 5
+    IEEE OUI	: 5cd2e4
+    Unique ID	: 0002643cf
+Checksum: correct
+"""
+
+HDPARM_SECURITY_NOT_SUPPORTED = b"""\
+Security:
+    Master password revision code = 65534
+    not supported
+    not enabled
+    not locked
+    not frozen
+    not	expired: security count
+        supported: enhanced erase
+    4min for SECURITY ERASE UNIT. 2min for ENHANCED SECURITY ERASE UNIT.
+"""
+
+HDPARM_SECURITY_SUPPORTED_NOT_ENABLED = b"""\
+Security:
+    Master password revision code = 65534
+        supported
+    not enabled
+    not locked
+    not frozen
+    not	expired: security count
+        supported: enhanced erase
+    4min for SECURITY ERASE UNIT. 2min for ENHANCED SECURITY ERASE UNIT.
+"""
+
+HDPARM_SECURITY_SUPPORTED_ENABLED = b"""\
+Security:
+    Master password revision code = 65534
+        supported
+        enabled
+    not locked
+    not frozen
+    not	expired: security count
+        supported: enhanced erase
+    4min for SECURITY ERASE UNIT. 2min for ENHANCED SECURITY ERASE UNIT.
+"""
+
+HDPARM_SECURITY_ALL_TRUE = b"""\
+Security:
+    Master password revision code = 65534
+        supported
+        enabled
+        locked
+        frozen
+    not	expired: security count
+        supported: enhanced erase
+    4min for SECURITY ERASE UNIT. 2min for ENHANCED SECURITY ERASE UNIT.
+"""
+
+NVME_IDCTRL_PROLOGUE = b"""\
+NVME Identify Controller:
+vid     : 0x8086
+ssvid   : 0x8086
+sn      : CVMD5066002T400AGN
+mn      : INTEL SSDPEDME400G4
+fr      : 8DV10131
+rab     : 0
+ieee    : 5cd2e4
+cmic    : 0
+mdts    : 5
+cntlid  : 0
+ver     : 0
+rtd3r   : 0
+rtd3e   : 0
+oaes    : 0
+ctratt  : 0
+acl     : 3
+aerl    : 3
+frmw    : 0x2
+lpa     : 0
+elpe    : 63
+npss    : 0
+avscc   : 0
+apsta   : 0
+wctemp  : 0
+cctemp  : 0
+mtfa    : 0
+hmpre   : 0
+hmmin   : 0
+tnvmcap : 0
+unvmcap : 0
+rpmbs   : 0
+edstt   : 0
+dsto    : 0
+fwug    : 0
+kas     : 0
+hctma   : 0
+mntmt   : 0
+mxtmt   : 0
+sanicap : 0
+hmminds : 0
+hmmaxd  : 0
+sqes    : 0x66
+cqes    : 0x44
+maxcmd  : 0
+nn      : 1
+fuses   : 0
+"""
+
+NVME_IDCTRL_OACS_FORMAT_SUPPORTED = b"""\
+oacs    : 0x6
+"""
+
+NVME_IDCTRL_OACS_FORMAT_UNSUPPORTED = b"""\
+oacs    : 0x4
+"""
+
+NVME_IDCTRL_ONCS_WRITEZ_SUPPORTED = b"""\
+oncs    : 0xe
+"""
+
+NVME_IDCTRL_ONCS_WRITEZ_UNSUPPORTED = b"""\
+oncs    : 0x6
+"""
+
+NVME_IDCTRL_FNA_CRYPTFORMAT_SUPPORTED = b"""\
+fna     : 0x7
+"""
+
+NVME_IDCTRL_FNA_CRYPTFORMAT_UNSUPPORTED = b"""\
+fna     : 0x3
+"""
+
+NVME_IDCTRL_EPILOGUE = b"""\
+vwc     : 0
+awun    : 0
+awupf   : 0
+nvscc   : 0
+acwu    : 0
+sgls    : 0
+subnqn  :
+ioccsz  : 0
+iorcsz  : 0
+icdoff  : 0
+ctrattr : 0
+msdbd   : 0
+ps    0 : mp:25.00W operational enlat:0 exlat:0 rrt:0 rrl:0
+          rwt:0 rwl:0 idle_power:- active_power:-
+"""