fix: 安全加固 (closed #2726)

Author: wyyalt

apps/backend/agent/solution_maker.py

@@ -41,6 +41,13 @@ def shell_quote(value: typing.Any) -> str:
     return shlex.quote(value)
 
 
+def normalize_host_identity_for_shell(value: typing.Any, auth_type: str) -> str:
+    value = "" if value is None else str(value)
+    if auth_type == constants.AuthType.KEY:
+        value = value.replace("\r\n", "\n").replace("\r", "\n").replace("\n", "\\n")
+    return value
+
+
 class ExecutionSolutionStepContent:
     def __init__(
         self,
@@ -795,6 +802,7 @@ def get_run_cmd_base_params(self) -> typing.List[str]:
         host_identity: str = (self.identity_data.password, self.identity_data.key)[
             self.identity_data.auth_type == constants.AuthType.KEY
         ]
+        host_identity = normalize_host_identity_for_shell(host_identity, self.identity_data.auth_type)
         login_ip: str = basic.compressed_ip(self.host.login_ip or self.host.inner_ip or self.host.inner_ipv6)
         run_cmd_params: typing.List[str] = [
             # 文件下载 / 回调服务信息

apps/backend/tests/components/collections/agent_new/test_install.py

@@ -626,6 +626,27 @@ def test_shell_solution(self):
             ],
         )
 
+    def test_shell_solution_with_private_key(self):
+        private_key = "-----BEGIN OPENSSH PRIVATE KEY-----\nkey-content\n-----END OPENSSH PRIVATE KEY-----"
+        host = models.Host.objects.get(bk_host_id=self.obj_factory.bk_host_ids[0])
+        models.IdentityData.objects.filter(bk_host_id=host.bk_host_id).update(
+            auth_type=constants.AuthType.KEY, key=private_key
+        )
+        host.identity = models.IdentityData.objects.get(bk_host_id=host.bk_host_id)
+        escaped_private_key = private_key.replace("\n", "\\n")
+
+        installation_tool = gen_commands(
+            self.LEGACY_SETUP_INFO, host, mock_data_utils.JOB_TASK_PIPELINE_ID, is_uninstall=False, sub_inst_id=0
+        )
+        solution_parse_result: Dict[str, Any] = self.execution_solution_parser(
+            installation_tool=installation_tool,
+            solution_type=constants.CommonExecutionSolutionType.SHELL.value,
+            run_cmd_param_extract={"host_solutions_json_b64": r"(.*) -HSJB (.*)"},
+        )
+
+        self.assertIn(f"--host-identity='{escaped_private_key}'", solution_parse_result["cmds"][0])
+        self.assertIn(f" -HAT {constants.AuthType.KEY}", solution_parse_result["cmds"][0])
+
     def test_target_host_shell_solution(self):
         host = models.Host.objects.get(bk_host_id=self.obj_factory.bk_host_ids[0])
         installation_tool = gen_commands(