From f896aa679368a3beb14b81fc89cac88a02af3a25 Mon Sep 17 00:00:00 2001 From: Katarzyna Treder Date: Fri, 13 Sep 2024 14:17:05 +0200 Subject: [PATCH] Add test for loading corrupted metadata Signed-off-by: Katarzyna Treder --- .../security/test_load_corrupted_metadata.py | 123 ++++++++++++++++++ 1 file changed, 123 insertions(+) create mode 100644 test/functional/tests/security/test_load_corrupted_metadata.py diff --git a/test/functional/tests/security/test_load_corrupted_metadata.py b/test/functional/tests/security/test_load_corrupted_metadata.py new file mode 100644 index 000000000..a7bf24a9b --- /dev/null +++ b/test/functional/tests/security/test_load_corrupted_metadata.py @@ -0,0 +1,123 @@ +# +# Copyright(c) 2021 Intel Corporation +# Copyright(c) 2023-2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest +import random + +from api.cas import casadm +from api.cas.cache_config import ( + CacheMode, + CacheLineSize, + CleaningPolicy, + PromotionPolicy, + SeqCutOffPolicy, +) +from storage_devices.device import Device +from core.test_run import TestRun +from storage_devices.ramdisk import RamDisk +from test_tools.dd import Dd +from test_utils.output import CmdException +from test_utils.size import Size, Unit + + +@pytest.mark.parametrizex("cache_mode", CacheMode) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("cleaning_policy", CleaningPolicy) +@pytest.mark.parametrizex("promotion_policy", PromotionPolicy) +@pytest.mark.parametrizex("seq_cutoff_policy", SeqCutOffPolicy) +def test_load_corrupted_metadata(cache_mode, cache_line_size, cleaning_policy, + promotion_policy, seq_cutoff_policy): + """ + title: Security test for loading cache with corrupted metadata. + description: | + Validate the ability of Open CAS to load cache instance and work if metadata is corrupted. + pass_criteria: + - If metadata is recognized as corrupted, load operation should be aborted. + - If not then executing I/O operations on CAS device should not result in kernel panic. + """ + + iteration_per_config = 5 + + with TestRun.step("Prepare RAM devices for test."): + cache_dev, core_dev = RamDisk.create(disk_size=Size(100, Unit.MiB), disk_count=2) + + for iteration in TestRun.iteration( + range(iteration_per_config), + f"Corrupt metadata {iteration_per_config} times." + ): + + with TestRun.step("Prepare cache and core."): + cache = casadm.start_cache(cache_dev, cache_mode, cache_line_size, force=True) + core = casadm.add_core(cache, core_dev) + metadata_size = int(cache.metadata_size_on_disk.get_value()) + + with TestRun.step("Configure cache."): + cache.set_cleaning_policy(cleaning_policy) + cache.set_promotion_policy(promotion_policy) + cache.set_seq_cutoff_policy(seq_cutoff_policy) + + with TestRun.step("Fill cache with random data."): + Dd().input('/dev/urandom') \ + .output(core.path) \ + .block_size(Size(1, Unit.Blocks512)) \ + .oflag('direct') \ + .run() + + with TestRun.step("Stop cache without flush."): + cache.stop(True) + + with TestRun.step("Corrupt metadata."): + corrupt_metadata(cache_dev, iteration, metadata_size) + + with TestRun.step("Try to load cache."): + loaded = False + try: + casadm.load_cache(cache_dev) + loaded = True + TestRun.LOGGER.info("Cache is loaded.") + except CmdException: + TestRun.LOGGER.info("Cache is not loaded.") + + if loaded: + with TestRun.step("Run random I/O traffic to cache."): + try: + Dd().input('/dev/urandom') \ + .output(core.path) \ + .block_size(Size(1, Unit.Blocks512)) \ + .oflag('direct') \ + .run() + except CmdException: + TestRun.LOGGER.error("Sending I/O requests to cache caused error.") + cache.stop() + break + + with TestRun.step("Stop cache."): + cache.stop() + + with TestRun.step("Delete symlinks and unload 'brd' module."): + RamDisk.remove_all() + + +def corrupt_metadata(cache_dev: Device, iteration: int, metadata_size: int): + superblock_max_bytes = int(Size(8, Unit.KiB).get_value()) + number_of_bits_to_corrupt = random.randint(1, 10) + corrupted_bytes = [] + for i in range(number_of_bits_to_corrupt): + random_mask = 1 << random.randrange(0, 7) + random_offset = random.randrange( + 0, superblock_max_bytes if iteration % 100 == 0 else metadata_size + ) + corrupted_bytes.append(random_offset) + corrupt_bits(cache_dev, random_mask, random_offset) + corrupted_bytes.sort() + TestRun.LOGGER.info(f"Corrupted bytes: {corrupted_bytes}") + + +def corrupt_bits(cache_dev: Device, mask: int, offset: int): + output = TestRun.executor.run(f"xxd -bits -len 1 -seek {offset} -postscript {cache_dev.path}") + corrupt_cmd = f"printf '%02x' $((0x{output.stdout}^{mask}))" + corrupt_cmd = f"{corrupt_cmd} | xxd -revert -postscript -seek {offset} - {cache_dev.path}" + TestRun.executor.run_expect_success(corrupt_cmd)