Skip to content

Commit

Permalink
Return to clean exec, ignoring CI fail to preserve code clarity
Browse files Browse the repository at this point in the history
  • Loading branch information
Kidev committed Jan 27, 2025
1 parent 90b53f2 commit 0db2724
Show file tree
Hide file tree
Showing 3 changed files with 14 additions and 84 deletions.
4 changes: 2 additions & 2 deletions aqt/commercial.py
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,7 @@ def gather_packages(self, installer_path: str) -> None:
]

try:
output = safely_run_save_output(installer_path, cmd, Settings.qt_installer_timeout)
output = safely_run_save_output(cmd, Settings.qt_installer_timeout)

# Extract the XML portion from the output
xml_start = output.find("<availablepackages>")
Expand Down Expand Up @@ -323,7 +323,7 @@ def install(self) -> None:

self.logger.info(f"Running: {cmd}")

safely_run(installer_path, cmd, Settings.qt_installer_timeout)
safely_run(cmd, Settings.qt_installer_timeout)
except Exception as e:
self.logger.error(f"Installation failed with exit code {e.__str__()}")
raise
Expand Down
92 changes: 11 additions & 81 deletions aqt/helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,9 @@
import os
import posixpath
import secrets
import shlex

Check warning on line 27 in aqt/helper.py

View check run for this annotation

Codacy Production / Codacy Static Code Analysis

aqt/helper.py#L27

Unused import shlex
import shutil
import subprocess

Check warning on line 29 in aqt/helper.py

View check run for this annotation

Codacy Production / Codacy Static Code Analysis

aqt/helper.py#L29

Consider possible security implications associated with the subprocess module.
import sys
from configparser import ConfigParser
from logging import Handler, getLogger
Expand Down Expand Up @@ -630,88 +632,16 @@ def setup_logging(env_key="LOG_CFG"):
logging.config.fileConfig(Settings.loggingconf)


def safely_run(path: Path, cmd: List[str], timeout: int) -> None:
"""
Executes a subprocess command through a dynamically created Python script.
"""
import shlex
from runpy import run_path

def safely_run(cmd: List[str], timeout: int) -> None:
try:
# Don't modify the original command - create the full path properly
full_cmd = " ".join(cmd)
script_content = f"""import subprocess
subprocess.run({shlex.quote(full_cmd)}, shell=True, timeout={timeout})"""

# Create the script path correctly
script_path = path.parent / "cmd.py"

# Write the script
script_path.write_text(script_content)
subprocess.run(cmd, shell=False, timeout=timeout)

Check warning on line 637 in aqt/helper.py

View check run for this annotation

Codacy Production / Codacy Static Code Analysis

aqt/helper.py#L637

subprocess call - check for execution of untrusted input.
except Exception:
raise

# Execute the script
run_path(str(script_path))

except FileNotFoundError as e:
print(f"Runner unable to be created or read: {e}")
except Exception as e:
print(f"Error during execution: {e}")


def safely_run_save_output(path: Union[str, Path], cmd: List[str], timeout: int) -> Any:
"""
Executes a command through a dynamic script and returns its output.
Similar to subprocess.run with capture_output=True.
"""
import json
import shlex
from pathlib import Path

def safely_run_save_output(cmd: List[str], timeout: int) -> Any:
try:
full_cmd = " ".join(cmd)
# Ensure path is a Path object
path_obj = Path(path) if isinstance(path, str) else path
# Create a temporary file to store the output
output_file = path_obj.parent / "cmd_output.json"

# Create script that captures output and saves to file
script_content = f"""
import subprocess
import json
try:
result = subprocess.run({shlex.quote(full_cmd)}, shell=True, capture_output=True, text=True, timeout={timeout})
output = {{"stdout": result.stdout, "stderr": result.stderr, "returncode": result.returncode}}
except subprocess.CalledProcessError as e:
output = {{"stdout": e.stdout, "stderr": e.stderr, "returncode": e.returncode}}
except Exception as e:
output = {{"stdout": "", "stderr": str(e), "returncode": -1}}
with open("{output_file}", "w") as f:
json.dump(output, f)
"""

script_path = path_obj.parent / "cmd.py"
script_path.write_text(script_content)

# Execute the script
from runpy import run_path

run_path(str(script_path))

# Read the output
with open(output_file) as f:
result = json.load(f)

# Clean up temporary files
script_path.unlink(missing_ok=True)
output_file.unlink(missing_ok=True)

# Handle errors similar to subprocess.run(check=True)
if result["returncode"] != 0:
raise RuntimeError(f"Command failed with exit code {result['returncode']}\nStderr: {result['stderr']}")

return result["stdout"]

except Exception as e:
raise RuntimeError(f"Failed to execute command: {e}")
result = subprocess.run(cmd, shell=False, capture_output=True, text=True, timeout=timeout)

Check warning on line 644 in aqt/helper.py

View check run for this annotation

Codacy Production / Codacy Static Code Analysis

aqt/helper.py#L644

subprocess call - check for execution of untrusted input.
return result
except Exception:
raise
2 changes: 1 addition & 1 deletion aqt/installer.py
Original file line number Diff line number Diff line change
Expand Up @@ -916,7 +916,7 @@ def run_list_qt_commercial(self, args) -> None:
]

# Run search and display output
output = safely_run_save_output(installer_path, cmd, Settings.qt_installer_timeout)
output = safely_run_save_output(cmd, Settings.qt_installer_timeout)
print(output)

except Exception as e:
Expand Down

0 comments on commit 0db2724

Please sign in to comment.