mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-04-29 20:13:00 +00:00
fix(backup): address second round of PR review findings
- api_v3: guard opts_dict with isinstance check after json.loads so a
non-object JSON payload (null, array, etc.) returns a 400 instead of a
500 AttributeError
- backup_manager: wrap tmp ZIP creation and os.replace in try/except so
the .zip.tmp temp file is always removed on any failure
- backup_manager: replace hardcoded Path("/tmp/_zip_check") sentinel in
validate_backup with a proper tempfile.TemporaryDirectory() so path
traversal checks are portable and leave no artifacts
- backup_restore.html: detect partial-success responses (plugins_failed or
errors non-empty) even when status is 'success' and render yellow/warning
styling and notify instead of green
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Chuck
@@ -291,48 +291,52 @@ def create_backup(
|
||||
|
||||
# Stream directly to a temp file so we never hold the whole ZIP in memory.
|
||||
tmp_path = zip_path.with_suffix(".zip.tmp")
|
||||
with zipfile.ZipFile(tmp_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
||||
# Config files.
|
||||
if (project_root / _CONFIG_REL).exists():
|
||||
zf.write(project_root / _CONFIG_REL, _CONFIG_REL.as_posix())
|
||||
contents.append("config")
|
||||
if (project_root / _SECRETS_REL).exists():
|
||||
zf.write(project_root / _SECRETS_REL, _SECRETS_REL.as_posix())
|
||||
contents.append("secrets")
|
||||
if (project_root / _WIFI_REL).exists():
|
||||
zf.write(project_root / _WIFI_REL, _WIFI_REL.as_posix())
|
||||
contents.append("wifi")
|
||||
try:
|
||||
with zipfile.ZipFile(tmp_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
||||
# Config files.
|
||||
if (project_root / _CONFIG_REL).exists():
|
||||
zf.write(project_root / _CONFIG_REL, _CONFIG_REL.as_posix())
|
||||
contents.append("config")
|
||||
if (project_root / _SECRETS_REL).exists():
|
||||
zf.write(project_root / _SECRETS_REL, _SECRETS_REL.as_posix())
|
||||
contents.append("secrets")
|
||||
if (project_root / _WIFI_REL).exists():
|
||||
zf.write(project_root / _WIFI_REL, _WIFI_REL.as_posix())
|
||||
contents.append("wifi")
|
||||
|
||||
# User-uploaded fonts.
|
||||
user_fonts = iter_user_fonts(project_root)
|
||||
if user_fonts:
|
||||
for font in user_fonts:
|
||||
arcname = font.relative_to(project_root).as_posix()
|
||||
zf.write(font, arcname)
|
||||
contents.append("fonts")
|
||||
# User-uploaded fonts.
|
||||
user_fonts = iter_user_fonts(project_root)
|
||||
if user_fonts:
|
||||
for font in user_fonts:
|
||||
arcname = font.relative_to(project_root).as_posix()
|
||||
zf.write(font, arcname)
|
||||
contents.append("fonts")
|
||||
|
||||
# Plugin uploads.
|
||||
plugin_uploads = iter_plugin_uploads(project_root)
|
||||
if plugin_uploads:
|
||||
for upload in plugin_uploads:
|
||||
arcname = upload.relative_to(project_root).as_posix()
|
||||
zf.write(upload, arcname)
|
||||
contents.append("plugin_uploads")
|
||||
# Plugin uploads.
|
||||
plugin_uploads = iter_plugin_uploads(project_root)
|
||||
if plugin_uploads:
|
||||
for upload in plugin_uploads:
|
||||
arcname = upload.relative_to(project_root).as_posix()
|
||||
zf.write(upload, arcname)
|
||||
contents.append("plugin_uploads")
|
||||
|
||||
# Installed plugins manifest.
|
||||
plugins = list_installed_plugins(project_root)
|
||||
if plugins:
|
||||
zf.writestr(
|
||||
PLUGINS_MANIFEST_NAME,
|
||||
json.dumps(plugins, indent=2),
|
||||
)
|
||||
contents.append("plugins")
|
||||
# Installed plugins manifest.
|
||||
plugins = list_installed_plugins(project_root)
|
||||
if plugins:
|
||||
zf.writestr(
|
||||
PLUGINS_MANIFEST_NAME,
|
||||
json.dumps(plugins, indent=2),
|
||||
)
|
||||
contents.append("plugins")
|
||||
|
||||
# Manifest goes last so that `contents` reflects what we actually wrote.
|
||||
manifest = _build_manifest(contents, project_root)
|
||||
zf.writestr(MANIFEST_NAME, json.dumps(manifest, indent=2))
|
||||
# Manifest goes last so that `contents` reflects what we actually wrote.
|
||||
manifest = _build_manifest(contents, project_root)
|
||||
zf.writestr(MANIFEST_NAME, json.dumps(manifest, indent=2))
|
||||
|
||||
os.replace(tmp_path, zip_path)
|
||||
os.replace(tmp_path, zip_path)
|
||||
except Exception:
|
||||
tmp_path.unlink(missing_ok=True)
|
||||
raise
|
||||
logger.info("Created backup %s (%d bytes)", zip_path, zip_path.stat().st_size)
|
||||
return zip_path
|
||||
|
||||
@@ -391,15 +395,17 @@ def validate_backup(zip_path: Path) -> Tuple[bool, str, Dict[str, Any]]:
|
||||
return False, "Backup is missing manifest.json", {}
|
||||
|
||||
total = 0
|
||||
for info in zf.infolist():
|
||||
if info.file_size > _MAX_MEMBER_BYTES:
|
||||
return False, f"Member {info.filename} is too large", {}
|
||||
total += info.file_size
|
||||
if total > _MAX_TOTAL_BYTES:
|
||||
return False, "Backup exceeds maximum allowed size", {}
|
||||
# Safety: reject members with unsafe paths up front.
|
||||
if _safe_extract_path(Path("/tmp/_zip_check"), info.filename) is None:
|
||||
return False, f"Unsafe path in backup: {info.filename}", {}
|
||||
with tempfile.TemporaryDirectory() as _sandbox:
|
||||
sandbox = Path(_sandbox)
|
||||
for info in zf.infolist():
|
||||
if info.file_size > _MAX_MEMBER_BYTES:
|
||||
return False, f"Member {info.filename} is too large", {}
|
||||
total += info.file_size
|
||||
if total > _MAX_TOTAL_BYTES:
|
||||
return False, "Backup exceeds maximum allowed size", {}
|
||||
# Safety: reject members with unsafe paths up front.
|
||||
if _safe_extract_path(sandbox, info.filename) is None:
|
||||
return False, f"Unsafe path in backup: {info.filename}", {}
|
||||
|
||||
try:
|
||||
manifest_raw = zf.read(MANIFEST_NAME).decode("utf-8")
|
||||
|
||||
@@ -1296,6 +1296,8 @@ def backup_restore():
|
||||
opts_dict = json.loads(raw_opts)
|
||||
except json.JSONDecodeError:
|
||||
return jsonify({'status': 'error', 'message': 'Invalid options JSON'}), 400
|
||||
if not isinstance(opts_dict, dict):
|
||||
return jsonify({'status': 'error', 'message': 'options must be an object'}), 400
|
||||
|
||||
opts = backup_manager.RestoreOptions(
|
||||
restore_config=_coerce_to_bool(opts_dict.get('restore_config', True)),
|
||||
|
||||
@@ -313,11 +313,14 @@
|
||||
throw new Error(payload.message || msgs || 'Restore had errors');
|
||||
}
|
||||
const data = payload.data || {};
|
||||
const hasPartial = (data.plugins_failed || []).length > 0 || (data.errors || []).length > 0;
|
||||
const result = document.getElementById('restore-result');
|
||||
result.className = 'bg-green-50 border-green-200 text-green-800 border rounded-md p-4';
|
||||
result.className = (hasPartial
|
||||
? 'bg-yellow-50 border-yellow-200 text-yellow-800'
|
||||
: 'bg-green-50 border-green-200 text-green-800') + ' border rounded-md p-4';
|
||||
result.classList.remove('hidden');
|
||||
result.innerHTML = `
|
||||
<h3 class="font-medium mb-2">Restore complete</h3>
|
||||
<h3 class="font-medium mb-2">${hasPartial ? 'Restore complete with warnings' : 'Restore complete'}</h3>
|
||||
<div><strong>Restored:</strong> ${(data.restored || []).map(escapeHtml).join(', ') || 'none'}</div>
|
||||
<div><strong>Skipped:</strong> ${(data.skipped || []).map(escapeHtml).join(', ') || 'none'}</div>
|
||||
<div><strong>Plugins installed:</strong> ${(data.plugins_installed || []).map(escapeHtml).join(', ') || 'none'}</div>
|
||||
@@ -325,7 +328,7 @@
|
||||
<div><strong>Errors:</strong> ${(data.errors || []).map(escapeHtml).join('; ') || 'none'}</div>
|
||||
<p class="mt-2">Restart the display service to apply all changes.</p>
|
||||
`;
|
||||
notify('Restore complete', 'success');
|
||||
notify(hasPartial ? 'Restore complete with warnings' : 'Restore complete', hasPartial ? 'warning' : 'success');
|
||||
} catch (err) {
|
||||
notify('Restore failed: ' + err.message, 'error');
|
||||
} finally {
|
||||
|
||||
Reference in New Issue
Block a user