chore: tune backup function and related docs
This commit is contained in:
@@ -1,53 +1,15 @@
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from config.settings import Settings
|
||||
|
||||
BACKUP_APP_ID = "remnawave-minishop"
|
||||
BACKUP_FILENAME_PREFIX = "remnawave-minishop-backup-"
|
||||
BACKUP_FORMAT_VERSION = 1
|
||||
BACKUP_MANIFEST_NAME = "manifest.json"
|
||||
|
||||
|
||||
def backup_signature_secret(settings: Settings) -> str:
|
||||
configured = str(getattr(settings, "BACKUP_ARCHIVE_SIGNATURE_SECRET", "") or "").strip()
|
||||
return configured or settings.BOT_TOKEN
|
||||
|
||||
|
||||
def canonical_manifest_payload(manifest: dict[str, Any]) -> bytes:
|
||||
payload = json.loads(json.dumps(manifest, ensure_ascii=False))
|
||||
archive = payload.get("archive")
|
||||
if isinstance(archive, dict):
|
||||
archive.pop("signature", None)
|
||||
return json.dumps(
|
||||
payload,
|
||||
ensure_ascii=False,
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
).encode("utf-8")
|
||||
|
||||
|
||||
def sign_manifest(manifest: dict[str, Any], settings: Settings) -> str:
|
||||
return hmac.new(
|
||||
backup_signature_secret(settings).encode("utf-8"),
|
||||
canonical_manifest_payload(manifest),
|
||||
hashlib.sha256,
|
||||
).hexdigest()
|
||||
|
||||
|
||||
def verify_manifest_signature(manifest: dict[str, Any], settings: Settings) -> bool:
|
||||
archive = manifest.get("archive") if isinstance(manifest.get("archive"), dict) else {}
|
||||
signature = str(archive.get("signature") or "")
|
||||
if not signature:
|
||||
return False
|
||||
expected = sign_manifest(manifest, settings)
|
||||
return hmac.compare_digest(signature, expected)
|
||||
|
||||
|
||||
def file_sha256(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as handle:
|
||||
@@ -79,14 +41,12 @@ def attach_archive_integrity(
|
||||
manifest: dict[str, Any],
|
||||
*,
|
||||
file_records: list[dict[str, Any]],
|
||||
settings: Settings,
|
||||
) -> None:
|
||||
manifest["app"] = BACKUP_APP_ID
|
||||
manifest["format_version"] = BACKUP_FORMAT_VERSION
|
||||
manifest["archive"] = {
|
||||
"files": file_records,
|
||||
}
|
||||
manifest["archive"]["signature"] = sign_manifest(manifest, settings)
|
||||
|
||||
|
||||
def write_manifest(source_dir: Path, manifest: dict[str, Any]) -> None:
|
||||
|
||||
@@ -21,7 +21,6 @@ from bot.services.backup_archive import (
|
||||
BACKUP_MANIFEST_NAME,
|
||||
attach_archive_integrity,
|
||||
build_file_records,
|
||||
verify_manifest_signature,
|
||||
write_manifest,
|
||||
write_zip_from_directory,
|
||||
)
|
||||
@@ -153,16 +152,6 @@ class BackupRestoreService:
|
||||
with zipfile.ZipFile(archive_path) as archive:
|
||||
self._validate_zip_members(archive.infolist())
|
||||
manifest = self._read_manifest(archive)
|
||||
signature_valid = self._archive_signature_valid(manifest)
|
||||
signature_required = getattr(
|
||||
self.settings,
|
||||
"BACKUP_ARCHIVE_SIGNATURE_REQUIRED",
|
||||
True,
|
||||
)
|
||||
if signature_required and not signature_valid:
|
||||
raise BackupArchiveError("Archive manifest signature is not valid")
|
||||
if not signature_valid:
|
||||
warnings.append("manifest signature is not valid")
|
||||
has_database = self._find_database_dump_member(archive) is not None
|
||||
compose_members = self._compose_file_members(archive)
|
||||
|
||||
@@ -401,7 +390,6 @@ class BackupRestoreService:
|
||||
attach_archive_integrity(
|
||||
manifest,
|
||||
file_records=build_file_records(staging_dir),
|
||||
settings=self.settings,
|
||||
)
|
||||
write_manifest(staging_dir, manifest)
|
||||
tmp_archive = archive_path.with_name(f"{archive_path.name}.tmp")
|
||||
@@ -551,18 +539,12 @@ class BackupRestoreService:
|
||||
raise BackupArchiveError("Archive manifest format is not supported")
|
||||
return manifest
|
||||
|
||||
def _archive_signature_valid(self, manifest: dict[str, Any]) -> bool:
|
||||
return verify_manifest_signature(manifest, self.settings)
|
||||
|
||||
def _validate_archive_for_restore(self, archive_path: Path) -> None:
|
||||
if not zipfile.is_zipfile(archive_path):
|
||||
raise BackupArchiveError("Archive is not a valid ZIP file")
|
||||
with zipfile.ZipFile(archive_path) as archive:
|
||||
self._validate_zip_members(archive.infolist())
|
||||
manifest = self._read_manifest(archive)
|
||||
if getattr(self.settings, "BACKUP_ARCHIVE_SIGNATURE_REQUIRED", True):
|
||||
if not self._archive_signature_valid(manifest):
|
||||
raise BackupArchiveError("Archive manifest signature is not valid")
|
||||
self._validate_archive_integrity(archive, manifest)
|
||||
|
||||
def _validate_archive_integrity(
|
||||
|
||||
@@ -179,7 +179,6 @@ class BackupWorker:
|
||||
attach_archive_integrity(
|
||||
manifest,
|
||||
file_records=build_file_records(staging_dir),
|
||||
settings=self.settings,
|
||||
)
|
||||
write_manifest(staging_dir, manifest)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user