diff --git a/README.md b/README.md index 7f1e8d2..22e85ea 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,8 @@ chmod +x setup_cups.sh ./setup_cups.sh --driver-manager-auth ``` +驱动管理器默认最大上传 `512MB`,单个安装命令默认超时 `900` 秒。需要调整时可在服务环境变量中设置 `DRIVER_MANAGER_MAX_UPLOAD_MB` 和 `DRIVER_MANAGER_COMMAND_TIMEOUT`。 + ## 驱动选项 安装时可选择以下驱动: diff --git a/cups-driver-manager/cups-driver-manager.service b/cups-driver-manager/cups-driver-manager.service index 3f0894b..376be54 100644 --- a/cups-driver-manager/cups-driver-manager.service +++ b/cups-driver-manager/cups-driver-manager.service @@ -14,6 +14,8 @@ RestartSec=5 Environment=DRIVER_MANAGER_USERNAME=admin Environment=DRIVER_MANAGER_PASSWORD=admin Environment=DRIVER_MANAGER_ALLOW_PRIVATE_ONLY=1 +Environment=DRIVER_MANAGER_MAX_UPLOAD_MB=512 +Environment=DRIVER_MANAGER_COMMAND_TIMEOUT=900 # 安全设置 NoNewPrivileges=false diff --git a/cups-driver-manager/driver_manager.py b/cups-driver-manager/driver_manager.py index 9cad4a4..6d46161 100644 --- a/cups-driver-manager/driver_manager.py +++ b/cups-driver-manager/driver_manager.py @@ -12,7 +12,9 @@ import tempfile import shutil import tarfile import zipfile +import gzip import ipaddress +import uuid from pathlib import Path from functools import wraps @@ -22,9 +24,17 @@ from werkzeug.utils import secure_filename app = Flask(__name__) app.secret_key = os.urandom(24) +def env_int(name, default): + try: + return max(1, int(os.environ.get(name, default))) + except ValueError: + return default + # 配置 UPLOAD_FOLDER = '/tmp/cups-drivers' -MAX_CONTENT_LENGTH = 100 * 1024 * 1024 # 100MB +MAX_UPLOAD_MB = env_int('DRIVER_MANAGER_MAX_UPLOAD_MB', 512) +MAX_CONTENT_LENGTH = MAX_UPLOAD_MB * 1024 * 1024 +COMMAND_TIMEOUT = env_int('DRIVER_MANAGER_COMMAND_TIMEOUT', 900) ALLOWED_EXTENSIONS = {'deb', 'ppd', 'gz', 'tar', 'tgz', 'zip', 'rpm', 'sh', 'run'} # 管理员凭据(可通过环境变量设置) @@ -104,11 +114,11 @@ def allowed_file(filename): """检查文件类型是否允许""" if '.' not in filename: return False + file_type = get_file_type(filename) + if file_type == 'unknown': + return False ext = filename.rsplit('.', 1)[1].lower() - # 处理 .tar.gz 情况 - if filename.endswith('.tar.gz'): - return True - return ext in ALLOWED_EXTENSIONS + return ext in ALLOWED_EXTENSIONS or file_type in ('tar.gz', 'ppd') def get_file_type(filename): """获取文件类型""" @@ -130,13 +140,47 @@ def get_file_type(filename): else: return 'unknown' -def run_command(cmd, shell=False): +def upload_extension(filename): + """保留真实扩展名,避免中文文件名被 secure_filename 清空后丢失类型。""" + filename_lower = filename.lower() + for ext in ('.tar.gz', '.ppd.gz', '.tgz', '.deb', '.ppd', '.tar', '.zip', '.rpm', '.sh', '.run'): + if filename_lower.endswith(ext): + return ext + return Path(filename_lower).suffix + +def safe_upload_filename(filename): + safe_name = secure_filename(filename) + ext = upload_extension(filename) + if not safe_name: + safe_name = f'driver{ext}' + elif ext and safe_name.lower() == ext.lstrip('.'): + safe_name = f'driver{ext}' + elif ext and not safe_name.lower().endswith(ext): + safe_name = f'{Path(safe_name).stem or "driver"}{ext}' + return f'{uuid.uuid4().hex}_{safe_name}' + +def find_files_by_suffix(root_dir, suffixes): + suffixes = tuple(s.lower() for s in suffixes) + matches = [] + for path in Path(root_dir).rglob('*'): + if path.is_file() and path.name.lower().endswith(suffixes): + matches.append(path) + return sorted(matches, key=lambda p: str(p).lower()) + +def run_command(cmd, shell=False, cwd=None, input_text=None): """执行命令并返回结果""" try: if shell: - result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=300) + result = subprocess.run( + cmd, shell=True, capture_output=True, text=True, + timeout=COMMAND_TIMEOUT, cwd=cwd, input=input_text, + encoding='utf-8', errors='replace' + ) else: - result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) + result = subprocess.run( + cmd, capture_output=True, text=True, timeout=COMMAND_TIMEOUT, + cwd=cwd, input=input_text, encoding='utf-8', errors='replace' + ) return { 'success': result.returncode == 0, 'stdout': result.stdout, @@ -144,7 +188,7 @@ def run_command(cmd, shell=False): 'returncode': result.returncode } except subprocess.TimeoutExpired: - return {'success': False, 'stdout': '', 'stderr': '命令执行超时', 'returncode': -1} + return {'success': False, 'stdout': '', 'stderr': f'命令执行超时(>{COMMAND_TIMEOUT}秒)', 'returncode': -1} except Exception as e: return {'success': False, 'stdout': '', 'stderr': str(e), 'returncode': -1} @@ -152,14 +196,23 @@ def install_deb(filepath): """安装 .deb 包""" results = [] - # 先尝试直接安装 result = run_command(['dpkg', '-i', filepath]) - results.append(('dpkg -i', result)) + if result['success']: + results.append(('安装 DEB 包', result)) + return results - # 修复依赖 - if not result['success']: - fix_result = run_command(['apt-get', 'install', '-f', '-y']) - results.append(('apt-get install -f', fix_result)) + # dpkg 因依赖缺失失败是常见情况,继续修复依赖后重试,不把初次失败计为最终失败。 + results.append(('初次安装 DEB 包', { + 'success': True, + 'stdout': result['stdout'], + 'stderr': '初次 dpkg 安装未完成,正在尝试自动修复依赖后重试。\n' + result['stderr'], + 'returncode': 0 + })) + fix_result = run_command(['apt-get', 'install', '-f', '-y']) + results.append(('修复 DEB 依赖', fix_result)) + if fix_result['success']: + retry_result = run_command(['dpkg', '-i', filepath]) + results.append(('重新安装 DEB 包', retry_result)) return results @@ -177,39 +230,93 @@ def install_ppd(filepath): for ppd_dir in ppd_dirs: os.makedirs(ppd_dir, exist_ok=True) - # 复制PPD文件 filename = os.path.basename(filepath) - dest = os.path.join(ppd_dirs[0], filename) + is_gzipped = filename.lower().endswith('.ppd.gz') + dest_filename = filename[:-3] if is_gzipped else filename - try: - shutil.copy2(filepath, dest) - os.chmod(dest, 0o644) - results.append(('复制PPD文件', { - 'success': True, - 'stdout': f'已复制到 {dest}', - 'stderr': '', - 'returncode': 0 - })) - except Exception as e: - results.append(('复制PPD文件', { - 'success': False, - 'stdout': '', - 'stderr': str(e), - 'returncode': 1 - })) + for ppd_dir in ppd_dirs: + dest = os.path.join(ppd_dir, dest_filename) + try: + if is_gzipped: + with gzip.open(filepath, 'rb') as src, open(dest, 'wb') as dst: + shutil.copyfileobj(src, dst) + else: + shutil.copy2(filepath, dest) + os.chmod(dest, 0o644) + results.append((f'安装 PPD 到 {ppd_dir}', { + 'success': True, + 'stdout': f'已安装到 {dest}', + 'stderr': '', + 'returncode': 0 + })) + except Exception as e: + results.append((f'安装 PPD 到 {ppd_dir}', { + 'success': False, + 'stdout': '', + 'stderr': str(e), + 'returncode': 1 + })) return results -def install_tar_gz(filepath): - """安装 .tar.gz 包""" +def install_extracted_dir(extract_dir): + """从已解压目录中自动寻找可安装的驱动文件。""" results = [] + root_path = Path(extract_dir) - # 创建临时解压目录 + deb_files = find_files_by_suffix(extract_dir, ('.deb',)) + if deb_files: + for deb_file in deb_files: + results.extend(install_deb(str(deb_file))) + return results + + rpm_files = find_files_by_suffix(extract_dir, ('.rpm',)) + if rpm_files: + for rpm_file in rpm_files: + results.extend(install_rpm(str(rpm_file))) + return results + + install_script_names = {'install.sh', 'setup.sh', 'install', 'setup'} + for path in root_path.rglob('*'): + if path.is_file() and path.name.lower() in install_script_names: + os.chmod(path, 0o755) + if path.name.lower().endswith('.sh'): + result = run_command(['/bin/sh', str(path)], cwd=str(path.parent)) + else: + result = run_command([str(path)], cwd=str(path.parent)) + results.append(('执行安装脚本', result)) + return results + + for path in root_path.rglob('*'): + if path.is_file() and path.name.lower() == 'makefile': + make_result = run_command(['make', '-C', str(path.parent)]) + results.append(('make', make_result)) + if make_result['success']: + install_result = run_command(['make', '-C', str(path.parent), 'install']) + results.append(('make install', install_result)) + return results + + ppd_files = find_files_by_suffix(extract_dir, ('.ppd', '.ppd.gz')) + if ppd_files: + for ppd_file in ppd_files: + results.extend(install_ppd(str(ppd_file))) + return results + + results.append(('查找安装方式', { + 'success': False, + 'stdout': '', + 'stderr': '未找到 DEB、RPM、安装脚本、Makefile 或 PPD 文件', + 'returncode': 1 + })) + return results + +def install_tar_gz(filepath): + """安装 .tar/.tar.gz/.tgz 包""" + results = [] extract_dir = tempfile.mkdtemp(prefix='driver_') try: - # 解压 - with tarfile.open(filepath, 'r:gz') as tar: + with tarfile.open(filepath, 'r:*') as tar: safe_extract_tar(tar, extract_dir) results.append(('解压文件', { 'success': True, @@ -217,48 +324,7 @@ def install_tar_gz(filepath): 'stderr': '', 'returncode': 0 })) - - # 查找安装脚本 - install_scripts = ['install.sh', 'setup.sh', 'install', 'setup'] - found_script = None - - for root, dirs, files in os.walk(extract_dir): - for script in install_scripts: - if script in files: - found_script = os.path.join(root, script) - break - if found_script: - break - - if found_script: - os.chmod(found_script, 0o755) - result = run_command(found_script, shell=True) - results.append(('执行安装脚本', result)) - else: - # 查找 Makefile - for root, dirs, files in os.walk(extract_dir): - if 'Makefile' in files: - make_result = run_command(['make', '-C', root]) - results.append(('make', make_result)) - if make_result['success']: - install_result = run_command(['make', '-C', root, 'install']) - results.append(('make install', install_result)) - break - else: - # 查找 PPD 文件 - ppd_files = list(Path(extract_dir).rglob('*.ppd')) - if ppd_files: - for ppd_file in ppd_files: - ppd_results = install_ppd(str(ppd_file)) - results.extend(ppd_results) - else: - results.append(('查找安装方式', { - 'success': False, - 'stdout': '', - 'stderr': '未找到安装脚本、Makefile或PPD文件', - 'returncode': 1 - })) - + results.extend(install_extracted_dir(extract_dir)) except Exception as e: results.append(('解压文件', { 'success': False, @@ -289,45 +355,7 @@ def install_zip(filepath): 'stderr': '', 'returncode': 0 })) - - # 查找 deb 文件 - deb_files = list(Path(extract_dir).rglob('*.deb')) - if deb_files: - for deb_file in deb_files: - deb_results = install_deb(str(deb_file)) - results.extend(deb_results) - return results - - # 查找安装脚本 - install_scripts = ['install.sh', 'setup.sh', 'install', 'setup'] - found_script = None - - for root, dirs, files in os.walk(extract_dir): - for script in install_scripts: - if script in files: - found_script = os.path.join(root, script) - break - if found_script: - break - - if found_script: - os.chmod(found_script, 0o755) - result = run_command(found_script, shell=True) - results.append(('执行安装脚本', result)) - else: - # 查找 PPD 文件 - ppd_files = list(Path(extract_dir).rglob('*.ppd')) - if ppd_files: - for ppd_file in ppd_files: - ppd_results = install_ppd(str(ppd_file)) - results.extend(ppd_results) - else: - results.append(('查找安装方式', { - 'success': False, - 'stdout': '', - 'stderr': '未找到deb包、安装脚本或PPD文件', - 'returncode': 1 - })) + results.extend(install_extracted_dir(extract_dir)) except Exception as e: results.append(('解压文件', { @@ -357,7 +385,7 @@ def install_rpm(filepath): # 使用 alien 转换 work_dir = os.path.dirname(filepath) - convert_result = run_command(f'cd {work_dir} && alien -d {filepath}', shell=True) + convert_result = run_command(['alien', '-d', filepath], cwd=work_dir) results.append(('转换RPM为DEB', convert_result)) if convert_result['success']: @@ -381,11 +409,14 @@ def install_script(filepath): if 'hplip' in filename and 'plugin' in filename: # 直接执行 .run 文件,使用 yes 自动确认交互式提示 # 注意:hp-plugin 命令需要 .asc 签名文件,比较麻烦,所以直接执行 - result = run_command(f'yes | sh {filepath}', shell=True) + result = run_command(['/bin/sh', filepath], input_text='y\n' * 20) results.append(('执行 HP 插件安装脚本', result)) else: - # 普通脚本直接执行 - result = run_command(filepath, shell=True) + # 普通脚本直接执行,避免文件名包含空格时 shell 拼接失败。 + if filename.endswith('.sh'): + result = run_command(['/bin/sh', filepath]) + else: + result = run_command([filepath]) results.append(('执行安装脚本', result)) return results @@ -545,6 +576,11 @@ HTML_TEMPLATE = ''' margin: 2px; display: inline-block; } + .text-muted { + color: #666; + font-size: 13px; + margin-top: 8px; + } .alert { padding: 15px; border-radius: 5px; @@ -686,11 +722,16 @@ HTML_TEMPLATE = ''' 支持的格式: .deb .ppd + .ppd.gz + .tar + .tgz .tar.gz .zip .rpm .sh + .run +

最大上传:{{ max_upload_mb }} MB。压缩包内可自动识别 DEB、RPM、PPD、PPD.GZ、安装脚本和 Makefile。


@@ -937,7 +978,17 @@ DRIVERS_TEMPLATE = ''' @app.route('/') @requires_auth def index(): - return render_template_string(HTML_TEMPLATE, results=None) + return render_template_string(HTML_TEMPLATE, results=None, max_upload_mb=MAX_UPLOAD_MB) + +@app.errorhandler(413) +def upload_too_large(error): + if request.path.startswith('/api/'): + return jsonify({ + 'success': False, + 'error': f'文件过大,当前最大允许上传 {MAX_UPLOAD_MB} MB' + }), 413 + flash(f'文件过大,当前最大允许上传 {MAX_UPLOAD_MB} MB', 'danger') + return redirect(url_for('index')) @app.route('/upload', methods=['POST']) @requires_auth @@ -957,7 +1008,7 @@ def upload_file(): return redirect(url_for('index')) # 保存文件 - filename = secure_filename(file.filename) + filename = safe_upload_filename(file.filename) filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename) file.save(filepath) @@ -980,7 +1031,7 @@ def upload_file(): else: flash('驱动安装过程中出现错误,请查看详细信息', 'danger') - return render_template_string(HTML_TEMPLATE, results=results) + return render_template_string(HTML_TEMPLATE, results=results, max_upload_mb=MAX_UPLOAD_MB) @app.route('/drivers') @requires_auth @@ -1030,7 +1081,7 @@ def api_install(): return jsonify({'success': False, 'error': '不支持的文件类型'}) # 保存文件 - filename = secure_filename(file.filename) + filename = safe_upload_filename(file.filename) filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename) file.save(filepath) diff --git a/setup_cups.sh b/setup_cups.sh index 2b2dbbe..00951c7 100755 --- a/setup_cups.sh +++ b/setup_cups.sh @@ -24,6 +24,8 @@ CUPS_ALLOWED_NET="${CUPS_ALLOWED_NET:-@LOCAL}" DRIVER_MANAGER_HOST="${DRIVER_MANAGER_HOST:-0.0.0.0}" DRIVER_MANAGER_USERNAME="${DRIVER_MANAGER_USERNAME:-admin}" DRIVER_MANAGER_PRIVATE_ONLY="${DRIVER_MANAGER_PRIVATE_ONLY:-1}" +DRIVER_MANAGER_MAX_UPLOAD_MB="${DRIVER_MANAGER_MAX_UPLOAD_MB:-512}" +DRIVER_MANAGER_COMMAND_TIMEOUT="${DRIVER_MANAGER_COMMAND_TIMEOUT:-900}" # 打印带颜色的信息 info() { echo -e "${BLUE}[信息]${NC} $1"; } @@ -1127,6 +1129,16 @@ install_driver_manager() { else sed -i "/DRIVER_MANAGER_PASSWORD=/a Environment=DRIVER_MANAGER_ALLOW_PRIVATE_ONLY=$DRIVER_MANAGER_PRIVATE_ONLY" /etc/systemd/system/cups-driver-manager.service fi + if grep -q "DRIVER_MANAGER_MAX_UPLOAD_MB" /etc/systemd/system/cups-driver-manager.service; then + sed -i "s/DRIVER_MANAGER_MAX_UPLOAD_MB=.*/DRIVER_MANAGER_MAX_UPLOAD_MB=$DRIVER_MANAGER_MAX_UPLOAD_MB/" /etc/systemd/system/cups-driver-manager.service + else + sed -i "/DRIVER_MANAGER_ALLOW_PRIVATE_ONLY=/a Environment=DRIVER_MANAGER_MAX_UPLOAD_MB=$DRIVER_MANAGER_MAX_UPLOAD_MB" /etc/systemd/system/cups-driver-manager.service + fi + if grep -q "DRIVER_MANAGER_COMMAND_TIMEOUT" /etc/systemd/system/cups-driver-manager.service; then + sed -i "s/DRIVER_MANAGER_COMMAND_TIMEOUT=.*/DRIVER_MANAGER_COMMAND_TIMEOUT=$DRIVER_MANAGER_COMMAND_TIMEOUT/" /etc/systemd/system/cups-driver-manager.service + else + sed -i "/DRIVER_MANAGER_MAX_UPLOAD_MB=/a Environment=DRIVER_MANAGER_COMMAND_TIMEOUT=$DRIVER_MANAGER_COMMAND_TIMEOUT" /etc/systemd/system/cups-driver-manager.service + fi else # 手动创建服务文件 cat > /etc/systemd/system/cups-driver-manager.service << EOF @@ -1145,6 +1157,8 @@ RestartSec=5 Environment="DRIVER_MANAGER_USERNAME=$service_user" Environment="DRIVER_MANAGER_PASSWORD=$service_pass" Environment=DRIVER_MANAGER_ALLOW_PRIVATE_ONLY=$DRIVER_MANAGER_PRIVATE_ONLY +Environment=DRIVER_MANAGER_MAX_UPLOAD_MB=$DRIVER_MANAGER_MAX_UPLOAD_MB +Environment=DRIVER_MANAGER_COMMAND_TIMEOUT=$DRIVER_MANAGER_COMMAND_TIMEOUT [Install] WantedBy=multi-user.target