Repository navigation
restart eebus client on config change - #4304
Merged
Merged
Conversation
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 1 security issue, and 9 other issues
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)
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="packages/modules/io_devices/eebus/api.py" line_range="77" />
<code_context>
-
- check=True,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- text=True
- )
</code_context>
<issue_to_address>
**EEBus client stops handling traffic**
When the client writes enough stderr to fill the pipe, `Popen` leaves stderr piped, and `read` calls `communicate()` only after the process exits. The child blocks writing to the full pipe while `poll()` keeps reporting it as running, so the client stops handling EEBus traffic and is not restarted.
Drain stderr while the process runs or redirect it somewhere that cannot block the client.
Also at `packages/modules/io_devices/eebus/api.py:76`, `packages/modules/io_devices/eebus/api.py:78`.
</issue_to_address>
### Comment 2
<location path="packages/modules/io_devices/eebus/api.py" line_range="68" />
<code_context>
+ nonlocal process
+ log.debug(f"Starte EEbus-Client für Steuerbox mit ID {config.id} und "
+ f"SKI {config.configuration.remote_ski}")
+ process = subprocess.Popen(
+ [get_executable_path(),
+ str(config.configuration.port),
</code_context>
<issue_to_address>
**Device stays unavailable on startup**
When the EEBus executable is missing or cannot be started, `get_executable_path()` returns `None` or an unusable path, so `Popen` raises during `ConfigurableIo` initialization and `create_io` never returns a device; the caller leaves the EEBus device unavailable.
Handle process-start failures without letting them escape device initialization, so the device can report the failure through its normal read path.
</issue_to_address>
### Comment 3
<location path="packages/helpermodules/subdata.py" line_range="1109" />
<code_context>
index = get_index(msg.topic)
if decode_payload(msg.payload) == "":
if "io"+index in var:
+ var["io"+index].shutdown()
if (var[f"io{index}"].config.type == "add_on" and
var[f"io{index}"].config.configuration.host == "localhost"):
</code_context>
<issue_to_address>
**Failed replacement leaves device stopped**
When creating a replacement device raises during a configuration change, `process_system_topic` calls `shutdown()` on the old device before `dev.create_io(config)`. If creation raises, assignment never replaces the old entry, leaving the stopped device in `var` and the requested configuration unapplied.
Create the replacement before shutting down and replacing the existing device.
Also at `packages/modules/io_devices/eebus/api.py:68`.
</issue_to_address>
### Comment 4
<location path="packages/modules/io_devices/eebus/api.py" line_range="106" />
<code_context>
- nonlocal thread_exception
-
- if is_thread_alive("eebus_binary") is False:
- run_eebus()
- if thread_exception is not None:
- exception_to_raise = thread_exception
</code_context>
<issue_to_address>
**Repeated failures spawn clients**
When the EEBus client repeatedly exits with a nonzero code, for example while MQTT or the control box is unavailable, `read()` starts a replacement every time it observes an exited process, then reports the prior nonzero exit; the next polling cycle repeats this without a delay or retry limit, repeatedly spawning processes and logging errors while the dependency remains unavailable.
Add bounded retry backoff or otherwise rate-limit restarts after repeated failures.
</issue_to_address>
### Comment 5
<location path="packages/modules/io_devices/eebus/api.py" line_range="98" />
<code_context>
+
+ def shutdown():
+ nonlocal stopped, process
+ with process_lock:
+ stopped = True
+ if process is not None:
</code_context>
<issue_to_address>
**Old reads overwrite new state**
When a configuration change or deletion races with a read that has passed the stopped check, `read` releases `process_lock` before finishing its broker work and returning state, while shutdown does not wait for in-flight reads. The old `ConfigurableIo.read()` can therefore publish stale EEBus data after replacement, overwriting the new instance’s values.
Prevent shutdown or replacement from completing until in-flight reads finish, or prevent reads from the old instance from publishing state afterward.
Also at `packages/helpermodules/subdata.py:1109`.
</issue_to_address>
### Comment 6
<location path="packages/modules/io_devices/eebus/api.py" line_range="116" />
<code_context>
+ msg = f"Fehlercode: {returncode}, Fehler: {stderr}"
+ control_command_log.error(msg)
+ raise ValueError(msg)
broker.start_finite_loop()
log.debug(f"Empfange MQTT Daten für EEBus {config.id}: {received_topics}")
io_state = IoState()
</code_context>
<issue_to_address>
**Old broker can reconnect**
When a configuration change shuts down the client after `read` releases the lock but before it starts the broker loop, `read` releases `process_lock` before calling `broker.start_finite_loop()`, so a concurrent `shutdown` can disconnect the broker first and then have the in-flight read start its loop afterward. The old subscription can reconnect and compete with the replacement client, which uses the same broker client ID.
Synchronize broker loop startup with shutdown so no read can start the old client’s loop after disconnection.
Also at `packages/modules/io_devices/eebus/api.py:95`.
</issue_to_address>
### Comment 7
<location path="packages/modules/io_devices/eebus/api_test.py" line_range="98-99" />
<code_context>
+ io.component_reader()
+
+ assert api.subprocess.Popen.call_count == 1
+ process.wait.assert_not_called()
+ process.communicate.assert_not_called()
+
+
</code_context>
<issue_to_address>
**Running client is never polled**
When the reader skips polling a still-running client, `test_running_client_is_only_polled_without_waiting` checks only that no process is started, waited on, or communicated with; it never asserts that `poll()` is called, so a reader that skips the running-client liveness check still passes this test.
Assert that `process.poll()` is called for each reader invocation.
</issue_to_address>
### Comment 8
<location path="packages/modules/io_devices/eebus/api_test.py" line_range="59" />
<code_context>
+
+ io.shutdown_handler()
+
+ process.terminate.assert_called_once_with()
+ if termination_times_out:
+ process.kill.assert_called_once_with()
</code_context>
<issue_to_address>
**Shutdown timeout is unchecked**
When shutdown calls `communicate()` without a timeout, `test_shutdown_waits_for_client_and_disconnects_broker` never asserts that `communicate` receives a timeout; its mock raises `TimeoutExpired` regardless of call arguments, so `communicate()` without a timeout passes and can block shutdown indefinitely.
Assert that shutdown calls `process.communicate(timeout=5)` before falling back to killing the process.
</issue_to_address>
### Comment 9
<location path="packages/modules/io_devices/eebus/api_test.py" line_range="57" />
<code_context>
+ io = api.create_io(Eebus(id=3))
+ io.initializer()
+
+ io.shutdown_handler()
+
+ process.terminate.assert_called_once_with()
</code_context>
<issue_to_address>
**Public shutdown path is bypassed**
When `ConfigurableIo.shutdown()` fails to invoke its configured shutdown handler, `test_shutdown_waits_for_client_and_disconnects_broker` calls the callback directly, and the config-change test uses a mock device; both pass if `ConfigurableIo.shutdown()` is a no-op and never invokes the callback.
Exercise `ConfigurableIo.shutdown()` on a real instance and assert that it invokes the handler.
Also at `packages/helpermodules/subdata_test.py:24`.
</issue_to_address>
### Comment 10
<location path="packages/modules/io_devices/eebus/api.py" line_range="68-79" />
<code_context>
process = subprocess.Popen(
[get_executable_path(),
str(config.configuration.port),
config.configuration.remote_ski,
f"{cert_path}/hems-cert-{config.id}.pem",
f"{cert_path}/hems-key-{config.id}.pem",
str(config.id),
f"{Path(__file__).resolve().parents[4]}/ramdisk/eebus_hems_client.log"],
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True
)
</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>| "io"+index not in var or | ||
| io_config != asdict(var["io"+index].config)): | ||
| if "io"+index in var: | ||
| var["io"+index].shutdown() |
There was a problem hiding this comment.
🟡 Medium · Failed replacement leaves device stopped
When creating a replacement device raises during a configuration change, process_system_topic calls shutdown() on the old device before dev.create_io(config). If creation raises, assignment never replaces the old entry, leaving the stopped device in var and the requested configuration unapplied.
Create the replacement before shutting down and replacing the existing device.
Also at packages/modules/io_devices/eebus/api.py:68.
Prompt for AI agents
In `packages/helpermodules/subdata.py` at line 1109:
**Failed replacement leaves device stopped**
When creating a replacement device raises during a configuration change, `process_system_topic` calls `shutdown()` on the old device before `dev.create_io(config)`. If creation raises, assignment never replaces the old entry, leaving the stopped device in `var` and the requested configuration unapplied.
Create the replacement before shutting down and replacing the existing device.
Also at `packages/modules/io_devices/eebus/api.py:68`.| stderr = "" | ||
| if process is not None: | ||
| _, stderr = process.communicate() | ||
| run_eebus() |
There was a problem hiding this comment.
🟡 Medium · Repeated failures spawn clients
When the EEBus client repeatedly exits with a nonzero code, for example while MQTT or the control box is unavailable, read() starts a replacement every time it observes an exited process, then reports the prior nonzero exit; the next polling cycle repeats this without a delay or retry limit, repeatedly spawning processes and logging errors while the dependency remains unavailable.
Add bounded retry backoff or otherwise rate-limit restarts after repeated failures.
Prompt for AI agents
In `packages/modules/io_devices/eebus/api.py` at line 106:
**Repeated failures spawn clients**
When the EEBus client repeatedly exits with a nonzero code, for example while MQTT or the control box is unavailable, `read()` starts a replacement every time it observes an exited process, then reports the prior nonzero exit; the next polling cycle repeats this without a delay or retry limit, repeatedly spawning processes and logging errors while the dependency remains unavailable.
Add bounded retry backoff or otherwise rate-limit restarts after repeated failures.| exception_to_raise = thread_exception | ||
| thread_exception = None | ||
| raise exception_to_raise | ||
| with process_lock: |
There was a problem hiding this comment.
🟡 Medium · Old reads overwrite new state
When a configuration change or deletion races with a read that has passed the stopped check, read releases process_lock before finishing its broker work and returning state, while shutdown does not wait for in-flight reads. The old ConfigurableIo.read() can therefore publish stale EEBus data after replacement, overwriting the new instance’s values.
Prevent shutdown or replacement from completing until in-flight reads finish, or prevent reads from the old instance from publishing state afterward.
Also at packages/helpermodules/subdata.py:1109.
Prompt for AI agents
In `packages/modules/io_devices/eebus/api.py` at line 98:
**Old reads overwrite new state**
When a configuration change or deletion races with a read that has passed the stopped check, `read` releases `process_lock` before finishing its broker work and returning state, while shutdown does not wait for in-flight reads. The old `ConfigurableIo.read()` can therefore publish stale EEBus data after replacement, overwriting the new instance’s values.
Prevent shutdown or replacement from completing until in-flight reads finish, or prevent reads from the old instance from publishing state afterward.
Also at `packages/helpermodules/subdata.py:1109`.| msg = f"Fehlercode: {returncode}, Fehler: {stderr}" | ||
| control_command_log.error(msg) | ||
| raise ValueError(msg) | ||
| broker.start_finite_loop() |
There was a problem hiding this comment.
🟡 Medium · Old broker can reconnect
When a configuration change shuts down the client after read releases the lock but before it starts the broker loop, read releases process_lock before calling broker.start_finite_loop(), so a concurrent shutdown can disconnect the broker first and then have the in-flight read start its loop afterward. The old subscription can reconnect and compete with the replacement client, which uses the same broker client ID.
Synchronize broker loop startup with shutdown so no read can start the old client’s loop after disconnection.
Also at packages/modules/io_devices/eebus/api.py:95.
Prompt for AI agents
In `packages/modules/io_devices/eebus/api.py` at line 116:
**Old broker can reconnect**
When a configuration change shuts down the client after `read` releases the lock but before it starts the broker loop, `read` releases `process_lock` before calling `broker.start_finite_loop()`, so a concurrent `shutdown` can disconnect the broker first and then have the in-flight read start its loop afterward. The old subscription can reconnect and compete with the replacement client, which uses the same broker client ID.
Synchronize broker loop startup with shutdown so no read can start the old client’s loop after disconnection.
Also at `packages/modules/io_devices/eebus/api.py:95`.
Contributor
There was a problem hiding this comment.
🟡 Changes recommended
Critical lifecycle races can let in-flight reads publish stale state after replacement, and an undrained stderr pipe can block the long-lived EEBus client and prevent recovery.
4 open issues
What changed in this PR
This PR adds lifecycle management for EEBus I/O clients so configuration changes and device deletion stop existing clients before creating replacements. It also prevents certificate generation from mutating the active configuration.
Changes:
- Replaces the EEBus worker thread with a restartable subprocess, including termination, timeout, failure reporting, and broker disconnection.
- Adds shutdown support to
ConfigurableIoand invokes it when I/O configurations are replaced or deleted. - Deep-copies EEBus configuration during certificate generation and adds coverage for lifecycle and certificate behavior.
| File | Description |
|---|---|
| packages/modules/io_devices/eebus/api.py | Implements EEBus subprocess lifecycle management and safe certificate configuration publishing. |
| packages/modules/io_devices/eebus/api_test.py | Tests certificate generation, shutdown, restart, and failure handling. |
| packages/modules/common/configurable_io.py | Adds a configurable shutdown callback to I/O devices. |
| packages/helpermodules/subdata.py | Stops existing I/O devices before replacement or deletion. |
| packages/helpermodules/subdata_test.py | Verifies shutdown ordering during I/O configuration changes and deletion. |
Review effort: Lite
| "io"+index not in var or | ||
| io_config != asdict(var["io"+index].config)): | ||
| if "io"+index in var: | ||
| var["io"+index].shutdown() |
Comment on lines
+39
to
+40
| def shutdown(self) -> None: | ||
| self.__shutdown_handler() |
Comment on lines
+77
to
+78
| stderr=subprocess.PIPE, | ||
| text=True |
Comment on lines
+98
to
+102
| with process_lock: | ||
| if stopped: | ||
| raise RuntimeError("EEBus-Client wurde beendet.") | ||
| returncode = process.poll() if process is not None else None | ||
| if process is None or returncode is not None: |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by Sourcery
Ensure EEBUS clients are cleanly stopped and reliably restarted when configuration or process state changes.
New Features:
Bug Fixes:
Enhancements:
Tests: