tests/TestWindowsWipe.py
Rama master · réplica de lectura del repositorio original.
# SPDX-License-Identifier: GPL-3.0-or-later
# Copyright (c) 2008-2026 Andrew Ziem.
#
# This work is licensed under the terms of the GNU GPL, version 3 or
# later. See the COPYING file in the top-level directory.
"""
Test case for module WindowsWipe
"""
# standard library
import os
import struct
import sys
import time
import unittest
from contextlib import suppress
from unittest import mock
import bleachbit
# local
from tests import common
from bleachbit.FileUtilities import children_in_directory, extended_path
# third party
if bleachbit.IS_WINDOWS:
import win32file
import pywintypes
from win32com.shell import shell
if bleachbit.IS_WINDOWS:
# importing WindowsWipe fails on non-Windows
# pylint: disable-next=ungrouped-imports
from bleachbit.WindowsWipe import (
check_os,
determine_win_version,
open_file,
close_file,
get_file_basic_info,
truncate_file,
volume_from_file,
get_volume_information,
get_extents,
write_zero_fill,
file_wipe,
wipe_file_direct,
extents_a_minus_b,
poll_clusters_freed,
wipe_extent_by_defrag,
clean_up,
GENERIC_READ,
GENERIC_WRITE
)
# pylint: disable-next=ungrouped-imports
from tests.TestWindows import WindowsLinksMixIn
else:
WindowsLinksMixIn = object
@common.skipUnlessWindows
class WindowsWipeTestCase(common.BleachbitTestCase, WindowsLinksMixIn):
"""Test case for module WindowsWipe"""
def test_check_os(self):
"""Unit test for check_os()"""
# This should not raise an exception on Windows
# pylint: disable-next=possibly-used-before-assignment
check_os()
def test_determine_win_version(self):
"""Unit test for determine_win_version"""
# pylint: disable-next=possibly-used-before-assignment
version, is_win_home = determine_win_version()
self.assertIsInstance(version, str)
self.assertIsInstance(is_win_home, bool)
def test_open_close_file(self):
"""Unit tests for open_file() and close_file"""
open_path = os.path.join(self.tempdir, 'open')
self.write_file(open_path, b'test data')
# pylint: disable-next=possibly-used-before-assignment
file_handle = open_file(open_path)
self.assertIsNotNone(file_handle)
# pylint: disable-next=possibly-used-before-assignment
close_file(file_handle)
def test_open_file_refuses_symlink(self):
"""open_file() must not follow a symlink to another file"""
target = self.write_file('open_file_target', b'keepme')
link = os.path.join(self.tempdir, 'open_file_link')
self._create_win_file_symlink(target, link)
with self.assertRaises(OSError):
open_file(link)
def test_get_extents_system32(self):
"""Unit test for get_extents() with system32 files"""
error_count = 0
zero_extents_count = 0
nonzero_extents_count = 0
heuristic_exceptions = []
large_files_count = 0
large_files_zero_extents = []
start_time = time.time()
for path in children_in_directory(os.path.expandvars(r'%windir%\system32')):
try:
file_handle = open_file(path)
# pylint: disable-next=possibly-used-before-assignment
except pywintypes.error as e:
# 5 = ERROR_ACCESS_DENIED
# 32 = ERROR_SHARING_VIOLATION
# 1920 = ERROR_FILE_ENCRYPTED (reparse points / AppExecutionAlias
# stubs under WindowsApps are not openable).
if e.winerror in (5, 32, 1920):
error_count += 1
continue
print(f"Error opening {path}: {e}")
raise e
except PermissionError:
# open_file() refuses a symlink (e.g. Docker Desktop's
# docker.exe in system32), which is expected here.
error_count += 1
continue
# pylint: disable-next=possibly-used-before-assignment
ret = get_extents(file_handle, filename=path)
self.assertIsInstance(ret, list)
close_file(file_handle)
if len(ret) == 0:
zero_extents_count += 1
else:
nonzero_extents_count += 1
# Files with size under ~500 bytes are usually resident in the
# MFT, so they have zero clusters. There may be a few exceptions.
fsize = os.path.getsize(path)
if fsize > 1000:
large_files_count += 1
if len(ret) == 0:
# WOF-compressed (Compact OS) files report zero extents
# without setting FILE_ATTRIBUTE_COMPRESSED.
large_files_zero_extents.append((path, fsize, len(ret)))
elif fsize < 200 and len(ret) != 0:
heuristic_exceptions.append(
(path, fsize, len(ret)))
total_files = zero_extents_count + nonzero_extents_count + error_count
elapsed_seconds = time.time() - start_time
if total_files > 0:
files_per_second = total_files / elapsed_seconds
else:
files_per_second = None
zero_frac = (len(large_files_zero_extents) / large_files_count
if large_files_count else 0)
print(f"\ntest_get_extents() stats: {error_count:,} errors; {zero_extents_count:,} files with zero extents; {nonzero_extents_count:,} files with nonzero extents; {int(elapsed_seconds)} seconds; {int(files_per_second) if files_per_second else None} files per second")
print(
f"large files: {large_files_count:,}; large files with zero extents: {len(large_files_zero_extents):,} ({zero_frac:.0%})")
self.assertGreater(zero_extents_count, 10)
self.assertGreater(nonzero_extents_count, 100)
# Print the first few heuristic exceptions for diagnostics.
for path, fsize, extent_count in heuristic_exceptions[:5]:
print(
f" heuristic exception: {path} size={fsize:,} extents={extent_count}")
self.assertLess(
len(heuristic_exceptions), 6,
f"Too many heuristic exceptions ({len(heuristic_exceptions)}): "
f"{heuristic_exceptions[:5]}")
# On Compact OS images most of system32 is WOF-compressed and reports
# zero extents, so only check this when WOF isn't dominating.
for path, fsize, extent_count in large_files_zero_extents[:5]:
print(f" large zero-extent file: {path} size={fsize:,}")
if zero_frac < 0.5:
self.assertLess(
len(large_files_zero_extents), 6,
f"Too many large files with zero extents "
f"({len(large_files_zero_extents)}): {large_files_zero_extents[:5]}")
else:
print(
"skipping large-file extent check: volume appears WOF-compressed (Compact OS)")
def test_get_extents_many_runs(self):
"""get_extents() reads more runs than fit in a 2 MB buffer"""
run_count = 200000
runs = b''.join(struct.pack('qq', (i + 1) * 8, 1000 + i * 8)
for i in range(run_count))
def device_io_control(_handle, _code, _input, out_size):
if out_size < 16 + len(runs):
# pylint: disable-next=possibly-used-before-assignment
raise pywintypes.error(234, 'DeviceIoControl',
'More data is available.')
return struct.pack('IIq', run_count, 0, 0) + runs
with mock.patch('bleachbit.WindowsWipe.DeviceIoControl',
side_effect=device_io_control):
extents = get_extents(None)
self.assertEqual(len(extents), run_count)
def test_get_extents_buffer_cap(self):
"""get_extents() caps buffer growth and does not catch non-Windows errors"""
sizes = []
def always_more_data(_handle, _code, _input, out_size):
sizes.append(out_size)
# pylint: disable-next=possibly-used-before-assignment
raise pywintypes.error(234, 'DeviceIoControl',
'More data is available.')
with mock.patch('bleachbit.WindowsWipe.DeviceIoControl',
side_effect=always_more_data):
with self.assertRaises(RuntimeError) as cm:
get_extents(None, filename='huge.bin')
self.assertEqual(
str(cm.exception),
"Failed to get retrieval pointers for file 'huge.bin'")
self.assertEqual(
sizes,
[4 * 1024, 32 * 1024, 256 * 1024,
2 * 1024**2, 16 * 1024**2, 128 * 1024**2])
with mock.patch('bleachbit.WindowsWipe.DeviceIoControl',
side_effect=MemoryError):
with self.assertRaises(MemoryError):
get_extents(None, filename='huge.bin')
def test_get_file_basic_info(self):
"""Unit test for get_file_basic_info()"""
path = self.write_file(filename='test.txt', text='test data')
file_handle = open_file(path)
# pylint: disable-next=possibly-used-before-assignment
file_size, is_special = get_file_basic_info(path, file_handle)
self.assertEqual(file_size, 9)
self.assertFalse(is_special)
close_file(file_handle)
from bleachbit.General import run_external
compact_cmd = ["compact", "/c", path]
(rc, _, _) = run_external(compact_cmd)
self.assertEqual(0, rc)
file_handle = open_file(path)
_, is_special_after = get_file_basic_info(path, file_handle)
close_file(file_handle)
self.assertTrue(is_special_after)
sparse_path = os.path.join(self.tempdir, 'sparse')
self.write_file(sparse_path, b'test data')
fsutil_cmd = ["fsutil", "sparse", "setflag", sparse_path]
(rc, _, _) = run_external(fsutil_cmd)
self.assertEqual(0, rc)
file_handle = open_file(sparse_path)
_, is_sparse = get_file_basic_info(sparse_path, file_handle)
close_file(file_handle)
self.assertTrue(is_sparse)
def test_truncate_file(self):
"""Unit test for truncate_file()"""
truncate_path = os.path.join(self.tempdir, 'truncate')
self.write_file(truncate_path, b'test data to be truncated')
self.assertGreater(os.path.getsize(truncate_path), 0)
# pylint: disable-next=possibly-used-before-assignment
file_handle = open_file(truncate_path, GENERIC_READ | GENERIC_WRITE)
# pylint: disable-next=possibly-used-before-assignment
truncate_file(file_handle)
close_file(file_handle)
self.assertEqual(os.path.getsize(truncate_path), 0)
def test_get_volume_information(self):
"""Unit test for get_volume_information"""
# pylint: disable-next=possibly-used-before-assignment
volume = volume_from_file(sys.executable)
self.assertIsInstance(volume, str)
self.assertTrue(volume.endswith('\\'))
self.assertTrue(os.path.exists(volume))
self.assertEqual(len(volume), 3)
# pylint: disable-next=possibly-used-before-assignment
volume_info = get_volume_information(volume)
self.assertIsNotNone(volume_info)
self.assertGreater(volume_info.sectors_per_cluster, 0)
self.assertGreater(volume_info.bytes_per_sector, 0)
self.assertGreater(volume_info.total_clusters, 0)
def test_volume_from_file(self):
"""volume_from_file() returns the drive root, also as a fallback"""
filename = self.write_file('volume_from_file', b'x')
path = extended_path(filename)
drive = os.path.splitdrive(filename)[0] + os.sep
self.assertEqual(volume_from_file(path), drive)
# pylint: disable-next=possibly-used-before-assignment
too_long = pywintypes.error(206, 'GetVolumePathName', 'too long')
for kwargs in ({'side_effect': too_long},
{'return_value': '\\\\?\\UNC\\server\\share\\'}):
with self.subTest(**kwargs), \
mock.patch('bleachbit.WindowsWipe.GetVolumePathName', **kwargs):
self.assertEqual(volume_from_file(path), drive)
def test_volume_from_file_junction(self):
"""volume_from_file() looks up the volume past a junction"""
base = self.mkdtemp()
target = self.mkdir(os.path.join(base, 'junction_target'))
link = os.path.join(base, 'junction_link')
self._create_win_junction(target, link)
self.write_file(os.path.join(target, 'x'), b'x')
with mock.patch('bleachbit.WindowsWipe.GetVolumePathName',
# pylint: disable-next=possibly-used-before-assignment
wraps=win32file.GetVolumePathName) as mock_volume_path:
volume_from_file(extended_path(os.path.join(link, 'x')))
resolved = mock_volume_path.call_args.args[0]
self.assertIn('junction_target', resolved)
self.assertNotIn('junction_link', resolved)
def test_write_zero_fill(self):
"""Unit test for write_zero_fill"""
tmp_path = os.path.join(self.tempdir, 'write_zero_fill')
self.write_file(tmp_path, b'test data')
file_handle = open_file(tmp_path, GENERIC_READ | GENERIC_WRITE)
write_length = 1024
# pylint: disable-next=possibly-used-before-assignment
write_zero_fill(file_handle, write_length)
close_file(file_handle)
self.assertEqual(os.path.getsize(tmp_path), write_length)
with open(tmp_path, 'rb') as f:
data = f.read()
self.assertEqual(data, b'\x00' * write_length)
def test_poll_clusters_freed_sleep(self):
"""poll_clusters_freed() waits between checks with a valid Sleep"""
# Allocated on the first check and free on the second, so the real
# win32api.Sleep runs once.
with mock.patch('bleachbit.WindowsWipe.get_volume_bitmap',
return_value=(b'\xff', 8)), \
mock.patch('bleachbit.WindowsWipe.check_extents',
side_effect=[(0, 1), (1, 0)]):
# pylint: disable-next=possibly-used-before-assignment
self.assertTrue(poll_clusters_freed(None, 8, [(0, 0)]))
def test_wipe_extent_by_defrag_compressed_folder(self):
"""The zero-fill file gets clusters to move in a compressed folder"""
folder = self.mkdir('compressed_folder')
from bleachbit.General import run_external
(rc, _, _) = run_external(['compact', '/c', folder])
if rc != 0:
self.skipTest('folder compression is not supported')
volume_info = get_volume_information(volume_from_file(folder))
cluster_size = (volume_info.sectors_per_cluster *
volume_info.bytes_per_sector)
tmp_path = os.path.join(folder, 'zero_fill')
# Clusters 0-15 look free and the moves are only recorded, so
# nothing on the volume is overwritten.
with mock.patch('bleachbit.WindowsWipe.get_volume_bitmap',
return_value=(bytes(2), 16)), \
mock.patch('bleachbit.WindowsWipe.move_file') as mock_move:
# pylint: disable-next=possibly-used-before-assignment
wipe_extent_by_defrag(None, 0, 15, cluster_size, 16, tmp_path)
moved = sum(call.args[4] for call in mock_move.call_args_list)
self.assertEqual(moved, 16)
self.assertNotExists(tmp_path)
def test_wipe_extent_by_defrag_existing_name(self):
"""wipe_extent_by_defrag() does not reuse a file or link at its path"""
base = self.mkdtemp()
existing = self.write_file(os.path.join(base, 'file'), b'keepme')
target = self.write_file(os.path.join(base, 'target'), b'')
link = os.path.join(base, 'link')
self._create_win_file_symlink(target, link)
os.remove(target)
for path in (existing, link):
with self.subTest(path=path), \
mock.patch('bleachbit.WindowsWipe.get_volume_bitmap',
return_value=(bytes(2), 16)), \
mock.patch('bleachbit.WindowsWipe.move_file'):
# pylint: disable-next=possibly-used-before-assignment
with self.assertRaises(pywintypes.error) as cm:
wipe_extent_by_defrag(None, 0, 0, 4096, 16, path)
# ERROR_FILE_EXISTS
self.assertEqual(cm.exception.winerror, 80)
with open(existing, 'rb') as f:
self.assertEqual(f.read(), b'keepme')
# The dangling link was not followed to create its target
self.assertNotExists(target)
def test_wipe_extent_by_defrag_error(self):
"""A failed zero-fill does not leave the temp file behind"""
tmp_path = os.path.join(self.tempdir, 'zero_fill_error')
# pylint: disable-next=possibly-used-before-assignment
disk_full = pywintypes.error(112, 'WriteFile', 'disk full')
with mock.patch('bleachbit.WindowsWipe.get_volume_bitmap',
return_value=(bytes(2), 16)), \
mock.patch('bleachbit.WindowsWipe.write_zero_fill',
side_effect=disk_full):
with self.assertRaises(pywintypes.error):
try:
wipe_extent_by_defrag(None, 0, 0, 4096, 16, tmp_path)
finally:
# As file_wipe() does
# pylint: disable-next=possibly-used-before-assignment
clean_up(None, None, tmp_path)
self.assertNotExists(tmp_path)
def test_file_wipe_basic(self):
"""Basic unit test for file_wipe"""
# pylint: disable-next=possibly-used-before-assignment
if not shell.IsUserAnAdmin():
self.skipTest("Test requires administrator privileges")
cases = [
{
"name": "file_wipe_basic_small",
"data": b'This is test data that should be wiped',
"expect_zero_extents": True,
},
{
"name": "file_wipe_basic_large",
"data": os.urandom(4 * 1024 * 1024),
"expect_zero_extents": False,
},
]
for case in cases:
tmp_path = os.path.join(self.tempdir, case["name"])
write_data = case["data"]
self.write_file(tmp_path, write_data)
file_handle = open_file(tmp_path)
try:
extents = get_extents(file_handle, filename=tmp_path)
finally:
close_file(file_handle)
if case["expect_zero_extents"]:
self.assertEqual(extents, [])
else:
self.assertIsInstance(extents, list)
self.assertGreater(len(extents), 0)
total_clusters = sum((end - start + 1) for start, end in extents)
if total_clusters > 0:
volume = volume_from_file(tmp_path)
volume_info = get_volume_information(volume)
cluster_size = (volume_info.sectors_per_cluster *
volume_info.bytes_per_sector)
self.assertGreaterEqual(
total_clusters * cluster_size, len(write_data))
# pylint: disable-next=possibly-used-before-assignment
file_wipe(tmp_path)
with open(tmp_path, 'rb') as f:
data = f.read()
self.assertEqual(data, b'\x00' * len(write_data))
def test_file_wipe_locked(self):
"""Test file_wipe() with a locked file"""
if not shell.IsUserAnAdmin():
self.skipTest("Test requires administrator privileges")
tmp_path = os.path.join(self.tempdir, 'file_wipe_locked')
self.write_file(tmp_path, b'This is test data that should be locked')
file_handle = None
try:
# pylint: disable-next=possibly-used-before-assignment
file_handle = win32file.CreateFileW(
tmp_path,
win32file.GENERIC_READ, # Only need read access to lock it
0, # No sharing to cause "access denied" error
None, # No security attributes
win32file.OPEN_EXISTING,
win32file.FILE_ATTRIBUTE_NORMAL,
None
)
with self.assertRaises(pywintypes.error) as context:
file_wipe(tmp_path)
# error 32: file in use
# error 5: access denied
self.assertIn(context.exception.winerror, [5, 32])
finally:
# Close the file handle to release the lock
if file_handle:
win32file.CloseHandle(file_handle)
def test_file_wipe_swapped_handles(self):
"""Test file_wipe() with swapped handles and extents"""
if not shell.IsUserAnAdmin():
self.skipTest("Test requires administrator privileges")
volume = volume_from_file(self.tempdir)
volume_info = get_volume_information(volume)
cluster_size = (volume_info.sectors_per_cluster *
volume_info.bytes_per_sector)
file_configs = [
{"path": os.path.join(self.tempdir, 'file_wipe_swap_1'),
"content": b'1',
"clusters": 1,
"handle": None},
{"path": os.path.join(self.tempdir, 'file_wipe_swap_2'),
"content": b'2',
"clusters": 2,
"handle": None}
]
try:
for config in file_configs:
with open(config["path"], 'wb') as f:
f.write(config["content"] *
(cluster_size * config["clusters"]))
with open(config["path"], 'rb') as f:
content = f.read()
config["content_before"] = content
expected_size = cluster_size * config["clusters"]
self.assertEqual(len(content), expected_size,
f"File should be exactly {config['clusters']} cluster(s) ({expected_size} bytes)")
self.assertEqual(
content, config["content"] * expected_size)
config["handle"] = open_file(
config["path"], GENERIC_READ | GENERIC_WRITE)
config["extents"] = get_extents(
config["handle"], filename=config["path"])
total_clusters = sum((end - start + 1)
for start, end in config["extents"])
self.assertEqual(total_clusters, config["clusters"],
f"File should have exactly {config['clusters']} cluster(s)")
config["size"], _ = get_file_basic_info(
config["path"], config["handle"])
# Wipe files with swapped handles and extents
# Wipe file1 using file2's handle and extents
# pylint: disable-next=possibly-used-before-assignment
wipe_file_direct(file_configs[1]["handle"], file_configs[0]["extents"],
cluster_size, file_configs[0]["size"])
# Wipe file2 using file1's handle and extents
wipe_file_direct(file_configs[0]["handle"], file_configs[1]["extents"],
cluster_size, file_configs[1]["size"])
for config in file_configs:
if config["handle"]:
close_file(config["handle"])
config["handle"] = None
for config in file_configs:
with open(config["path"], 'rb') as f:
config["content_after"] = f.read()
# Verify results
# File 1 (originally 1 cluster) wiped with File 2's extents (2 clusters)
self.assertNotEqual(file_configs[0]["content_after"], file_configs[0]["content_before"],
"File 1 still contains original content after wiping")
self.assertTrue(all(byte == 0 for byte in file_configs[0]["content_after"][:cluster_size]),
"First cluster of file 1 was not properly zeroed out")
# File 1 should grow to match file 2's size (2 clusters)
self.assertEqual(len(file_configs[0]["content_after"]), cluster_size * 2,
f"File 1 size should grow to 2 clusters ({cluster_size * 2} bytes)")
# File 2 (originally 2 clusters) wiped with File 1's extents (1 cluster)
self.assertNotEqual(file_configs[1]["content_after"], file_configs[1]["content_before"],
"File 2 still contains all original content after wiping")
# First cluster should be zeroed out
first_cluster = file_configs[1]["content_after"][:cluster_size]
self.assertTrue(all(byte == 0 for byte in first_cluster),
"First cluster of file 2 was not properly zeroed out")
# Second cluster should still have original data
second_cluster = file_configs[1]["content_after"][cluster_size:cluster_size * 2]
self.assertEqual(second_cluster, file_configs[1]["content"] * cluster_size,
"Second cluster of file 2 should still contain original data")
# File 2 size should remain the same (2 clusters)
self.assertEqual(len(file_configs[1]["content_after"]), cluster_size * 2,
f"File 2 size should remain 2 clusters ({cluster_size * 2} bytes)")
finally:
for config in file_configs:
if config.get("handle"):
with suppress(Exception):
close_file(config["handle"])
def test_not_included(self):
"""Notify users there are more tests"""
self.skipTest(
"More wiping tests available at https://github.com/bleachbit/windows-wipe/blob/master/testwipe.py")
def test_extents_a_minus_b(self):
"""Unit test for extents_a_minus_b()"""
def a_minus_b_as_set(a, b):
result = set()
# pylint: disable-next=possibly-used-before-assignment
for start, end in extents_a_minus_b(a, b):
self.assertLessEqual(start, end)
result.update(range(start, end + 1))
return result
def as_set(extents):
result = set()
for start, end in extents:
result.update(range(start, end + 1))
return result
cases = [
# (a, b)
([(0, 100)], [(0, 50)]), # tail of A past the last B range
([(500, 600)], [(0, 100)]), # B entirely before A, no overlap
([(0, 100)], [(150, 200)]), # B entirely after A, no overlap
([(0, 100)], []), # B empty
([], [(0, 100)]), # A empty
([(0, 100)], [(0, 100)]), # fully covered
([(0, 100)], [(40, 60)]), # B in the middle of A
([(0, 100)], [(0, 30), (40, 100)]), # multiple B ranges
([(0, 10), (20, 30), (40, 50)], [
(5, 25), (45, 60)]), # multiple A and B
([(10, 20), (30, 40)], [(50, 60)]), # B after all of A
]
for a, b in cases:
with self.subTest(a=a, b=b):
got = a_minus_b_as_set(a, b)
expected = as_set(a) - as_set(b)
self.assertEqual(got, expected)
if __name__ == '__main__':
unittest.main()