Переглянути джерело

fix: lire backup_info.json via sudo tar (archive root-owned) — restauration custom_dir et db_dump

Cédric Hansen 23 годин тому
батько
коміт
9ccf01b88f
4 змінених файлів з 33 додано та 33 видалено
  1. 2 9
      sources/app.py
  2. 4 10
      sources/jobs/custom_dir.py
  3. 12 14
      sources/jobs/db_dump.py
  4. 15 0
      sources/jobs/utils.py

+ 2 - 9
sources/app.py

@@ -83,15 +83,8 @@ def _inject_globals():
 def _read_archive_info(archive_name):
     backup_dir = app.config["YUNOHOST_BACKUP_DIR"]
     archive_path = os.path.join(backup_dir, archive_name + ".tar")
-    try:
-        import tarfile as _tarfile
-        with _tarfile.open(archive_path) as tar:
-            member = tar.extractfile("backup_info.json")
-            if member:
-                return json.loads(member.read())
-    except Exception:
-        pass
-    return {}
+    from jobs.utils import sudo_read_backup_info
+    return sudo_read_backup_info(archive_path)
 
 
 def _get_ynh_apps():

+ 4 - 10
sources/jobs/custom_dir.py

@@ -4,7 +4,6 @@ import os
 import pwd
 import re
 import subprocess
-import tarfile
 import tempfile
 import time
 from datetime import datetime
@@ -262,17 +261,12 @@ def _get_current_user():
 
 
 def _read_backup_info(archive_path):
-    try:
-        with tarfile.open(archive_path) as tar:
-            member = tar.extractfile("backup_info.json")
-            if member:
-                return json.loads(member.read())
-    except Exception:
-        pass
-    return {}
+    from jobs.utils import sudo_read_backup_info
+    return sudo_read_backup_info(archive_path)
 
 
 def read_backup_info_from_dir(archive_name, backup_dir):
     """Utilisé par app.py pour afficher les infos de restauration."""
+    from jobs.utils import sudo_read_backup_info
     archive_path = os.path.join(backup_dir, archive_name + ".tar")
-    return _read_backup_info(archive_path)
+    return sudo_read_backup_info(archive_path)

+ 12 - 14
sources/jobs/db_dump.py

@@ -1,7 +1,6 @@
 import json
 import os
 import subprocess
-import tarfile
 import tempfile
 import time
 from datetime import datetime
@@ -123,11 +122,16 @@ def restore_db_dump(archive_name, backup_dir):
     with tempfile.TemporaryDirectory() as tmpdir:
         dump_path = os.path.join(tmpdir, f"{dbname}.sql")
 
-        # Extraction du dump depuis l'archive
-        with tarfile.open(archive_path) as tar:
-            member = tar.getmember(f"db/{dbname}.sql")
-            with tar.extractfile(member) as src, open(dump_path, "wb") as dst:
-                dst.write(src.read())
+        # Extraction du dump depuis l'archive via sudo (backup_dir est 750 root)
+        result = subprocess.run(
+            ["sudo", "tar", "-xOf", archive_path, f"db/{dbname}.sql"],
+            capture_output=True,
+        )
+        if result.returncode != 0:
+            err = result.stderr.decode("utf-8", errors="replace").strip()
+            raise RuntimeError(f"Extraction du dump échouée : {err}")
+        with open(dump_path, "wb") as dst:
+            dst.write(result.stdout)
 
         if db_type == "mysql":
             return _restore_mysql(dbname, dump_path)
@@ -208,14 +212,8 @@ def _restore_postgresql(dbname, dump_path):
 
 
 def _read_backup_info(archive_path):
-    try:
-        with tarfile.open(archive_path) as tar:
-            member = tar.extractfile("backup_info.json")
-            if member:
-                return json.loads(member.read())
-    except Exception:
-        pass
-    return {}
+    from jobs.utils import sudo_read_backup_info
+    return sudo_read_backup_info(archive_path)
 
 
 def _write_tar(tmpdir, dump_path, dbname, archive_name, backup_dir, job, instance, instance_url):

+ 15 - 0
sources/jobs/utils.py

@@ -1,3 +1,4 @@
+import json
 import os
 import subprocess
 
@@ -57,6 +58,20 @@ def sudo_exists(path):
     return result.returncode == 0
 
 
+def sudo_read_backup_info(archive_path):
+    """Lit backup_info.json depuis un .tar root-owned via sudo tar -xOf."""
+    result = subprocess.run(
+        ["sudo", "tar", "-xOf", archive_path, "backup_info.json"],
+        capture_output=True,
+    )
+    if result.returncode == 0:
+        try:
+            return json.loads(result.stdout)
+        except Exception:
+            pass
+    return {}
+
+
 def sudo_listdir(directory):
     """Liste les fichiers d'un répertoire via sudo find si non accessible directement."""
     try: