diff --git a/docsgpt/deploy/native.py b/docsgpt/deploy/native.py index 876e912a..ab78d0ed 100644 --- a/docsgpt/deploy/native.py +++ b/docsgpt/deploy/native.py @@ -13,6 +13,7 @@ from __future__ import annotations import os import plistlib +import re import shlex import subprocess import sys @@ -47,6 +48,7 @@ def label_for(name: str) -> str: def launchd_plist(unit: Unit, label: Optional[str] = None) -> str: """The launchd agent for ``unit``: kept alive, with its output in the stack's log file.""" + _check_unit_values(unit) body = { "Label": label or label_for(unit.name), "ProgramArguments": list(unit.arguments), @@ -61,6 +63,26 @@ def launchd_plist(unit: Unit, label: Optional[str] = None) -> str: return plistlib.dumps(body).decode("utf-8") +_CONTROL_CHARACTERS = re.compile(r"[\x00-\x1f\x7f]") + + +def _reject_control_characters(what: str, value: str) -> None: + """Service files are line-based, so a newline in a value adds a directive instead of text.""" + if _CONTROL_CHARACTERS.search(value): + raise DeployError(f"{what} contains a control character, which a service file cannot carry: {value!r}") + + +def _check_unit_values(unit: Unit) -> None: + """Everything bound for a service file, checked before any of it is rendered.""" + _reject_control_characters("the working directory", unit.working_directory) + _reject_control_characters("the log file path", unit.log_file) + for key, value in unit.environment.items(): + _reject_control_characters("an environment name", key) + _reject_control_characters(f"the environment value for {key}", value) + for argument in unit.arguments: + _reject_control_characters("a command argument", argument) + + def _systemd_quote(value: str) -> str: """A unit-file value, double-quoted with backslashes and quotes escaped, as systemd reads them.""" escaped = value.replace("\\", "\\\\").replace('"', '\\"') @@ -69,6 +91,7 @@ def _systemd_quote(value: str) -> str: def systemd_unit(unit: Unit) -> str: """The systemd user unit for ``unit``; values are quoted, so a path with spaces survives.""" + _check_unit_values(unit) environment = "\n".join( f"Environment={_systemd_quote(f'{key}={value}')}" for key, value in sorted(unit.environment.items()) ) diff --git a/tests/deploy/test_native.py b/tests/deploy/test_native.py index db3b1dfb..31dc6729 100644 --- a/tests/deploy/test_native.py +++ b/tests/deploy/test_native.py @@ -487,6 +487,31 @@ class TestUnitFiles: assert 'WorkingDirectory="/srv/my \\"odd\\" dir"' in body assert 'Environment="DOCSGPT_HOME=/srv/my \\"odd\\" dir"' in body + def test_a_newline_in_a_path_is_refused_rather_than_quoted(self, tmp_path): + """A service file is line-based: quoting cannot hold a newline, it would add a directive.""" + unit = native.Unit( + name="docsgpt-api", + arguments=["/venv/bin/docsgpt", "api"], + environment={"DOCSGPT_HOME": str(tmp_path)}, + working_directory="/srv/x\nExecStart=/bin/sh -c evil", + log_file="/srv/api.log", + ) + with pytest.raises(DeployError, match="control character"): + native.systemd_unit(unit) + with pytest.raises(DeployError, match="control character"): + native.launchd_plist(unit) + + def test_a_newline_in_the_environment_is_refused(self, tmp_path): + unit = native.Unit( + name="docsgpt-api", + arguments=["/venv/bin/docsgpt", "api"], + environment={"DOCSGPT_HOME": "/srv/x\nEnvironment=EVIL=1"}, + working_directory=str(tmp_path), + log_file="/srv/api.log", + ) + with pytest.raises(DeployError, match="control character"): + native.systemd_unit(unit) + def test_an_argument_with_spaces_survives_the_systemd_unit(self, tmp_path): unit = self._unit(tmp_path) unit.arguments = ["/opt/my venv/bin/docsgpt", "api"]