k3nny
debc753f13
Replace all deprecated and unsafe command execution methods with secure subprocess.run() calls using list arguments. Changes: - Replace os.popen() with subprocess.run() in tisbackup_gui.py - Replace os.system() with subprocess.run() in tasks.py and backup_xva.py - Add input validation for device/partition names (regex-based) - Fix file operations to use context managers (with statement) - Remove wildcard import from shutil - Add timeout protection to all subprocess calls (5-30s) - Improve error handling with proper try/except blocks Security improvements: - Prevent command injection vulnerabilities in USB disk operations - Validate device paths with regex before system calls - Use list arguments instead of shell=True to prevent injection - Add proper error handling instead of silent failures Code quality improvements: - Replace deprecated os.popen() (deprecated since Python 2.6) - Use context managers for file operations - Remove wildcard imports for cleaner namespace - Add comprehensive error handling and logging Documentation: - Add SECURITY_IMPROVEMENTS.md documenting all changes - Document remaining security issues and recommendations - Include testing recommendations and migration notes BREAKING CHANGE: None - all changes are backward compatible 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
57 lines
1.5 KiB
Python
57 lines
1.5 KiB
Python
import logging
|
|
import os
|
|
import subprocess
|
|
|
|
from huey import RedisHuey
|
|
|
|
from tisbackup import tis_backup
|
|
|
|
huey = RedisHuey("tisbackup", host="localhost")
|
|
|
|
|
|
@huey.task()
|
|
def run_export_backup(base, config_file, mount_point, backup_sections):
|
|
try:
|
|
# Log
|
|
logger = logging.getLogger("tisbackup")
|
|
logger.setLevel(logging.INFO)
|
|
formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
|
|
handler = logging.StreamHandler()
|
|
handler.setFormatter(formatter)
|
|
logger.addHandler(handler)
|
|
|
|
# Main
|
|
logger.info("Running export....")
|
|
|
|
if backup_sections:
|
|
backup_sections = backup_sections.split(",")
|
|
else:
|
|
backup_sections = []
|
|
backup = tis_backup(dry_run=False, verbose=True, backup_base_dir=base)
|
|
backup.read_ini_file(config_file)
|
|
mount_point = mount_point
|
|
backup.export_backups(backup_sections, mount_point)
|
|
except Exception as e:
|
|
return str(e)
|
|
|
|
finally:
|
|
# Safely unmount using subprocess instead of os.system
|
|
try:
|
|
subprocess.run(["/bin/umount", mount_point], check=True, timeout=30)
|
|
os.rmdir(mount_point)
|
|
except (subprocess.CalledProcessError, subprocess.TimeoutExpired, OSError) as e:
|
|
logger.error(f"Failed to unmount {mount_point}: {e}")
|
|
return "ok"
|
|
|
|
|
|
def get_task():
|
|
return task
|
|
|
|
|
|
def set_task(my_task):
|
|
global task
|
|
task = my_task
|
|
|
|
|
|
task = None
|