237899745/S905L3A
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden driver manager uploads

Browse Source
This commit is contained in:
Codex
2026-06-28 22:12:20 +08:00
parent 88a3d3468c
commit 41eb8b8a6e
4 changed files with 196 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
cups-driver-manager/cups-driver-manager.service
View File

@@ -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

305
cups-driver-manager/driver_manager.py
View File

@@ -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 = '''
支持的格式：
<span>.deb</span>
<span>.ppd</span>
<span>.ppd.gz</span>
<span>.tar</span>
<span>.tgz</span>
<span>.tar.gz</span>
<span>.zip</span>
<span>.rpm</span>
<span>.sh</span>
<span>.run</span>
</div>
<p class="text-muted">最大上传：{{ max_upload_mb }} MB。压缩包内可自动识别 DEB、RPM、PPD、PPD.GZ、安装脚本和 Makefile。</p>
<br>
<button type="submit" class="btn" id="uploadBtn" disabled>开始安装</button>
</form>
@@ -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)