You've already forked openshot-qt
mirror of
https://github.com/OpenShot/openshot-qt.git
synced 2026-06-08 22:18:12 -07:00
Fixed the 11 Codacy nitpick
- Added explicit subprocess validation and fixed-tool paths in installer/fix_opencv_rpath.py. - Added file path validation for the objdump subprocess call in freeze.py. - Restricted urlopen usage to http/https in comfy_client.py and http_client.py. - Removed unused variables and the unnecessary elif in the flagged UI/model/metrics/cache files. - Also cleaned two nearby identical unused is_all_objects_update assignments in properties_model.py.
This commit is contained in:
@@ -179,10 +179,18 @@ def find_files(directory, patterns):
|
||||
|
||||
def find_windows_imports(binary_path):
|
||||
"""Return DLL imports reported by objdump for a Windows binary."""
|
||||
binary_path = os.path.abspath(binary_path)
|
||||
if not os.path.isfile(binary_path):
|
||||
log.warning("Unable to inspect Windows DLL imports for missing file: %s", binary_path)
|
||||
return None
|
||||
if "\x00" in binary_path:
|
||||
log.warning("Unable to inspect Windows DLL imports for invalid path")
|
||||
return None
|
||||
|
||||
log.info("Inspecting Windows DLL imports: %s", binary_path)
|
||||
try:
|
||||
output = subprocess.check_output(
|
||||
["objdump", "-p", binary_path],
|
||||
output = subprocess.check_output( # nosec B603,B607 - fixed tool, validated file path, no shell.
|
||||
["objdump", "-p", "--", binary_path],
|
||||
stderr=subprocess.STDOUT,
|
||||
universal_newlines=True,
|
||||
)
|
||||
|
||||
@@ -220,7 +220,7 @@ def collect_transitions(icon_size):
|
||||
("extra", extra_dir, os.listdir(extra_dir)),
|
||||
]
|
||||
|
||||
for group_type, dir_name, files in transition_groups:
|
||||
for _, dir_name, files in transition_groups:
|
||||
for filename in sorted(files):
|
||||
if filename[0] == "." or "thumbs.db" in filename.lower():
|
||||
continue
|
||||
|
||||
@@ -10,17 +10,40 @@ import argparse
|
||||
import glob
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import subprocess # nosec B404 - macOS packaging uses fixed local tooling.
|
||||
import sys
|
||||
|
||||
|
||||
ALLOWED_COMMANDS = {
|
||||
"install_name_tool": "/usr/bin/install_name_tool",
|
||||
"otool": "/usr/bin/otool",
|
||||
}
|
||||
|
||||
|
||||
def validate_tool_command(command):
|
||||
"""Validate subprocess commands used by this packaging helper."""
|
||||
if not command or command[0] not in ALLOWED_COMMANDS:
|
||||
raise ValueError("Unexpected command: {}".format(command[0] if command else ""))
|
||||
if any("\x00" in str(argument) for argument in command):
|
||||
raise ValueError("Command arguments must not contain null bytes")
|
||||
return [ALLOWED_COMMANDS[command[0]]] + command[1:]
|
||||
|
||||
|
||||
def run(command):
|
||||
validated_command = validate_tool_command(command)
|
||||
print(" ".join(command))
|
||||
subprocess.check_call(command)
|
||||
subprocess.check_call(validated_command) # nosec B603 - fixed executable, validated args, no shell.
|
||||
|
||||
|
||||
def otool_dependencies(path):
|
||||
output = subprocess.check_output(["otool", "-L", path], text=True)
|
||||
path = os.path.abspath(path)
|
||||
if not os.path.exists(path):
|
||||
raise FileNotFoundError(f"Binary not found: {path}")
|
||||
command = validate_tool_command(["otool", "-L", path])
|
||||
output = subprocess.check_output( # nosec B603 - fixed executable, validated path, no shell.
|
||||
command,
|
||||
text=True,
|
||||
)
|
||||
dependencies = []
|
||||
for line in output.splitlines()[1:]:
|
||||
line = line.strip()
|
||||
|
||||
@@ -42,12 +42,20 @@ from classes import http_client, info
|
||||
from classes.logger import log
|
||||
|
||||
|
||||
def _validate_urlopen_scheme(url):
|
||||
request_url = getattr(url, "full_url", url)
|
||||
scheme = urlparse(str(request_url)).scheme.lower()
|
||||
if scheme not in ("http", "https"):
|
||||
raise ValueError("Unsupported URL scheme for ComfyUI request: {}".format(scheme or "<empty>"))
|
||||
return request_url, scheme
|
||||
|
||||
|
||||
def urlopen(url, *args, **kwargs):
|
||||
"""Open Comfy URLs with packaged CA handling for HTTPS."""
|
||||
request_url = getattr(url, "full_url", url)
|
||||
if urlparse(str(request_url)).scheme.lower() == "https" and "context" not in kwargs:
|
||||
request_url, scheme = _validate_urlopen_scheme(url)
|
||||
if scheme == "https" and "context" not in kwargs:
|
||||
kwargs["context"] = http_client.ssl_context()
|
||||
return _stdlib_urlopen(url, *args, **kwargs)
|
||||
return _stdlib_urlopen(url, *args, **kwargs) # nosec B310 - URL schemes are restricted above.
|
||||
|
||||
|
||||
class ComfyProgressSocket:
|
||||
|
||||
@@ -31,6 +31,7 @@ from classes.logger import log
|
||||
DEFAULT_CONNECT_TIMEOUT = 2
|
||||
DEFAULT_READ_TIMEOUT = 5
|
||||
DOWNLOAD_READ_SIZE = 1024 * 1024
|
||||
URLLIB_ALLOWED_SCHEMES = ("http", "https")
|
||||
|
||||
|
||||
def ca_bundle_path():
|
||||
@@ -80,6 +81,14 @@ def http_fallback_url(url):
|
||||
return urlunparse(parsed._replace(scheme="http"))
|
||||
|
||||
|
||||
def validate_url_scheme(url, allowed_schemes=URLLIB_ALLOWED_SCHEMES):
|
||||
"""Validate URLs before passing them to urllib helpers."""
|
||||
scheme = urlparse(str(url)).scheme.lower()
|
||||
if scheme not in allowed_schemes:
|
||||
raise ValueError("Unsupported URL scheme: {}".format(scheme or "<empty>"))
|
||||
return scheme
|
||||
|
||||
|
||||
def urls_with_http_fallback(url):
|
||||
"""Return URL attempts in preferred order."""
|
||||
urls = [url]
|
||||
@@ -163,15 +172,16 @@ def download_file(urls, output_path, description, progress_callback=None, timeou
|
||||
|
||||
|
||||
def _download_file_once(url, output_path, progress_callback, timeout):
|
||||
scheme = validate_url_scheme(url)
|
||||
request = urllib.request.Request(
|
||||
url,
|
||||
headers={"User-Agent": "OpenShot/{}".format(info.VERSION)},
|
||||
)
|
||||
kwargs = {"timeout": sum(timeout) if isinstance(timeout, tuple) else timeout}
|
||||
if urlparse(url).scheme.lower() == "https":
|
||||
if scheme == "https":
|
||||
kwargs["context"] = ssl_context()
|
||||
|
||||
with urllib.request.urlopen(request, **kwargs) as response:
|
||||
with urllib.request.urlopen(request, **kwargs) as response: # nosec B310 - URL scheme is restricted above.
|
||||
total_size = response.headers.get("Content-Length")
|
||||
total_size = int(total_size) if total_size else 0
|
||||
downloaded_size = 0
|
||||
|
||||
@@ -215,7 +215,7 @@ def _send_metric_worker():
|
||||
}
|
||||
|
||||
try:
|
||||
r = http_client.post_json(
|
||||
http_client.post_json(
|
||||
url,
|
||||
payload,
|
||||
"metrics",
|
||||
|
||||
@@ -541,7 +541,7 @@ class PropertiesModel(updates.UpdateInterface):
|
||||
# Create reference
|
||||
clip_data = c.data
|
||||
if object_id:
|
||||
clip_data, is_all_objects_update = self._tracked_object_clip_data(c.data, object_id)
|
||||
clip_data, _ = self._tracked_object_clip_data(c.data, object_id)
|
||||
if not clip_data:
|
||||
log.debug("No clip data found for this object id")
|
||||
return
|
||||
@@ -699,7 +699,7 @@ class PropertiesModel(updates.UpdateInterface):
|
||||
# Create reference
|
||||
clip_data = c.data
|
||||
if object_id:
|
||||
clip_data, is_all_objects_update = self._tracked_object_clip_data(c.data, object_id)
|
||||
clip_data, _ = self._tracked_object_clip_data(c.data, object_id)
|
||||
if not isinstance(clip_data, dict):
|
||||
clip_data = {}
|
||||
|
||||
@@ -979,7 +979,7 @@ class PropertiesModel(updates.UpdateInterface):
|
||||
# Create reference
|
||||
clip_data = c.data
|
||||
if object_id:
|
||||
clip_data, is_all_objects_update = self._tracked_object_clip_data(c.data, object_id)
|
||||
clip_data, _ = self._tracked_object_clip_data(c.data, object_id)
|
||||
if not clip_data:
|
||||
log.debug("No clip data found for this object id")
|
||||
return
|
||||
|
||||
@@ -508,7 +508,7 @@ class ProcessEffect(QDialog):
|
||||
if os.path.getsize(path) <= 0:
|
||||
return False, _("Model file is empty.")
|
||||
return self.validate_onnx_model_load(path)
|
||||
elif validator == "classes":
|
||||
if validator == "classes":
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as classes_file:
|
||||
class_names = [line.strip() for line in classes_file if line.strip()]
|
||||
@@ -524,7 +524,7 @@ class ProcessEffect(QDialog):
|
||||
self.file_validation_timer.stop()
|
||||
|
||||
all_valid = True
|
||||
for setting, field in self.file_fields.items():
|
||||
for field in self.file_fields.values():
|
||||
path = field["path"].text()
|
||||
valid, message = self.validate_file_param(field["param"], path)
|
||||
status = field["status"]
|
||||
|
||||
Reference in New Issue
Block a user