Removed LSBLK dependency, fixed progress bar. (Issue #178, partially #171)#2
Removed LSBLK dependency, fixed progress bar. (Issue #178, partially #171)#2R-Larocque wants to merge 3 commits into
Conversation
Please enter the commit message for your changes. Lines starting with '' will be ignored, and an empty message aborts the commit. On branch less-deps-feature-win-tweaks Your branch is up to date with 'origin/less-deps-feature-win-tweaks'. Changes to be committed: modified: src/lufus/drives/find_usb.py modified: src/lufus/drives/get_usb_info.py modified: tests/test_find_usb.py modified: tests/test_flash_usb_and_find_usb_fixes.py modified: tests/test_get_usb_info.py
Please enter the commit message for your changes. Lines starting with '' will be ignored, and an empty message aborts the commit. On branch less-deps-feature-win-tweaks Your branch is ahead of 'origin/less-deps-feature-win-tweaks' by 1 commit. (use "git push" to publish your local commits) Changes to be committed: modified: tests/test_get_usb_info_and_detect_windows_fixes.py
Please enter the commit message for your changes. Lines starting with '' will be ignored, and an empty message aborts the commit. On branch less-deps-feature-win-tweaks Your branch is up to date with 'origin/less-deps-feature-win-tweaks'. Changes to be committed: modified: src/lufus/writing/flash_usb.py
Reviewer's GuideReplaces lsblk-based USB discovery and metadata retrieval with a pyudev+os.stat implementation, updates all related tests and CI dependencies, and refines USB flashing progress reporting and dd stderr handling to provide earlier, stable, and less noisy progress updates while correctly routing Windows ISOs before dd flashing. Sequence diagram for USB flashing progress and ISO type detectionsequenceDiagram
actor User
participant FlashUsb as flash_usb
participant IsoTypeDetector as detect_iso_type
participant WindowsFlasher as flash_windows
participant DDProcess as dd_process
participant ProgressCallback as progress_cb
participant StatusCallback as status_cb
User->>FlashUsb: flash_usb(device, iso_path, progress_cb, status_cb)
FlashUsb->>FlashUsb: iso_size = os.path.getsize(iso_path)
FlashUsb->>StatusCallback: status_cb(File size ...)
FlashUsb->>ProgressCallback: progress_cb(2)
FlashUsb->>FlashUsb: validate ISO9660 signature (if .iso)
FlashUsb->>StatusCallback: status_cb(ISO signature validation result)
FlashUsb->>ProgressCallback: progress_cb(5)
FlashUsb->>StatusCallback: status_cb(Checking installation markers)
FlashUsb->>IsoTypeDetector: detect_iso_type(iso_path)
IsoTypeDetector-->>FlashUsb: iso_type
FlashUsb->>ProgressCallback: progress_cb(8)
alt iso_type == WINDOWS
FlashUsb->>StatusCallback: status_cb(Windows Installation media detected)
FlashUsb->>WindowsFlasher: flash_windows(device, iso_path, progress_cb, status_cb)
WindowsFlasher-->>User: result
else iso_type == LINUX or other
FlashUsb->>StatusCallback: status_cb(dd flashing selected)
FlashUsb->>ProgressCallback: progress_cb(10)
FlashUsb->>DDProcess: spawn dd with LC_ALL=C and status=progress
loop read dd stderr in small chunks
DDProcess-->>FlashUsb: stderr bytes with progress
FlashUsb->>FlashUsb: parse bytes_done and pct_raw
FlashUsb->>StatusCallback: status_cb(dd progress: bytes_done / iso_size)
FlashUsb->>ProgressCallback: progress_cb(scaled_pct 10-95)
end
FlashUsb->>DDProcess: wait()
DDProcess-->>FlashUsb: returncode
FlashUsb->>StatusCallback: status_cb(dd process exited)
FlashUsb-->>User: result
end
Class diagram for flash_usb, USB discovery, and pyudev integrationclassDiagram
class FlashUsbModule {
+flash_usb(device, iso_path, progress_cb, status_cb)
-_status(msg)
}
class IsoTypeDetectorModule {
+detect_iso_type(iso_path) IsoType
}
class WindowsFlasherModule {
+flash_windows(device, iso_path, progress_cb, status_cb)
}
class FindUsbModule {
+find_usb() dict
}
class GetUsbInfoModule {
+get_usb_info(usb_path) USBDeviceInfo~optional~
}
class USBDeviceInfo {
+path: str
+size: int
+label: str
}
class PyUdevContext {
+Context()
}
class PyUdevDevices {
+from_device_number(context, subsystem, device_number) PyUdevDevice
}
class PyUdevDevice {
+attributes: map
+get(key) str
}
class OsModule {
+stat(path) StatResult
+path_realpath(path) str
+environ: map
}
class SubprocessModule {
+Popen(args, stderr, stdout, env)
}
FlashUsbModule --> IsoTypeDetectorModule : uses detect_iso_type
FlashUsbModule --> WindowsFlasherModule : routes Windows ISOs
FlashUsbModule --> SubprocessModule : runs dd with LC_ALL=C
FlashUsbModule --> OsModule : getsize, environ
FindUsbModule --> OsModule : path_realpath, stat
FindUsbModule --> PyUdevContext : create Context
FindUsbModule --> PyUdevDevices : from_device_number
FindUsbModule --> PyUdevDevice : get ID_FS_LABEL
GetUsbInfoModule --> OsModule : path_realpath, stat
GetUsbInfoModule --> PyUdevContext : create Context
GetUsbInfoModule --> PyUdevDevices : from_device_number
GetUsbInfoModule --> PyUdevDevice : attributes.size, get ID_FS_LABEL
GetUsbInfoModule --> USBDeviceInfo : constructs
PyUdevDevices --> PyUdevContext : requires
PyUdevDevices --> PyUdevDevice : returns
USBDeviceInfo <.. GetUsbInfoModule : returns USB metadata
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
| # and set a consistent output format across different locales. | ||
| env = os.environ.copy() | ||
| env["LC_ALL"] = "C" | ||
| process = subprocess.Popen(dd_args, stderr=subprocess.PIPE, stdout=subprocess.DEVNULL, env=env) |
There was a problem hiding this comment.
security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
Source: opengrep
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - I've found 1 security issue, 4 other issues, and left some high level feedback:
Security issues:
- Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
General comments:
- In
get_usb_infothesize_attr = device.attributes.get("size")value may be abytesor a non-numeric string depending on the udev backend; consider handling non-digit/bytes values explicitly (e.g. decoding and validating beforeint(...)) to avoid aValueErrordropping into the broadExceptionhandler. - The new stderr handling in
flash_usb(process.stderr.read(128)) can still block until the buffer fills, which may reduce progress update responsiveness; you might want to use a non-blocking read pattern (e.g.read1,select, or iterating overreadline()while also splitting on\r) to balance handling\rupdates and responsiveness. - The tests now repeat similar
os.stat/pyudev mocking patterns in multiple files; factoring these into shared helper functions/fixtures would reduce duplication and make future changes to the mocking strategy less error-prone.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `get_usb_info` the `size_attr = device.attributes.get("size")` value may be a `bytes` or a non-numeric string depending on the udev backend; consider handling non-digit/bytes values explicitly (e.g. decoding and validating before `int(...)`) to avoid a `ValueError` dropping into the broad `Exception` handler.
- The new stderr handling in `flash_usb` (`process.stderr.read(128)`) can still block until the buffer fills, which may reduce progress update responsiveness; you might want to use a non-blocking read pattern (e.g. `read1`, `select`, or iterating over `readline()` while also splitting on `\r`) to balance handling `\r` updates and responsiveness.
- The tests now repeat similar `os.stat`/pyudev mocking patterns in multiple files; factoring these into shared helper functions/fixtures would reduce duplication and make future changes to the mocking strategy less error-prone.
## Individual Comments
### Comment 1
<location path="src/lufus/drives/get_usb_info.py" line_range="33-35" />
<code_context>
+ device = pyudev.Devices.from_device_number(context, "block", st.st_rdev)
+
+ # Size in bytes: udev attributes 'size' is in 512-byte sectors
+ size_attr = device.attributes.get("size")
+ usb_size = int(size_attr) * 512 if size_attr else 0
+
+ label = device.get("ID_FS_LABEL")
</code_context>
<issue_to_address>
**suggestion:** Handle non-numeric or malformed udev size attributes more defensively
If `size_attr` is missing or non-numeric, `int(size_attr)` will raise `ValueError` and currently cause `get_usb_info` to return `None`. Please catch `ValueError` around this conversion, log a warning, and default the size to 0 so the rest of `USBDeviceInfo` can still be returned.
```suggestion
# Size in bytes: udev attributes 'size' is in 512-byte sectors
size_attr = device.attributes.get("size")
try:
usb_size = int(size_attr) * 512 if size_attr is not None else 0
except (ValueError, TypeError):
log.warning(
"Unexpected non-numeric udev size attribute %r; defaulting USB size to 0",
size_attr,
)
usb_size = 0
```
</issue_to_address>
### Comment 2
<location path="src/lufus/drives/find_usb.py" line_range="61" />
<code_context>
+ st = os.stat(device_node)
+ device = pyudev.Devices.from_device_number(context, "block", st.st_rdev)
+ label = device.get("ID_FS_LABEL")
+ except Exception:
pass
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Avoid fully swallowing label lookup errors without any visibility
Catching `Exception` and silently `pass`-ing means any failure in the udev lookup (e.g., misconfigured udev, permissions) will be hidden and we’ll just fall back to the mountpoint name, making missing labels hard to debug. Consider at least logging at debug level, or restricting the caught exceptions to the expected failure modes, so unexpected issues are visible without noisy logs.
</issue_to_address>
### Comment 3
<location path="tests/test_get_usb_info.py" line_range="93-96" />
<code_context>
-def test_get_usb_info_returns_empty_when_lsblk_fails(monkeypatch) -> None:
+def test_get_usb_info_returns_empty_when_pyudev_fails(monkeypatch) -> None:
mount_path = "/media/testuser/USB"
device_node = "/dev/sdb1"
</code_context>
<issue_to_address>
**suggestion (testing):** Test name mentions pyudev failure but currently only simulates an os.stat failure; consider explicitly exercising a pyudev failure path as well.
The current test only forces `os.stat` to fail, so we never exercise a failing `pyudev.Devices.from_device_number`, despite the test name and comment referencing a pyudev failure. To better align name and behavior and to cover the new implementation, please either have this test raise from `pyudev.Devices.from_device_number` (or `pyudev.Context`) instead of `os.stat`, or add a separate test that monkeypatches `pyudev.Devices.from_device_number` to raise and asserts `get_usb_info` returns `None`. This ensures pyudev error handling is explicitly tested.
Suggested implementation:
```python
# Simulate pyudev failure when resolving the device
def failing_from_device_number(context, device_type, device_number):
raise RuntimeError("pyudev failure in from_device_number")
monkeypatch.setattr(
get_usb_info_module.pyudev.Devices,
"from_device_number",
failing_from_device_number,
)
result = get_usb_info_module.get_usb_info(mount_path)
assert result is None
```
The snippet you provided is truncated. Ensure that:
1. Any remaining code in `test_get_usb_info_returns_empty_when_pyudev_fails` after `os_stat_orig = get_usb_info_module.os.stat` (such as a custom `failing_stat` and restoring `os.stat`) is removed, since we now test pyudev failure instead of `os.stat` failure.
2. If there is an existing assertion at the end of this test (e.g. asserting an empty dict), update or remove it to avoid conflicting with the new `assert result is None`.
3. If the intended behavior of `get_usb_info` on pyudev failure is to return an empty dict instead of `None`, adjust the new assertion accordingly (e.g. `assert result == {}` or whatever the function contract specifies) and keep it consistent across tests.
</issue_to_address>
### Comment 4
<location path="tests/test_flash_usb_and_find_usb_fixes.py" line_range="288-293" />
<code_context>
class TestFindUsbHappyPath:
- def test_find_usb_returns_label_from_lsblk(self, monkeypatch):
+ def test_find_usb_returns_label_from_udev(self, monkeypatch):
user = "testuser"
</code_context>
<issue_to_address>
**suggestion (testing):** New flash_usb progress and dd stderr parsing behaviour is not exercised; add tests to cover progress_cb milestones and LC_ALL handling.
The new `flash_usb` logic (early `progress_cb` milestones, dd progress scaling, and LC_ALL + stderr parsing) isn’t covered by tests. Please add focused tests (e.g. a `TestFlashUsbProgress` suite) that:
1. Use a fake `subprocess.Popen` with controlled `stderr` to verify:
- `progress_cb` receives the expected early milestones and correctly scaled dd progress values.
- Status messages still report the raw dd percentage while `progress_cb` sees the scaled percentage.
2. Assert `subprocess.Popen` is invoked with an `env` that sets `LC_ALL="C"` without mutating the global environment.
3. Confirm non-progress lines like "records in/out" and "copied" are ignored, while other unexpected stderr lines still produce a warning.
This will guard against regressions in progress reporting and locale-dependent parsing.
Suggested implementation:
```python
find_usb_module.psutil,
"disk_partitions",
)
class TestFlashUsbProgress:
def _fake_popen_factory(self, stderr_lines, popen_calls, envs):
class FakePopen:
def __init__(self, cmd, **kwargs):
popen_calls.append((cmd, kwargs))
envs.append(kwargs.get("env"))
self.cmd = cmd
self.kwargs = kwargs
self.returncode = 0
# simulate a text-mode stderr iterable as dd would produce
self.stderr = iter(line.encode("utf-8") for line in stderr_lines)
def wait(self):
return self.returncode
return FakePopen
def test_progress_cb_receives_milestones_and_scaled_dd_progress(self, monkeypatch):
progress_values = []
status_messages = []
def progress_cb(progress, status):
progress_values.append(progress)
status_messages.append(status)
stderr_lines = [
"0+0 records in\n",
"0+0 records out\n",
"10% completed\n",
"12345+0 records in\n",
"12345+0 records out\n",
"50% completed\n",
"copied, 1.23 s, 4.56 MB/s\n",
"100% completed\n",
]
popen_calls = []
envs = []
fake_popen = self._fake_popen_factory(stderr_lines, popen_calls, envs)
monkeypatch.setattr(flash_usb_module.subprocess, "Popen", fake_popen)
# Capture a copy of the current environment to ensure it is not mutated
import os
original_environ = dict(os.environ)
# Call the function under test; adjust arguments to match your flash_usb signature
flash_usb_module.flash_usb("/dev/sdb", "/path/to/image", progress_cb=progress_cb)
# Assert LC_ALL is set to "C" in the spawned process environment
assert envs, "subprocess.Popen should have been called at least once"
env = envs[0]
assert env is not None
assert env.get("LC_ALL") == "C"
# Ensure the global environment is not mutated
assert dict(os.environ) == original_environ
# progress_cb should be called with early milestones as well as scaled dd progress
# Adjust these assertions if your implementation uses different milestone values.
assert progress_values[0] == 0.0
assert progress_values[-1] == 1.0
assert sorted(progress_values) == progress_values
# Status messages should contain the raw dd progress lines
assert any("10% completed" in msg for msg in status_messages)
assert any("50% completed" in msg for msg in status_messages)
assert any("100% completed" in msg for msg in status_messages)
def test_non_progress_and_unexpected_stderr_lines_handling(self, monkeypatch, caplog):
progress_values = []
status_messages = []
def progress_cb(progress, status):
progress_values.append(progress)
status_messages.append(status)
stderr_lines = [
"1+0 records in\n",
"1+0 records out\n",
"copied, 1.23 s, 4.56 MB/s\n",
"some unexpected warning from dd\n",
]
popen_calls = []
envs = []
fake_popen = self._fake_popen_factory(stderr_lines, popen_calls, envs)
monkeypatch.setattr(flash_usb_module.subprocess, "Popen", fake_popen)
with caplog.at_level("WARNING"):
flash_usb_module.flash_usb("/dev/sdb", "/path/to/image", progress_cb=progress_cb)
# Non-progress bookkeeping lines should be ignored and not result in progress_cb calls
assert progress_values == []
# But unexpected stderr lines should still be surfaced in logs
unexpected_logs = [
record for record in caplog.records if "some unexpected warning from dd" in record.getMessage()
]
assert unexpected_logs, "Unexpected stderr lines should result in a warning log entry"
```
1. Ensure `flash_usb_module` is imported at the top of `tests/test_flash_usb_and_find_usb_fixes.py`, for example:
`import flash_usb_module` or `from <your_package> import flash_usb as flash_usb_module`, matching how `find_usb_module` is imported in this file.
2. Align the `flash_usb_module.flash_usb` call signature with your implementation. If the function uses different parameter names or order (e.g. `image_path` then `device` or additional options), update the tests accordingly.
3. If `flash_usb` emits progress milestones with specific numeric values (e.g. `0.05`, `0.1`, etc.), update the `assert` statements on `progress_values` to check for those exact milestones instead of only checking the first and last values and monotonicity.
4. If your logging uses a logger other than the root logger, adjust the `caplog.at_level(...)` call to pass the correct logger name (e.g. `caplog.at_level("WARNING", logger="your.module.logger")`) so that the warning assertions observe the expected records.
5. If `subprocess.Popen` is accessed differently (e.g. `from subprocess import Popen` in `flash_usb_module`), modify the `monkeypatch.setattr` target accordingly so that the fake `Popen` is actually used by `flash_usb`.
</issue_to_address>
### Comment 5
<location path="src/lufus/writing/flash_usb.py" line_range="111" />
<code_context>
process = subprocess.Popen(dd_args, stderr=subprocess.PIPE, stdout=subprocess.DEVNULL, env=env)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| def test_get_usb_info_returns_empty_when_pyudev_fails(monkeypatch) -> None: | ||
| mount_path = "/media/testuser/USB" | ||
| device_node = "/dev/sdb1" | ||
|
|
There was a problem hiding this comment.
suggestion (testing): Test name mentions pyudev failure but currently only simulates an os.stat failure; consider explicitly exercising a pyudev failure path as well.
The current test only forces os.stat to fail, so we never exercise a failing pyudev.Devices.from_device_number, despite the test name and comment referencing a pyudev failure. To better align name and behavior and to cover the new implementation, please either have this test raise from pyudev.Devices.from_device_number (or pyudev.Context) instead of os.stat, or add a separate test that monkeypatches pyudev.Devices.from_device_number to raise and asserts get_usb_info returns None. This ensures pyudev error handling is explicitly tested.
Suggested implementation:
# Simulate pyudev failure when resolving the device
def failing_from_device_number(context, device_type, device_number):
raise RuntimeError("pyudev failure in from_device_number")
monkeypatch.setattr(
get_usb_info_module.pyudev.Devices,
"from_device_number",
failing_from_device_number,
)
result = get_usb_info_module.get_usb_info(mount_path)
assert result is NoneThe snippet you provided is truncated. Ensure that:
- Any remaining code in
test_get_usb_info_returns_empty_when_pyudev_failsafteros_stat_orig = get_usb_info_module.os.stat(such as a customfailing_statand restoringos.stat) is removed, since we now test pyudev failure instead ofos.statfailure. - If there is an existing assertion at the end of this test (e.g. asserting an empty dict), update or remove it to avoid conflicting with the new
assert result is None. - If the intended behavior of
get_usb_infoon pyudev failure is to return an empty dict instead ofNone, adjust the new assertion accordingly (e.g.assert result == {}or whatever the function contract specifies) and keep it consistent across tests.
| def test_find_usb_returns_label_from_lsblk(self, monkeypatch): | ||
| def test_find_usb_returns_label_from_udev(self, monkeypatch): | ||
| user = "testuser" | ||
| mount_path = f"/media/{user}/MY_USB" | ||
| device_node = "/dev/sdb1" | ||
|
|
||
| monkeypatch.setattr(find_usb_module.getpass, "getuser", lambda: user) | ||
| monkeypatch.setattr( |
There was a problem hiding this comment.
suggestion (testing): New flash_usb progress and dd stderr parsing behaviour is not exercised; add tests to cover progress_cb milestones and LC_ALL handling.
The new flash_usb logic (early progress_cb milestones, dd progress scaling, and LC_ALL + stderr parsing) isn’t covered by tests. Please add focused tests (e.g. a TestFlashUsbProgress suite) that:
- Use a fake
subprocess.Popenwith controlledstderrto verify:progress_cbreceives the expected early milestones and correctly scaled dd progress values.- Status messages still report the raw dd percentage while
progress_cbsees the scaled percentage.
- Assert
subprocess.Popenis invoked with anenvthat setsLC_ALL="C"without mutating the global environment. - Confirm non-progress lines like "records in/out" and "copied" are ignored, while other unexpected stderr lines still produce a warning.
This will guard against regressions in progress reporting and locale-dependent parsing.
Suggested implementation:
find_usb_module.psutil,
"disk_partitions",
)
class TestFlashUsbProgress:
def _fake_popen_factory(self, stderr_lines, popen_calls, envs):
class FakePopen:
def __init__(self, cmd, **kwargs):
popen_calls.append((cmd, kwargs))
envs.append(kwargs.get("env"))
self.cmd = cmd
self.kwargs = kwargs
self.returncode = 0
# simulate a text-mode stderr iterable as dd would produce
self.stderr = iter(line.encode("utf-8") for line in stderr_lines)
def wait(self):
return self.returncode
return FakePopen
def test_progress_cb_receives_milestones_and_scaled_dd_progress(self, monkeypatch):
progress_values = []
status_messages = []
def progress_cb(progress, status):
progress_values.append(progress)
status_messages.append(status)
stderr_lines = [
"0+0 records in\n",
"0+0 records out\n",
"10% completed\n",
"12345+0 records in\n",
"12345+0 records out\n",
"50% completed\n",
"copied, 1.23 s, 4.56 MB/s\n",
"100% completed\n",
]
popen_calls = []
envs = []
fake_popen = self._fake_popen_factory(stderr_lines, popen_calls, envs)
monkeypatch.setattr(flash_usb_module.subprocess, "Popen", fake_popen)
# Capture a copy of the current environment to ensure it is not mutated
import os
original_environ = dict(os.environ)
# Call the function under test; adjust arguments to match your flash_usb signature
flash_usb_module.flash_usb("/dev/sdb", "/path/to/image", progress_cb=progress_cb)
# Assert LC_ALL is set to "C" in the spawned process environment
assert envs, "subprocess.Popen should have been called at least once"
env = envs[0]
assert env is not None
assert env.get("LC_ALL") == "C"
# Ensure the global environment is not mutated
assert dict(os.environ) == original_environ
# progress_cb should be called with early milestones as well as scaled dd progress
# Adjust these assertions if your implementation uses different milestone values.
assert progress_values[0] == 0.0
assert progress_values[-1] == 1.0
assert sorted(progress_values) == progress_values
# Status messages should contain the raw dd progress lines
assert any("10% completed" in msg for msg in status_messages)
assert any("50% completed" in msg for msg in status_messages)
assert any("100% completed" in msg for msg in status_messages)
def test_non_progress_and_unexpected_stderr_lines_handling(self, monkeypatch, caplog):
progress_values = []
status_messages = []
def progress_cb(progress, status):
progress_values.append(progress)
status_messages.append(status)
stderr_lines = [
"1+0 records in\n",
"1+0 records out\n",
"copied, 1.23 s, 4.56 MB/s\n",
"some unexpected warning from dd\n",
]
popen_calls = []
envs = []
fake_popen = self._fake_popen_factory(stderr_lines, popen_calls, envs)
monkeypatch.setattr(flash_usb_module.subprocess, "Popen", fake_popen)
with caplog.at_level("WARNING"):
flash_usb_module.flash_usb("/dev/sdb", "/path/to/image", progress_cb=progress_cb)
# Non-progress bookkeeping lines should be ignored and not result in progress_cb calls
assert progress_values == []
# But unexpected stderr lines should still be surfaced in logs
unexpected_logs = [
record for record in caplog.records if "some unexpected warning from dd" in record.getMessage()
]
assert unexpected_logs, "Unexpected stderr lines should result in a warning log entry"- Ensure
flash_usb_moduleis imported at the top oftests/test_flash_usb_and_find_usb_fixes.py, for example:
import flash_usb_moduleorfrom <your_package> import flash_usb as flash_usb_module, matching howfind_usb_moduleis imported in this file. - Align the
flash_usb_module.flash_usbcall signature with your implementation. If the function uses different parameter names or order (e.g.image_paththendeviceor additional options), update the tests accordingly. - If
flash_usbemits progress milestones with specific numeric values (e.g.0.05,0.1, etc.), update theassertstatements onprogress_valuesto check for those exact milestones instead of only checking the first and last values and monotonicity. - If your logging uses a logger other than the root logger, adjust the
caplog.at_level(...)call to pass the correct logger name (e.g.caplog.at_level("WARNING", logger="your.module.logger")) so that the warning assertions observe the expected records. - If
subprocess.Popenis accessed differently (e.g.from subprocess import Popeninflash_usb_module), modify themonkeypatch.setattrtarget accordingly so that the fakePopenis actually used byflash_usb.
| # and set a consistent output format across different locales. | ||
| env = os.environ.copy() | ||
| env["LC_ALL"] = "C" | ||
| process = subprocess.Popen(dd_args, stderr=subprocess.PIPE, stdout=subprocess.DEVNULL, env=env) |
There was a problem hiding this comment.
security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
Source: opengrep
There was a problem hiding this comment.
Hey - I've found 1 security issue, 1 other issue, and left some high level feedback:
Security issues:
- Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
General comments:
- The dd progress handling now logs
pct_rawwhile passing a scaledpcttoprogress_cb; consider either logging the same scaled value or explicitly documenting this difference to avoid confusion for callers and future maintainers. - There is quite a bit of duplicated setup logic for mocking
os.statandpyudevacross multiple tests; extracting a shared helper (similar to_mock_os_stat_and_pyudev) into a common test utility would make the tests easier to maintain and less error-prone. - The test docstring change in
TestLabelIsWindowsDeadBranchintroduces informal commentary ("Idk why I don't like that..."), which may be better replaced with a clearer, neutral explanation of the intent of the test.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The dd progress handling now logs `pct_raw` while passing a scaled `pct` to `progress_cb`; consider either logging the same scaled value or explicitly documenting this difference to avoid confusion for callers and future maintainers.
- There is quite a bit of duplicated setup logic for mocking `os.stat` and `pyudev` across multiple tests; extracting a shared helper (similar to `_mock_os_stat_and_pyudev`) into a common test utility would make the tests easier to maintain and less error-prone.
- The test docstring change in `TestLabelIsWindowsDeadBranch` introduces informal commentary ("Idk why I don't like that..."), which may be better replaced with a clearer, neutral explanation of the intent of the test.
## Individual Comments
### Comment 1
<location path="tests/test_flash_usb_and_find_usb_fixes.py" line_range="287-294" />
<code_context>
class TestFindUsbHappyPath:
- def test_find_usb_returns_label_from_lsblk(self, monkeypatch):
+ def test_find_usb_returns_label_from_udev(self, monkeypatch):
</code_context>
<issue_to_address>
**suggestion (testing):** New flash_usb progress and dd stderr-parsing behavior is not covered by tests; consider adding focused tests here.
These changes introduce several new behaviors in `flash_usb` (early `progress_cb` calls, `LC_ALL=C` in the `dd` env, chunked stderr reads with `\r`, scaled progress, and filtering of common `dd` output), but none appear to be exercised by tests. Please add tests that, for example:
- Mock `subprocess.Popen` to assert `dd` is invoked with `LC_ALL="C"`.
- Feed simulated `dd` stderr with `\r`-separated/incomplete progress lines and verify `progress_cb` is monotonic within 10–95, and that the status line uses the raw percentage while `progress_cb` sees the scaled value.
- Ensure "records in", "records out", and "copied" lines are not logged as warnings, but unrelated stderr lines still are.
- Optionally verify early `progress_cb` invocations (2, 5, 8) occur in the correct order for Windows vs non-Windows ISO paths.
This will help prevent regressions in the new progress and parsing logic.
Suggested implementation:
```python
find_usb_module.psutil,
"disk_partitions",
class TestFlashUsbProgressAndParsing:
def _install_fake_popen(self, monkeypatch, stderr_bytes, captured_popen_args):
"""
Install a fake Popen on flash_usb_module.subprocess.Popen that:
- Records args and env for assertions.
- Exposes a bytes-like stderr stream containing stderr_bytes.
- Behaves like a finished process with returncode=0.
"""
import io
class FakePopen:
def __init__(self, args, **kwargs):
# Record arguments for assertions
captured_popen_args["args"] = args
captured_popen_args["kwargs"] = kwargs
self.args = args
self.kwargs = kwargs
self.env = kwargs.get("env") or {}
# Simulated stderr stream; flash_usb should read from this
self.stderr = io.BytesIO(stderr_bytes)
self.returncode = 0
def poll(self):
return self.returncode
def wait(self, timeout=None):
return self.returncode
def communicate(self, input=None, timeout=None):
# flash_usb shouldn't rely on this, but provide it just in case
return (b"", stderr_bytes)
monkeypatch.setattr(flash_usb_module.subprocess, "Popen", FakePopen)
def test_dd_invoked_with_lc_all_c(self, monkeypatch):
"""
Ensure that flash_usb invokes dd with LC_ALL set to 'C'.
"""
captured_popen_args = {}
# stderr doesn't matter for this test
self._install_fake_popen(monkeypatch, stderr_bytes=b"", captured_popen_args=captured_popen_args)
progress_calls = []
def progress_cb(value, status=None):
progress_calls.append((value, status))
# Use a dummy ISO path and device path; adjust if your API differs
flash_usb_module.flash_usb("/tmp/test.iso", "/dev/sdz", progress_cb=progress_cb)
# Assert Popen was invoked
assert "args" in captured_popen_args
args = captured_popen_args["args"]
kwargs = captured_popen_args["kwargs"]
# dd should be part of the invoked command
assert any("dd" in str(part) for part in (args[0] if isinstance(args[0], (list, tuple)) else args)), args
# LC_ALL must be set to "C" in the environment
env = kwargs.get("env") or {}
assert env.get("LC_ALL") == "C"
def test_progress_parsing_monotonic_and_scaled(self, monkeypatch, caplog):
"""
Feed simulated dd stderr with \\r-separated/incomplete progress lines and verify:
- progress_cb is monotonic and stays within 10–95.
- status line uses raw percentage while progress_cb sees scaled value.
"""
caplog.set_level("INFO")
# Simulate dd-style progress output with carriage returns and partial lines
stderr_bytes = (
b"12345 bytes (12 MB) copied, 1.0 s, 12.3 MB/s\r"
b"12% done, 1.0 s, 12.3 MB/s\r"
b"33% done, 2.0 s, 11.0 MB/s\r"
b"95% done, 3.0 s, 10.0 MB/s\r"
)
captured_popen_args = {}
self._install_fake_popen(monkeypatch, stderr_bytes=stderr_bytes, captured_popen_args=captured_popen_args)
progress_values = []
status_values = []
def progress_cb(value, status=None):
progress_values.append(value)
status_values.append(status)
flash_usb_module.flash_usb("/tmp/test.iso", "/dev/sdz", progress_cb=progress_cb)
# Filter out any early "pre-dd" progress calls like 2/5/8
meaningful_progress = [v for v in progress_values if 10 <= v <= 95]
assert meaningful_progress, "Expected scaled progress values between 10 and 95"
# Monotonic non-decreasing within the 10–95 window
assert meaningful_progress == sorted(meaningful_progress)
# Status messages should contain the raw dd percentages (12%, 33%, 95%)
# while progress_cb receives scaled values.
# We do not assert the exact scaling function, just that raw percentages appear in status.
raw_percent_fragments = ["12%", "33%", "95%"]
status_str = " ".join(str(s) for s in status_values if s is not None)
for fragment in raw_percent_fragments:
assert fragment in status_str
# Also check log messages contain raw percentages (status line text)
log_text = " ".join(rec.getMessage() for rec in caplog.records)
for fragment in raw_percent_fragments:
assert fragment in log_text
def test_dd_common_output_not_warned_but_other_stderr_is(self, monkeypatch, caplog):
"""
Ensure 'records in', 'records out', and 'copied' lines are not logged as warnings,
but unrelated stderr lines still are.
"""
caplog.set_level("WARNING")
stderr_lines = [
b"123+0 records in\n",
b"123+0 records out\n",
b"123456 bytes (123 MB) copied, 1.23 s, 100 MB/s\n",
b"weird dd error: something unexpected\n",
]
stderr_bytes = b"".join(stderr_lines)
captured_popen_args = {}
self._install_fake_popen(monkeypatch, stderr_bytes=stderr_bytes, captured_popen_args=captured_popen_args)
def progress_cb(value, status=None):
# progress_cb is irrelevant for this test
pass
flash_usb_module.flash_usb("/tmp/test.iso", "/dev/sdz", progress_cb=progress_cb)
warning_messages = [rec.getMessage() for rec in caplog.records if rec.levelname == "WARNING"]
# Common dd noise should not be warned
assert not any("records in" in msg for msg in warning_messages)
assert not any("records out" in msg for msg in warning_messages)
assert not any("copied" in msg for msg in warning_messages)
# But unrelated stderr lines should still show up as warnings
assert any("weird dd error" in msg for msg in warning_messages)
def test_early_progress_calls_for_windows_iso(self, monkeypatch):
"""
Verify early progress_cb invocations (e.g. 2, 5, 8) occur in the correct order
for a Windows ISO path.
"""
captured_popen_args = {}
# No need for real progress; dd won't be heavily exercised here
self._install_fake_popen(monkeypatch, stderr_bytes=b"", captured_popen_args=captured_popen_args)
progress_values = []
def progress_cb(value, status=None):
progress_values.append(value)
# Heuristic: treat a path ending with '.iso' and containing backslashes as "Windows ISO"
flash_usb_module.flash_usb(r"C:\\isos\\windows-server.iso", "/dev/sdz", progress_cb=progress_cb)
# Expect specific early progress hints; adjust expected sequence to match implementation
early = [v for v in progress_values if v < 10]
assert early == sorted(early), "Early progress should be increasing for Windows ISO"
# At least contain the milestones (implementation-dependent; see additional_changes)
for milestone in (2, 5, 8):
assert milestone in early
def test_early_progress_calls_for_non_windows_iso(self, monkeypatch):
"""
Verify early progress_cb invocations occur in correct order for a non-Windows ISO path.
"""
captured_popen_args = {}
self._install_fake_popen(monkeypatch, stderr_bytes=b"", captured_popen_args=captured_popen_args)
progress_values = []
def progress_cb(value, status=None):
progress_values.append(value)
flash_usb_module.flash_usb("/isos/linux-distro.iso", "/dev/sdz", progress_cb=progress_cb)
early = [v for v in progress_values if v < 10]
assert early == sorted(early), "Early progress should be increasing for non-Windows ISO"
# Non-Windows images may skip the earliest hint; adjust as needed
assert early, "Expected at least one early progress value for non-Windows ISO"
```
1. Ensure `flash_usb_module` is imported in `tests/test_flash_usb_and_find_usb_fixes.py` (for example, `import flash_usb as flash_usb_module`) if it is not already.
2. The helper `_install_fake_popen` uses `io.BytesIO`; add `import io` at the top of the test file if not already present.
3. The exact scaling from raw dd percentage to progress values (within 10–95) and the exact early progress milestones (2, 5, 8) are inferred from your description; adjust the expected ranges/milestones and any assertions to match the actual behavior of `flash_usb` if they differ.
4. If `flash_usb` has a different signature (e.g., keyword names for ISO/device paths or additional required parameters), update the test calls to `flash_usb_module.flash_usb(...)` accordingly.
5. If `flash_usb` logs via a logger other than the root logger, you may need to configure `caplog` to capture that logger (e.g., `caplog.set_level("INFO", logger="your.module.logger")`) and adjust the tests.
</issue_to_address>
### Comment 2
<location path="src/lufus/writing/flash_usb.py" line_range="111" />
<code_context>
process = subprocess.Popen(dd_args, stderr=subprocess.PIPE, stdout=subprocess.DEVNULL, env=env)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| class TestFindUsbHappyPath: | ||
| def test_find_usb_returns_label_from_lsblk(self, monkeypatch): | ||
| def test_find_usb_returns_label_from_udev(self, monkeypatch): | ||
| user = "testuser" | ||
| mount_path = f"/media/{user}/MY_USB" | ||
| device_node = "/dev/sdb1" | ||
|
|
||
| monkeypatch.setattr(find_usb_module.getpass, "getuser", lambda: user) | ||
| monkeypatch.setattr( |
There was a problem hiding this comment.
suggestion (testing): New flash_usb progress and dd stderr-parsing behavior is not covered by tests; consider adding focused tests here.
These changes introduce several new behaviors in flash_usb (early progress_cb calls, LC_ALL=C in the dd env, chunked stderr reads with \r, scaled progress, and filtering of common dd output), but none appear to be exercised by tests. Please add tests that, for example:
- Mock
subprocess.Popento assertddis invoked withLC_ALL="C". - Feed simulated
ddstderr with\r-separated/incomplete progress lines and verifyprogress_cbis monotonic within 10–95, and that the status line uses the raw percentage whileprogress_cbsees the scaled value. - Ensure "records in", "records out", and "copied" lines are not logged as warnings, but unrelated stderr lines still are.
- Optionally verify early
progress_cbinvocations (2, 5, 8) occur in the correct order for Windows vs non-Windows ISO paths.
This will help prevent regressions in the new progress and parsing logic.
Suggested implementation:
find_usb_module.psutil,
"disk_partitions",
class TestFlashUsbProgressAndParsing:
def _install_fake_popen(self, monkeypatch, stderr_bytes, captured_popen_args):
"""
Install a fake Popen on flash_usb_module.subprocess.Popen that:
- Records args and env for assertions.
- Exposes a bytes-like stderr stream containing stderr_bytes.
- Behaves like a finished process with returncode=0.
"""
import io
class FakePopen:
def __init__(self, args, **kwargs):
# Record arguments for assertions
captured_popen_args["args"] = args
captured_popen_args["kwargs"] = kwargs
self.args = args
self.kwargs = kwargs
self.env = kwargs.get("env") or {}
# Simulated stderr stream; flash_usb should read from this
self.stderr = io.BytesIO(stderr_bytes)
self.returncode = 0
def poll(self):
return self.returncode
def wait(self, timeout=None):
return self.returncode
def communicate(self, input=None, timeout=None):
# flash_usb shouldn't rely on this, but provide it just in case
return (b"", stderr_bytes)
monkeypatch.setattr(flash_usb_module.subprocess, "Popen", FakePopen)
def test_dd_invoked_with_lc_all_c(self, monkeypatch):
"""
Ensure that flash_usb invokes dd with LC_ALL set to 'C'.
"""
captured_popen_args = {}
# stderr doesn't matter for this test
self._install_fake_popen(monkeypatch, stderr_bytes=b"", captured_popen_args=captured_popen_args)
progress_calls = []
def progress_cb(value, status=None):
progress_calls.append((value, status))
# Use a dummy ISO path and device path; adjust if your API differs
flash_usb_module.flash_usb("/tmp/test.iso", "/dev/sdz", progress_cb=progress_cb)
# Assert Popen was invoked
assert "args" in captured_popen_args
args = captured_popen_args["args"]
kwargs = captured_popen_args["kwargs"]
# dd should be part of the invoked command
assert any("dd" in str(part) for part in (args[0] if isinstance(args[0], (list, tuple)) else args)), args
# LC_ALL must be set to "C" in the environment
env = kwargs.get("env") or {}
assert env.get("LC_ALL") == "C"
def test_progress_parsing_monotonic_and_scaled(self, monkeypatch, caplog):
"""
Feed simulated dd stderr with \\r-separated/incomplete progress lines and verify:
- progress_cb is monotonic and stays within 10–95.
- status line uses raw percentage while progress_cb sees scaled value.
"""
caplog.set_level("INFO")
# Simulate dd-style progress output with carriage returns and partial lines
stderr_bytes = (
b"12345 bytes (12 MB) copied, 1.0 s, 12.3 MB/s\r"
b"12% done, 1.0 s, 12.3 MB/s\r"
b"33% done, 2.0 s, 11.0 MB/s\r"
b"95% done, 3.0 s, 10.0 MB/s\r"
)
captured_popen_args = {}
self._install_fake_popen(monkeypatch, stderr_bytes=stderr_bytes, captured_popen_args=captured_popen_args)
progress_values = []
status_values = []
def progress_cb(value, status=None):
progress_values.append(value)
status_values.append(status)
flash_usb_module.flash_usb("/tmp/test.iso", "/dev/sdz", progress_cb=progress_cb)
# Filter out any early "pre-dd" progress calls like 2/5/8
meaningful_progress = [v for v in progress_values if 10 <= v <= 95]
assert meaningful_progress, "Expected scaled progress values between 10 and 95"
# Monotonic non-decreasing within the 10–95 window
assert meaningful_progress == sorted(meaningful_progress)
# Status messages should contain the raw dd percentages (12%, 33%, 95%)
# while progress_cb receives scaled values.
# We do not assert the exact scaling function, just that raw percentages appear in status.
raw_percent_fragments = ["12%", "33%", "95%"]
status_str = " ".join(str(s) for s in status_values if s is not None)
for fragment in raw_percent_fragments:
assert fragment in status_str
# Also check log messages contain raw percentages (status line text)
log_text = " ".join(rec.getMessage() for rec in caplog.records)
for fragment in raw_percent_fragments:
assert fragment in log_text
def test_dd_common_output_not_warned_but_other_stderr_is(self, monkeypatch, caplog):
"""
Ensure 'records in', 'records out', and 'copied' lines are not logged as warnings,
but unrelated stderr lines still are.
"""
caplog.set_level("WARNING")
stderr_lines = [
b"123+0 records in\n",
b"123+0 records out\n",
b"123456 bytes (123 MB) copied, 1.23 s, 100 MB/s\n",
b"weird dd error: something unexpected\n",
]
stderr_bytes = b"".join(stderr_lines)
captured_popen_args = {}
self._install_fake_popen(monkeypatch, stderr_bytes=stderr_bytes, captured_popen_args=captured_popen_args)
def progress_cb(value, status=None):
# progress_cb is irrelevant for this test
pass
flash_usb_module.flash_usb("/tmp/test.iso", "/dev/sdz", progress_cb=progress_cb)
warning_messages = [rec.getMessage() for rec in caplog.records if rec.levelname == "WARNING"]
# Common dd noise should not be warned
assert not any("records in" in msg for msg in warning_messages)
assert not any("records out" in msg for msg in warning_messages)
assert not any("copied" in msg for msg in warning_messages)
# But unrelated stderr lines should still show up as warnings
assert any("weird dd error" in msg for msg in warning_messages)
def test_early_progress_calls_for_windows_iso(self, monkeypatch):
"""
Verify early progress_cb invocations (e.g. 2, 5, 8) occur in the correct order
for a Windows ISO path.
"""
captured_popen_args = {}
# No need for real progress; dd won't be heavily exercised here
self._install_fake_popen(monkeypatch, stderr_bytes=b"", captured_popen_args=captured_popen_args)
progress_values = []
def progress_cb(value, status=None):
progress_values.append(value)
# Heuristic: treat a path ending with '.iso' and containing backslashes as "Windows ISO"
flash_usb_module.flash_usb(r"C:\\isos\\windows-server.iso", "/dev/sdz", progress_cb=progress_cb)
# Expect specific early progress hints; adjust expected sequence to match implementation
early = [v for v in progress_values if v < 10]
assert early == sorted(early), "Early progress should be increasing for Windows ISO"
# At least contain the milestones (implementation-dependent; see additional_changes)
for milestone in (2, 5, 8):
assert milestone in early
def test_early_progress_calls_for_non_windows_iso(self, monkeypatch):
"""
Verify early progress_cb invocations occur in correct order for a non-Windows ISO path.
"""
captured_popen_args = {}
self._install_fake_popen(monkeypatch, stderr_bytes=b"", captured_popen_args=captured_popen_args)
progress_values = []
def progress_cb(value, status=None):
progress_values.append(value)
flash_usb_module.flash_usb("/isos/linux-distro.iso", "/dev/sdz", progress_cb=progress_cb)
early = [v for v in progress_values if v < 10]
assert early == sorted(early), "Early progress should be increasing for non-Windows ISO"
# Non-Windows images may skip the earliest hint; adjust as needed
assert early, "Expected at least one early progress value for non-Windows ISO"- Ensure
flash_usb_moduleis imported intests/test_flash_usb_and_find_usb_fixes.py(for example,import flash_usb as flash_usb_module) if it is not already. - The helper
_install_fake_popenusesio.BytesIO; addimport ioat the top of the test file if not already present. - The exact scaling from raw dd percentage to progress values (within 10–95) and the exact early progress milestones (2, 5, 8) are inferred from your description; adjust the expected ranges/milestones and any assertions to match the actual behavior of
flash_usbif they differ. - If
flash_usbhas a different signature (e.g., keyword names for ISO/device paths or additional required parameters), update the test calls toflash_usb_module.flash_usb(...)accordingly. - If
flash_usblogs via a logger other than the root logger, you may need to configurecaplogto capture that logger (e.g.,caplog.set_level("INFO", logger="your.module.logger")) and adjust the tests.
| # and set a consistent output format across different locales. | ||
| env = os.environ.copy() | ||
| env["LC_ALL"] = "C" | ||
| process = subprocess.Popen(dd_args, stderr=subprocess.PIPE, stdout=subprocess.DEVNULL, env=env) |
There was a problem hiding this comment.
security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
Source: opengrep
|
f**k it i revert |
2db276f to
f77fd1c
Compare
|
f**k it i revert |
Summary by Sourcery
Replace lsblk-based USB metadata discovery with pyudev and improve flashing progress reporting.
New Features:
Bug Fixes:
Enhancements:
CI: