Conversation
Refactor services to use centralized system helpers (ServiceControl, PackageManager, run_privileged, is_command_available) instead of ad-hoc subprocess/os calls. Updated deployment, firewall, ftp, nginx, php, python, security, and ssl services to call the new utilities and added error/timeout handling improvements and package manager abstraction. Added backend/app/utils/system.py implementing the helpers and backend/tests/test_utils_system.py with unit tests. This centralizes privileged command execution and service/package management, simplifying tests and platform differences.
There was a problem hiding this comment.
Pull request overview
Refactors backend service modules to use a centralized app.utils.system helper layer for privileged command execution, package management, and systemd service control, with accompanying unit tests.
Changes:
- Added
backend/app/utils/system.pyprovidingrun_privileged,is_command_available,PackageManager, andServiceControl. - Updated multiple service modules to replace ad-hoc
subprocess/os.path.existslogic with the centralized helpers. - Added
backend/tests/test_utils_system.pyunit tests for the new utilities and bumpedVERSION.
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| backend/app/utils/system.py | Introduces centralized helpers for privileged subprocess execution, package manager detection/install, and systemd service control. |
| backend/tests/test_utils_system.py | Adds unit tests for the new system utilities (direct-import approach to avoid Flask deps). |
| backend/app/services/ssl_service.py | Migrates certbot/systemctl/openssl calls to run_privileged and ServiceControl. |
| backend/app/services/security_service.py | Uses PackageManager/ServiceControl/run_privileged for ClamAV/Fail2ban/audit operations. |
| backend/app/services/python_service.py | Switches systemd operations for gunicorn services to ServiceControl. |
| backend/app/services/php_service.py | Replaces privileged command execution and systemctl calls with the new helpers. |
| backend/app/services/nginx_service.py | Replaces privileged file/systemctl operations with run_privileged/ServiceControl. |
| backend/app/services/ftp_service.py | Uses PackageManager and ServiceControl for install/status/service control paths. |
| backend/app/services/firewall_service.py | Refactors firewall detection/control/rules to use PackageManager, ServiceControl, run_privileged, and is_command_available. |
| backend/app/services/deployment_service.py | Uses ServiceControl.restart(..., check=True) for restart steps during deploy/rollback. |
| VERSION | Bumps version from 1.2.80 to 1.2.81. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| if isinstance(cmd, str): | ||
| # Shell-mode — caller is responsible for quoting | ||
| needs_sudo = os.geteuid() != 0 and not cmd.lstrip().startswith('sudo ') | ||
| if needs_sudo: | ||
| cmd = f'sudo {cmd}' | ||
| else: | ||
| cmd = list(cmd) | ||
| needs_sudo = os.geteuid() != 0 and cmd[0] != 'sudo' | ||
| if needs_sudo: | ||
| cmd = ['sudo'] + cmd | ||
|
|
||
| kwargs.setdefault('capture_output', True) | ||
| kwargs.setdefault('text', True) | ||
| return subprocess.run(cmd, **kwargs) |
There was a problem hiding this comment.
run_privileged() treats a string cmd as “shell-mode”, but it calls subprocess.run(cmd, shell=False); with a string argument this will try to execute a literal filename containing spaces (e.g. sudo systemctl restart nginx) and will fail at runtime. Either (a) drop support for str entirely and require List[str], or (b) require callers to pass shell=True explicitly (and document the security implications), or (c) split safely (e.g. shlex.split) and keep shell=False.
| # Stub parent packages so patch() resolution never hits Flask imports. | ||
| if 'app' not in sys.modules: | ||
| sys.modules['app'] = types.ModuleType('app') | ||
| if 'app.utils' not in sys.modules: | ||
| _utils = types.ModuleType('app.utils') | ||
| sys.modules['app.utils'] = _utils | ||
| sys.modules['app'].utils = _utils # type: ignore[attr-defined] | ||
|
|
||
| sys.modules['app.utils.system'] = _module | ||
| sys.modules['app.utils'].system = _module # type: ignore[attr-defined] | ||
| _spec.loader.exec_module(_module) |
There was a problem hiding this comment.
This test module permanently stubs sys.modules['app'] / sys.modules['app.utils']. If any other tests (now or in the future) import the real app package, they’ll get the stub instead, causing confusing cross-test failures. Consider using a pytest fixture that snapshots/restores the affected sys.modules entries during the test run, or load the module under a non-app.* name and patch that name instead.
| mock_run.return_value = subprocess.CompletedProcess([], 0) | ||
| run_privileged('systemctl restart nginx') | ||
| args, _ = mock_run.call_args | ||
| assert args[0] == 'sudo systemctl restart nginx' | ||
|
|
||
| @patch('app.utils.system.subprocess.run') | ||
| @patch('app.utils.system.os.geteuid', return_value=1000, create=True) | ||
| def test_string_command_no_double_sudo(self, _euid, mock_run): | ||
| mock_run.return_value = subprocess.CompletedProcess([], 0) | ||
| run_privileged('sudo systemctl restart nginx') | ||
| args, _ = mock_run.call_args | ||
| assert args[0] == 'sudo systemctl restart nginx' |
There was a problem hiding this comment.
The string-command tests encode behavior that won’t work in practice: subprocess.run() with a string argument does not split into argv unless shell=True is set. These assertions will pass with mocking but the real call will fail. Update the tests to match the intended API (e.g., require List[str] commands, or assert shell=True when a string command is accepted).
| mock_run.return_value = subprocess.CompletedProcess([], 0) | |
| run_privileged('systemctl restart nginx') | |
| args, _ = mock_run.call_args | |
| assert args[0] == 'sudo systemctl restart nginx' | |
| @patch('app.utils.system.subprocess.run') | |
| @patch('app.utils.system.os.geteuid', return_value=1000, create=True) | |
| def test_string_command_no_double_sudo(self, _euid, mock_run): | |
| mock_run.return_value = subprocess.CompletedProcess([], 0) | |
| run_privileged('sudo systemctl restart nginx') | |
| args, _ = mock_run.call_args | |
| assert args[0] == 'sudo systemctl restart nginx' | |
| """String commands are not supported; callers must pass List[str].""" | |
| with pytest.raises(TypeError): | |
| run_privileged('systemctl restart nginx') | |
| @patch('app.utils.system.subprocess.run') | |
| @patch('app.utils.system.os.geteuid', return_value=1000, create=True) | |
| def test_string_command_no_double_sudo(self, _euid, mock_run): | |
| """Even if starting with 'sudo', string commands must be rejected.""" | |
| with pytest.raises(TypeError): | |
| run_privileged('sudo systemctl restart nginx') |
|
|
||
| from app.utils.system import ServiceControl, run_privileged | ||
|
|
There was a problem hiding this comment.
After switching to ServiceControl/run_privileged, this file no longer appears to use the subprocess module (no subprocess. references remain). Consider removing the now-unused import subprocess to avoid lint warnings and reduce noise.
| import subprocess | ||
| import re | ||
|
|
||
| from app.utils.system import PackageManager, ServiceControl, run_privileged |
There was a problem hiding this comment.
run_privileged is imported here but not used anywhere in this module (only PackageManager / ServiceControl are referenced). Removing the unused import will avoid lint issues and keep the refactor minimal.
| from app.utils.system import PackageManager, ServiceControl, run_privileged | |
| from app.utils.system import PackageManager, ServiceControl |
| ufw_result = run_privileged(['ufw', 'status'], timeout=5) | ||
| if ufw_result.returncode == 0 and 'active' in ufw_result.stdout.lower(): | ||
| checks['passed_checks'] += 1 | ||
| checks['findings'].append({'severity': 'pass', 'message': 'UFW firewall is active'}) | ||
| else: | ||
| firewalld_result = subprocess.run(['firewall-cmd', '--state'], capture_output=True, text=True, timeout=5) | ||
| firewalld_result = run_privileged(['firewall-cmd', '--state'], timeout=5) | ||
| if firewalld_result.returncode == 0 and 'running' in firewalld_result.stdout.lower(): |
There was a problem hiding this comment.
run_privileged() will prepend sudo when not root, which can break read-only status checks like ufw status / firewall-cmd --state in environments without passwordless sudo (e.g., returning “a terminal is required” and causing a false negative). For audit/status checks, consider running without sudo (plain subprocess.run) or extend the helper API with an explicit sudo=False/require_root=False option.
| if installed: | ||
| # Check if running | ||
| result = subprocess.run( | ||
| ['firewall-cmd', '--state'], | ||
| capture_output=True, text=True | ||
| ) | ||
| result = run_privileged(['firewall-cmd', '--state']) | ||
| running = 'running' in result.stdout.lower() | ||
|
|
||
| if running: | ||
| # Get default zone | ||
| result = subprocess.run( | ||
| ['firewall-cmd', '--get-default-zone'], | ||
| capture_output=True, text=True | ||
| ) | ||
| result = run_privileged(['firewall-cmd', '--get-default-zone']) | ||
| default_zone = result.stdout.strip() |
There was a problem hiding this comment.
These firewalld/ufw status checks now use run_privileged(), which may unnecessarily force sudo and fail in non-interactive contexts that don’t have passwordless sudo (leading to incorrect installed/running/active results). For read-only queries like firewall-cmd --state / --get-default-zone / ufw status, consider running without sudo or adding a helper flag to avoid sudo for status commands.
| if needs_sudo: | ||
| cmd = f'sudo {cmd}' | ||
| else: | ||
| cmd = list(cmd) |
There was a problem hiding this comment.
run_privileged() assumes a non-empty list command (cmd[0]), so run_privileged([]) will raise IndexError. It would be safer to validate cmd is non-empty and raise a clear ValueError/RuntimeError (or return early) to avoid hard-to-debug crashes if a caller accidentally passes an empty list.
| cmd = list(cmd) | |
| cmd = list(cmd) | |
| if not cmd: | |
| raise ValueError("run_privileged() expected a non-empty command list") |
| @@ -247,10 +193,7 @@ def _get_firewalld_rules(cls) -> Dict: | |||
| }) | |||
|
|
|||
| # Get ports | |||
| result = subprocess.run( | |||
| ['firewall-cmd', '--list-ports'], | |||
| capture_output=True, text=True | |||
| ) | |||
| result = run_privileged(['firewall-cmd', '--list-ports']) | |||
| ports = result.stdout.strip().split() if result.stdout.strip() else [] | |||
| for port in ports: | |||
| port_num, protocol = port.split('/') if '/' in port else (port, 'tcp') | |||
| @@ -263,10 +206,7 @@ def _get_firewalld_rules(cls) -> Dict: | |||
| }) | |||
|
|
|||
| # Get rich rules (includes IP blocks) | |||
| result = subprocess.run( | |||
| ['firewall-cmd', '--list-rich-rules'], | |||
| capture_output=True, text=True | |||
| ) | |||
| result = run_privileged(['firewall-cmd', '--list-rich-rules']) | |||
| rich_rules = result.stdout.strip().split('\n') if result.stdout.strip() else [] | |||
There was a problem hiding this comment.
Similarly, run_privileged() is used for listing firewalld rules (--list-services/ports/rich-rules). If sudo is required, these calls can fail in non-interactive environments without passwordless sudo and make the UI appear empty. Consider using non-sudo calls for read-only listing, or ensure the helper supports opting out of sudo for these commands.
Refactor services to use centralized system helpers (ServiceControl, PackageManager, run_privileged, is_command_available) instead of ad-hoc subprocess/os calls. Updated deployment, firewall, ftp, nginx, php, python, security, and ssl services to call the new utilities and added error/timeout handling improvements and package manager abstraction. Added backend/app/utils/system.py implementing the helpers and backend/tests/test_utils_system.py with unit tests. This centralizes privileged command execution and service/package management, simplifying tests and platform differences.