fix(utils): quote kill_process args, replace subprocess which with shutil
- shlex.quote() in kill_process prevents shell injection - find_binary now uses shutil.which() instead of spawning 'which' - Add shell_split() helper for environment.py - Remove inline import re from escape_markup_text
This commit is contained in:
1 file changed
+8
-11
+8
-11
@@ -3,6 +3,8 @@ import html
|
|||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
|
import shlex
|
||||||
|
import shutil
|
||||||
import subprocess
|
import subprocess
|
||||||
import threading
|
import threading
|
||||||
import time
|
import time
|
||||||
@@ -284,8 +286,6 @@ def is_special_workspace(client: dict) -> bool:
|
|||||||
|
|
||||||
|
|
||||||
def escape_markup_text(text: str) -> str:
|
def escape_markup_text(text: str) -> str:
|
||||||
import re
|
|
||||||
|
|
||||||
clean = re.sub(r"<[^>]+>", "", text)
|
clean = re.sub(r"<[^>]+>", "", text)
|
||||||
return html.escape(clean.replace("\n", " "))
|
return html.escape(clean.replace("\n", " "))
|
||||||
|
|
||||||
@@ -304,7 +304,7 @@ def spawn_detached(args: list[str]) -> subprocess.Popen:
|
|||||||
|
|
||||||
|
|
||||||
def kill_process(process_name: str):
|
def kill_process(process_name: str):
|
||||||
exec_shell_command_async(f"pkill {process_name}", lambda *_: None)
|
exec_shell_command_async(f"pkill {shlex.quote(process_name)}", lambda *_: None)
|
||||||
|
|
||||||
|
|
||||||
def is_process_running(process_name: str) -> bool:
|
def is_process_running(process_name: str) -> bool:
|
||||||
@@ -340,14 +340,11 @@ def find_process_pid(process_name: str, timeout: float | None = None) -> list[st
|
|||||||
|
|
||||||
# Binary lookup
|
# Binary lookup
|
||||||
def find_binary(name: str) -> str | None:
|
def find_binary(name: str) -> str | None:
|
||||||
try:
|
return shutil.which(name)
|
||||||
result = subprocess.run(["which", name], capture_output=True, text=True)
|
|
||||||
return result.stdout.strip() or None
|
|
||||||
except Exception as e:
|
def shell_split(s: str) -> list[str]:
|
||||||
logger.warning(
|
return shlex.split(s)
|
||||||
f"[functions] result = subprocess.run(['which', name], capture_output=T... failed: {e}"
|
|
||||||
)
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
# General utilities
|
# General utilities
|
||||||
|
|||||||
Reference in new issue
Block a user