diff --git a/core/deploy.py b/core/deploy.py index 92ec254..9276f68 100644 --- a/core/deploy.py +++ b/core/deploy.py @@ -7,6 +7,7 @@ Business logic for project deployments: - Dependency chain traversal via project_links """ +import shlex import sqlite3 import subprocess import time @@ -50,7 +51,7 @@ def _build_ssh_cmd(project: dict, command: str) -> list[str]: proxy_jump = project.get("ssh_proxy_jump") deploy_path = project.get("deploy_path") - full_cmd = f"cd {deploy_path} && {command}" if deploy_path else command + full_cmd = f"cd {shlex.quote(deploy_path)} && {command}" if deploy_path else command cmd = ["ssh"] if ssh_key: diff --git a/tests/test_deploy.py b/tests/test_deploy.py index 115680f..c29807b 100644 --- a/tests/test_deploy.py +++ b/tests/test_deploy.py @@ -7,12 +7,16 @@ Covers: - core/db.py: deploy columns migration """ +import shlex import pytest from unittest.mock import patch, MagicMock from core.db import init_db, _migrate from core import models -from core.deploy import build_deploy_steps, execute_deploy, get_deploy_chain, deploy_with_dependents +from core.deploy import ( + build_deploy_steps, execute_deploy, get_deploy_chain, deploy_with_dependents, + _build_ssh_cmd, +) # --------------------------------------------------------------------------- @@ -452,3 +456,138 @@ class TestDeployColumnsMigration: tables = {row["name"] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()} assert "project_links" in tables conn.close() + + +# --------------------------------------------------------------------------- +# 8. _build_ssh_cmd — security: shlex.quote on deploy_path (fix #426) +# --------------------------------------------------------------------------- + +class TestSSHBuildCmd: + def test_deploy_path_is_shlex_quoted(self): + """deploy_path must be quoted via shlex.quote to prevent command injection.""" + project = { + "deploy_host": "10.0.0.1", + "deploy_path": "/srv/my app/v2", + } + cmd = _build_ssh_cmd(project, "git pull") + full_cmd_arg = cmd[-1] + assert shlex.quote("/srv/my app/v2") in full_cmd_arg + + def test_deploy_path_with_spaces_no_raw_unquoted(self): + """Path with spaces must NOT appear unquoted after 'cd '.""" + project = { + "deploy_host": "myserver", + "deploy_path": "/srv/app with spaces", + } + cmd = _build_ssh_cmd(project, "git pull") + full_cmd_arg = cmd[-1] + assert "cd /srv/app with spaces " not in full_cmd_arg + + def test_deploy_path_with_dollar_sign_is_quoted(self): + """Path containing $ must be quoted to prevent shell variable expansion.""" + project = { + "deploy_host": "myserver", + "deploy_path": "/srv/$PROJECT", + } + cmd = _build_ssh_cmd(project, "git pull") + full_cmd_arg = cmd[-1] + assert "cd /srv/$PROJECT " not in full_cmd_arg + assert shlex.quote("/srv/$PROJECT") in full_cmd_arg + + def test_normal_path_still_works(self): + """Standard path without special chars should still produce valid cd command.""" + project = { + "deploy_host": "10.0.0.5", + "deploy_path": "/srv/api", + } + cmd = _build_ssh_cmd(project, "git pull") + full_cmd_arg = cmd[-1] + assert "/srv/api" in full_cmd_arg + assert "git pull" in full_cmd_arg + + def test_no_deploy_path_uses_command_directly(self): + """When deploy_path is None, command is used as-is without cd.""" + project = { + "deploy_host": "myserver", + "deploy_path": None, + } + cmd = _build_ssh_cmd(project, "git pull") + full_cmd_arg = cmd[-1] + assert "cd" not in full_cmd_arg + assert full_cmd_arg == "git pull" + + +# --------------------------------------------------------------------------- +# 9. deploy_with_dependents — cascade deploy unit tests +# --------------------------------------------------------------------------- + +class TestDeployWithDependents: + def _setup_chain(self, conn): + """Create api + frontend where frontend depends_on api.""" + models.create_project(conn, "api", "API", "/srv/api") + models.create_project(conn, "fe", "Frontend", "/srv/fe") + models.update_project(conn, "api", deploy_runtime="docker", deploy_path="/srv/api") + models.update_project(conn, "fe", deploy_runtime="static", deploy_path="/srv/fe") + models.create_project_link(conn, "fe", "api", "depends_on") + + def test_deploys_main_and_dependent_on_success(self, conn): + """If main project deploys successfully, dependent is also deployed.""" + self._setup_chain(conn) + with patch("core.deploy.subprocess.run", return_value=_make_proc()): + result = deploy_with_dependents(conn, "api") + assert result["success"] is True + assert "fe" in result["dependents_deployed"] + + def test_main_failure_skips_dependents(self, conn): + """If main project deployment fails, dependents are NOT deployed.""" + self._setup_chain(conn) + with patch("core.deploy.subprocess.run", return_value=_make_proc(returncode=1)): + result = deploy_with_dependents(conn, "api") + assert result["success"] is False + assert result["dependents_deployed"] == [] + + def test_unknown_project_returns_error(self, conn): + """Deploying non-existent project returns success=False with error key.""" + result = deploy_with_dependents(conn, "nonexistent") + assert result["success"] is False + assert "error" in result + + def test_no_dependents_returns_empty_list(self, conn): + """Project with no dependents returns empty dependents_deployed list.""" + models.create_project(conn, "solo", "Solo", "/srv/solo") + models.update_project(conn, "solo", deploy_runtime="python", deploy_path="/srv/solo") + with patch("core.deploy.subprocess.run", return_value=_make_proc()): + result = deploy_with_dependents(conn, "solo") + assert result["success"] is True + assert result["dependents_deployed"] == [] + + def test_result_contains_main_steps_and_results(self, conn): + """Result always includes 'steps' and 'results' keys from main project.""" + models.create_project(conn, "api2", "API2", "/srv/api2") + models.update_project(conn, "api2", deploy_runtime="node", deploy_path="/srv/api2") + with patch("core.deploy.subprocess.run", return_value=_make_proc()): + result = deploy_with_dependents(conn, "api2") + assert "steps" in result + assert "results" in result + + +# --------------------------------------------------------------------------- +# 10. build_deploy_steps — python runtime full steps with restart_cmd +# --------------------------------------------------------------------------- + +class TestBuildDeployStepsPythonRestartCmd: + def test_python_with_restart_cmd_full_steps(self): + """Python runtime + restart_cmd yields full 3-step list.""" + p = {"deploy_runtime": "python", "deploy_restart_cmd": "systemctl restart myapp"} + steps = build_deploy_steps(p) + assert steps == ["git pull", "pip install -r requirements.txt", "systemctl restart myapp"] + + def test_node_with_custom_restart_cmd_appends_as_fourth_step(self): + """Node runtime default ends with pm2 restart all; custom cmd is appended after.""" + p = {"deploy_runtime": "node", "deploy_restart_cmd": "pm2 restart myservice"} + steps = build_deploy_steps(p) + assert steps[0] == "git pull" + assert steps[1] == "npm install --production" + assert steps[2] == "pm2 restart all" + assert steps[3] == "pm2 restart myservice" + assert len(steps) == 4